The GitHub Actions job "Required Checks" on texera.git/main has failed. Run started by GitHub user github-merge-queue[bot] (triggered by github-merge-queue[bot]).
Head commit for run: 7e4a9b46b863fc13fa76ba74980a0f0c4beff48f / Xinyuan Lin <[email protected]> fix(pyamber): stop EndWorker from consuming straggler messages (#6522) ### What changes were proposed in this PR? **Root cause.** The Python worker's `EndWorkerHandler` guard logs its "unprocessed messages" warning through `input_queue.get()` — a destructive, blocking read that removes a pending message — and then `assert input_queue.is_empty()`. With exactly one straggler message the straggler is silently dropped and `EndWorker` is acknowledged as success; with two or more, one message is still destroyed and the RPC fails with a bare `AssertionError` — and because the coordinator retries `EndWorker`, every retry that hits the guard eats another queued message until the last one is dropped under a success ack. | queue state at `EndWorker` | before | after | |---|---|---| | empty | ack | ack (unchanged) | | 1 straggler | straggler **dropped**, then ack | RPC fails, straggler kept | | ≥ 2 stragglers | 1 dropped, then `AssertionError` | RPC fails, all kept | **Fix.** `end_worker_handler.py` reads the queued count once via `input_queue.size()` and branches on it, so nothing is consumed. When the queue is non-empty it logs the pending count and raises `RuntimeError("worker still has unprocessed messages")` instead of consuming a message and asserting. The raise rides the existing failure path — `AsyncRPCServer.receive` converts a handler exception into a `ControlError` reply, `AsyncRPCClient.fulfillPromise` on the coordinator turns it into a failed future, and `RegionExecutionManager.terminateWorkersWithRetry` re-sends `EndWorker` on a fixed `killRetryDelay` (bounded by `maxTerminationAttempts`), succeeding once the queue has drained — the exact Python analogue of the Scala `EndHandler`'s `Future.exception`: ``` coordinator worker input queue at EndWorker arrival | [ReturnInvocation, ..., EndWorker] |-- EndWorker ---------------->| |<- ControlError --------------| size() for the log; nothing consumed | (retry after delay) |-- EndWorker ---------------->| queue drained by the main loop |<- EmptyReturn ---------------| safe to gracefulStop ``` The local is annotated as `InternalQueue` because `size()` lives on `InternalQueue`, not the base `IQueue` interface. No new inspection API (e.g. `peek()`) is added — `InternalQueue` stays non-peekable, like `queue.Queue` — so the change is confined to the handler plus its new test. ### Any related issues, documentation, discussions? Closes #6521 ### How was this PR tested? TDD — the tests were written first and fail against the unfixed handler (with 1 straggler: no exception is raised and the message vanishes; with ≥ 2: `AssertionError` instead of a clean RPC failure): - New `src/test/python/core/architecture/handlers/control/test_end_worker_handler.py`: acks on an empty queue; fails the RPC with one straggler; does **not** consume the straggler; keeps all messages with two stragglers; acks again once the queue drains (the retry protocol end-to-end). This is the Python analogue of the Scala `EndHandlerSpec`. Ran locally: `cd amber && pytest -m "not integration"` on the touched test file plus the full unit suite, and `ruff check src/main/python src/test/python && ruff format --check src/main/python src/test/python`. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Fable 5) Report URL: https://github.com/apache/texera/actions/runs/30150537116 With regards, GitHub Actions via GitBox
