aglinxinyuan opened a new issue, #6918: URL: https://github.com/apache/texera/issues/6918
### What happened? `EndWorker` can acknowledge termination while unprocessed **data** sits in the worker's per-channel FIFO queues, letting the coordinator `gracefulStop` the actor and silently drop tuples. The check in `EndHandler` inspects only the raw arrival queue (`dp.inputManager.inputMessageQueue`). But the DP thread's main loop drains that queue wholesale into per-channel `AmberFIFOChannel.fifoQueue`s (step 1, `DPThread.scala:141-160`) **before** it processes a single message (step 3), so a `DataFrame` that arrived slightly before `EndWorker` is invisible to the check: ``` arrival queue: [ EndWorker ] <- checked by EndHandler (empty after step-1 drain => "success") channel queues: [ DataFrame, ... ] <- never checked; dropped when the actor is stopped ``` On success the coordinator runs `removeActorRef` + `removeControlChannel` + `gracefulStop` (`RegionExecutionManager.scala:190-208`), so any data still queued in a channel is lost. This is the false-negative half of the predicate; the false-positive half (a queued coordinator `ReturnInvocation` failing healthy teardowns) was fixed in #6916, which deliberately left this direction unchanged because it is a behavior change in the strict direction. A fix needs care: the predicate must cover **data** channels only (e.g. `!inputGateway.getAllDataChannels.exists(_.hasMessage)`) — on the control channel, the reply to the `portCompleted` that triggered termination legitimately sits behind `EndWorker` by construction (see #6916). It should also consider out-of-order stashed messages (`AmberFIFOChannel.ofoMap`) and half-consumed state (`inputManager.hasUnfinishedInput` / `outputManager.hasUnfinishedOutput`), and needs an integration soak to confirm no paused/backpressured-channel scenario can make region termination fail all 150 attempts. ### How to reproduce? Unit-level (deterministic, follows the `EndHandlerSpec` harness): build a `DataProcessor` over an empty arrival queue, push a message into a data channel via `dp.inputGateway.getChannel(dataChannelId).acceptMessage(WorkflowFIFOMessage(dataChannelId, 0, DataFrame(...)))`, then call `endWorker` — it succeeds even though a data message is queued. In production the window is the race between an upstream flush and region completion. ### Version/Branch main (observed at 429be110a7; discovered during the investigation for #6916). -- 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]
