aglinxinyuan opened a new issue, #6521: URL: https://github.com/apache/texera/issues/6521
### What happened? `EndWorkerHandler.end_worker` in the Python worker guards against acknowledging `EndWorker` while its input queue still holds unprocessed messages — but the guard's warning log calls `input_queue.get()` inside the f-string, which **removes** a pending message from the queue, and then asserts emptiness: https://github.com/apache/texera/blob/39db110e74b4bdcf0e4b0df961c25973e6bbb20c/amber/src/main/python/core/architecture/handlers/control/end_worker_handler.py#L40-L47 | stragglers in the queue | what happens today | |---|---| | exactly 1 | straggler silently consumed → `assert` passes → `EndWorker` acknowledged as success, message dropped | | 2 or more | one straggler consumed → `AssertionError` → the RPC fails with a misleading error, and the coordinator's retry loop makes it worse: every retry that hits the guard destroys another queued message, until the last straggler is dropped under a success ack | The Scala worker handles the same race correctly: `EndHandler` logs via the non-destructive `inputMessageQueue.peek()` and returns `Future.exception`, so the coordinator's `terminateWorkersWithRetry` re-sends `EndWorker` after the worker drains its queue: https://github.com/apache/texera/blob/39db110e74b4bdcf0e4b0df961c25973e6bbb20c/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/EndHandler.scala#L41-L55 ``` coordinator worker input queue at EndWorker arrival | [ReturnInvocation, ..., EndWorker] |-- EndWorker ---------------->| | Scala: peek() + fail RPC --> coordinator retries --> ack after drain | Python: get() DROPS a message --> false ack (or AssertionError if >=2) ``` The coordinator side of the retry protocol already exists and is pinned by `RegionExecutionManagerSpec` ("retry EndWorker failures and delay gracefulStop until a retry succeeds"); the Python worker just never participates in it — it either lies (acks after dropping a message) or fails with an `AssertionError` unrelated to the real condition. A second gap underneath: `InternalQueue` exposes no `peek()` at all — only the destructive, blocking `get()` — even though the underlying `LinkedBlockingMultiQueue` already implements a non-destructive `peek()`. **Expected behavior:** the Python worker mirrors the Scala semantics — warn via a non-destructive peek and fail the RPC, leaving the queue intact. (The plumbing already exists: a handler exception is converted by `AsyncRPCServer.receive` into a `ControlError` reply, which `AsyncRPCClient.fulfillPromise` on the coordinator turns into a failed future feeding the retry loop.) ### How to reproduce? Race window: `EndWorker` is queued behind another not-yet-processed message on a Python UDF worker — e.g. a `ReturnInvocation` reply from CONTROLLER that arrived just before teardown. Observed repeatedly in control-flow (loop/If) end-to-end runs: the worker logs `Received EndHandler before all messages are processed` and the queued reply vanishes (or the RPC dies with `AssertionError`). Deterministic repro at the unit level: put one `DCMElement` into a worker's `InternalQueue` and invoke `end_worker` — it acknowledges success and the element is gone: ```python input_queue = InternalQueue() input_queue.put(dcm_element) # a pending control message await handler.end_worker(EmptyRequest()) # returns EmptyReturn() — should fail assert input_queue.is_empty() # True — the message was silently dropped ``` ### Version/Branch 1.3.0-incubating-SNAPSHOT (main) ### Relevant log output ``` WARNING ... Received EndHandler before all messages are processed. Unprocessed messages: DCMElement(tag=ChannelIdentity(from_worker_id=ActorVirtualIdentity(name='CONTROLLER'), ...), payload=DirectControlMessagePayloadV2(return_invocation=ReturnInvocation(...))) ``` -- 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]
