aglinxinyuan opened a new pull request, #7624:
URL: https://github.com/apache/texera/pull/7624
### What changes were proposed in this PR?
**Root cause.** `MainLoop._process_end_channel` ends with `self.complete()`,
which transits the worker to `COMPLETED`. Nothing between that call and the
last control check above it drains control, but the coordinator pauses on its
own schedule — so a `PauseWorker` command can land in that window and leave the
worker `PAUSED`. `PAUSED -> COMPLETED` is forbidden by the transition graph, so
`transit_to` raised `InvalidTransitionException`, which killed the main loop
thread:
```
main_loop.py:674 _process_end_channel -> self.complete()
main_loop.py:305 complete() -> state_manager.transit_to(COMPLETED)
state_manager.py:80 -> InvalidTransitionException:
Cannot transit from PAUSED to
COMPLETED
```
```
Before: pause lands before complete() -> PAUSED -> COMPLETED ->
InvalidTransition, thread dies
After: pause lands before complete() -> wait for Resume -> RUNNING ->
COMPLETED
```
The worker now waits the pause out before completing, which is what the
Scala runtime already does: `DPThread`'s input selection only picks *control*
channels while `pauseManager.isPaused`, so a paused Scala worker never advances
to completion.
The wait is guarded on `pause_manager.is_paused()` rather than draining
control unconditionally. `_check_and_process_control` blocks while the data
lane is disabled, and **backpressure disables that lane too**
(`DisableType.DISABLE_BY_BACKPRESSURE`) — an unguarded drain here would park a
worker that is merely backpressured. Guarded, the happy path is untouched: not
paused, loop body never runs.
Deliberately *not* changed: the transition graph. Adding `PAUSED ->
COMPLETED` would be the smaller diff, but `WORKER_STATE_TRANSITIONS` mirrors
Scala's `WorkerStateManager` (`PAUSED -> Set(RUNNING)`) by contract, and the
two must agree.
> Note for #5913: its stated cause is now stale. It attributes the flake to
a 100-row `smallCsvScanOpDesc` finishing before the pause lands and prescribes
a larger source; #5915 already did that, and the spec has used
`slowRegionSourceOpDesc(numTuple = 30, delaySeconds = 0.25)` since. The
surviving race is this one, in pyamber, and is independent of how long the
source runs.
### Any related issues, documentation, discussions?
Closes #5913
### How was this PR tested?
New regression test, written first and confirmed to fail red against
unmodified source with the exact production traceback (`Cannot transit from
PAUSED to COMPLETED` at `_process_end_channel` -> `complete()`).
`test_pause_landing_before_completion_defers_it_until_resume` makes the race
deterministic by pausing inside `all_ports_completed()` — the last call before
`complete()`, with no control check in between, i.e. exactly the production
window. It pins both directions:
- **negative** — while paused: the main loop thread stays alive, the worker
sits in `PAUSED`, and no `WorkerExecutionCompleted` is announced;
- **positive** — after `Resume`: the worker reaches `COMPLETED` and
announces it.
From `amber/`:
- `python -m pytest src/test/python/core/runnables/test_main_loop.py` — 35
passed (34 pre-existing + the new one).
- The new test run 5x consecutively — passed every time (it is a race
regression, so repetition matters).
- `ruff check src/main/python src/test/python && ruff format --check
src/main/python src/test/python` — clean.
Pre-existing local failures in `test_iceberg_document.py`,
`test_tuple.py::test_hash` and `test_expression_evaluator.py` were confirmed
identical with this change stashed, so they are unrelated to it.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
