aglinxinyuan opened a new pull request, #6522:
URL: https://github.com/apache/texera/pull/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 |
Changes:
- `end_worker_handler.py`: log via a non-destructive `peek()` and raise
`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` after the queue drains — the exact Python analogue of the Scala
`EndHandler`'s `Future.exception`:
```
coordinator worker input queue at EndWorker arrival
| [ReturnInvocation, ..., EndWorker]
|-- EndWorker ---------------->|
|<- ControlError --------------| peek() for the log; nothing consumed
| (retry after delay)
|-- EndWorker ---------------->| queue drained by the main loop
|<- EmptyReturn ---------------| safe to gracefulStop
```
- `internal_queue.py`: expose `InternalQueue.peek()`, delegating to the
existing non-destructive `LinkedBlockingMultiQueue.peek()` (mirrors what the
Scala `EndHandler` reads via `inputMessageQueue.peek()`).
### 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`.
- `src/test/python/core/models/test_internal_queue.py`: `peek()` on an empty
queue returns `None`; `peek()` is non-destructive under repeated calls;
`peek()` always surfaces exactly the element the next `get()` returns across
sub-queues (deliberately not pinning control-before-data ordering — the
registration-order quirk is already documented by this file's `xfail`).
Ran locally: `cd amber && pytest -m "not integration"` on the touched test
files 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)
--
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]
