ReconfigurationSpec: source-propagation case hangs after UDF processes EndChannel

[Yicong-Huang]

What happened?

ReconfigurationSpec's "Engine should propagate reconfiguration through a source operator in workflow" case (added in #4220) consistently times out at Await.result(completion, Duration.fromMinutes(1)).

The workflow under test is pythonSourceOpDesc(10000) -> pythonOpDesc(), reconfiguring only the downstream UDF. From the worker logs, the reconfiguration ECM does propagate correctly:

1. Source receives the test-reconfigure-1 ECM (cmd=None) and forwards it to the UDF.
2. UDF receives and processes the embedded UpdateExecutor command.
3. After resume, source produces all tuples and sends EndChannel to UDF.
4. UDF logs process channel ECM ... EndChannel and then goes silent for the full 1-minute timeout — no _process_end_channel-driven port_completed / complete() activity, no Region 0 successfully terminated from the controller.

The other four cases in the spec pass (single-op python UDF reconfigure, java operator reconfigure, source-as-target rejection, two-UDF chain). The failure is specific to the multi-worker source-propagation path with a Python source.

#4531 (which restored the web-service entrypoint and added per-worker completion notification) does not fix this — the test still times out on main after that PR.

To unblock CI I'm temporarily marking this case ignore in a follow-up PR; this issue tracks the real fix.

How to reproduce?

sbt "WorkflowExecutionService/testOnly *ReconfigurationSpec"

Expected: 5/5 tests pass.
Observed: 4/5 pass; should propagate reconfiguration through a source operator in workflow fails with:

com.twitter.util.TimeoutException: 1.minutes
  at com.twitter.util.Promise.ready(Promise.scala:680)
  ...
  at org.apache.texera.amber.engine.e2e.ReconfigurationSpec.shouldReconfigure(ReconfigurationSpec.scala:180)

Version

1.1.0-incubating (Pre-release/Master)

Commit Hash (Optional)

98ba87e

[Yicong-Huang]

Root cause: #4424 (ef66190f22 fix: add missing context switches for repeated state processing)

CI history on main for the scala (ubuntu-22.04, 11) job:

commit	PR	result
e635dd02	#4220 (test added)	✅
76480d98 … 23901be2	(10 commits, incl. #4506, #4504, #4508, #4510, #4518, #4521, #4526, #4523, #4512, #4533, #4520, #4531)	✅ all green
ef66190f	#4424	❌ first failure — should propagate reconfiguration through a source operator in workflow *** FAILED *** (1.minutes TimeoutException)
05b271c5 …	#4441 onward	flaky (mix of pass/fail)

#4424 changed TupleProcessingManager.get_internal_marker() from an atomic get-and-clear to a read-only get, and moved the clear to after the marker handler in MainLoop._process_ecm. It also added _switch_context() calls inside MainLoop._process_state and DataProcessor.process_state.

Why it breaks this specific case:

• The source-propagation test is the only ReconfigurationSpec case that runs Python source → Python UDF with the UpdateExecutor reconfigure ECM and the trailing EndChannel ECM both flowing through the UDF's data-channel ECM path. The other four cases either use a Scala/CSV source or take the single-op fries branch (direct control RPC).
• When the UDF's MainLoop processes the EndChannel ECM, the handler sets current_internal_marker = EndChannel(). _process_ecm then dispatches to _process_end_channel, which calls process_input_state / process_input_tuple. Both _switch_context() to the DataProcessor thread.
• Pre-fix: add missing context switches for repeated state processing #4424 DataProcessor.run() read the marker via the get-and-clear get_internal_marker(), processed it once, then on the next iteration saw None and proceeded normally. Post-fix: add missing context switches for repeated state processing #4424 the marker stays set until _process_ecm clears it after _process_end_channel returns — but _process_end_channel cannot return until DataProcessor yields back, and DataProcessor keeps re-reading the same EndChannel marker and re-invoking process_internal_marker (produce_state_on_finish, on_finish).
• Net effect: the UDF never reaches complete(), no port_completed reaches the controller, region never terminates, Await.result(completion, Duration.fromMinutes(1)) in ReconfigurationSpec.shouldReconfigure times out.

Reverting #4424 outright is not a fix — it re-breaks the pause/state-processing path that #4424 was added to repair (locally, reverting it makes 3/5 cases time out at the 10s pausedReached await). The correct fix should keep #4424's context-switch additions but ensure the marker is consumed exactly once on the DataProcessor side, e.g. clear current_internal_marker immediately after the read in DataProcessor.run() (restoring atomic consume), or before the first _switch_context inside _process_end_channel / _process_start_channel.

cc @aglinxinyuan

[Yicong-Huang]

One clarification to my root-cause note above:

The regression is not that the reconfiguration ECM gets blocked by pause. In this path the ECM is carried in-band on the data channel, but ECM_PAUSE only pauses normal tuple intake until the aligned ECM is handled; the worker should still process the ECM itself while paused.

What #4424 broke is the marker lifetime after that ECM is handled. current_internal_marker is no longer consumed atomically on the DataProcessor side, so the source's internal EndChannel marker remains visible across the _switch_context() round-trips inside _process_end_channel. After the reconfiguration ECM handler runs, that stale marker can be observed / re-entered again, which corrupts the end-of-stream handshake.

That is why this reproduces on the Python-source propagation path rather than on the Scala/CSV-source cases: the ECM delivery during pause is behaving as designed here; the broken part is the marker consume/clear semantics that #4424 changed.

[aglinxinyuan]

Thanks for the great analysis. You’re probably the only person who fully understands this lifecycle.
One question: why was the test case flaky instead of consistently failing? Also, how would you design a test to properly guard this lifecycle?

[Yicong-Huang]

The flaky is edge case that depends on time and order, it is a race condition. It is unrealistic to guard through all of those potential race condition cases. But our design, especially python UDF design, should act as a single thread, so that if we followed the design, no race condition should happen.

I am glad this specific case was caught post merge. Given the limited CI resource we have, not always we can catch bugs pre-merge. But with more CI/CD runs we could identify more cases like this post merge. That's also good enough.

I will add more assertions and test cases to make per-merge CI more robust. and count on post-merge repeats to detect and alert us flaky case like this.

[chenlica]

Given the depth and subtlety of this issue, it's very important to have enough comments to keep track of it. I am glad @Yicong-Huang is already doing that.