What happened?
A queryStatistics request can reach a worker after the coordinator has sent it EndWorker, which makes the worker correctly refuse termination and produces the loud retry that #6891 is about — from a source that message ordering cannot fix.
QueryWorkerStatisticsHandler walks the physical plan in layers and awaits each layer before emitting the next:
// QueryWorkerStatisticsHandler.scala:171
Future.collect(futures).flatMap(_ => processLayers(rest))
So one statistics query is spread across many coordinator rounds. Both kinds of query have this shape:
- completion-time (filtered) — fired by
WorkerExecutionCompletedHandler.scala:54-58 and PortCompletedHandler. It is not limited to the sender: opFilter is expanded to all transitive upstream operators (:106-121), so the traversal walks the sender's whole upstream cone.
- periodic (full-graph) — the coordinator's timer-driven collection, which walks every layer.
Meanwhile region termination is driven by port completion. Once the last port of a region is booked, EndWorker fans out to that region's workers. If a later layer of an in-flight statistics traversal then emits queryStatistics to one of those workers, that worker has a genuine unprocessed control message when it handles EndWorker, so EndHandler fails the request — 2 ERRORs + 2 WARNs + 2 stack traces, then the 200 ms retry.
Note this is not the trailing-reply race fixed in #6960. There, the queued element is a reply that carries no work, so the worker can safely ignore it. Here it is a real request that must be processed, so failing EndWorker is the correct behaviour — the defect is that the coordinator queries workers it is simultaneously tearing down. Existing partial mitigations: completed operators are skipped (:150-153), and the completing worker's own query is emitted early.
Possible directions:
- Make the final statistics query part of the termination sequence (query, then
EndWorker) instead of racing it.
- Skip statistics queries to workers of a region whose termination has started.
- Have
terminateWorkers wait for statistics queries outstanding to that region's workers.
How to reproduce?
Not reproduced in the logs I have — all three worker still has unprocessed messages occurrences across the pre-#6960 ubuntu+macos amber-integration runs were ReturnInvocations, none were ControlInvocations. This is a code-path finding: the interleaving above requires a statistics traversal to still be walking layers when a region completes. The observable symptom would be an EndHandler warning whose queued payload is a ControlInvocation(queryStatistics, ...) rather than a ReturnInvocation, so that string is the thing to grep for in teardown logs.
Version/Branch
main (observed at a61702f; found while reviewing #6960, which fixes the trailing-reply half of #6891).
What happened?
A
queryStatisticsrequest can reach a worker after the coordinator has sent itEndWorker, which makes the worker correctly refuse termination and produces the loud retry that #6891 is about — from a source that message ordering cannot fix.QueryWorkerStatisticsHandlerwalks the physical plan in layers and awaits each layer before emitting the next:So one statistics query is spread across many coordinator rounds. Both kinds of query have this shape:
WorkerExecutionCompletedHandler.scala:54-58andPortCompletedHandler. It is not limited to the sender:opFilteris expanded to all transitive upstream operators (:106-121), so the traversal walks the sender's whole upstream cone.Meanwhile region termination is driven by port completion. Once the last port of a region is booked,
EndWorkerfans out to that region's workers. If a later layer of an in-flight statistics traversal then emitsqueryStatisticsto one of those workers, that worker has a genuine unprocessed control message when it handlesEndWorker, soEndHandlerfails the request — 2 ERRORs + 2 WARNs + 2 stack traces, then the 200 ms retry.Note this is not the trailing-reply race fixed in #6960. There, the queued element is a reply that carries no work, so the worker can safely ignore it. Here it is a real request that must be processed, so failing
EndWorkeris the correct behaviour — the defect is that the coordinator queries workers it is simultaneously tearing down. Existing partial mitigations: completed operators are skipped (:150-153), and the completing worker's own query is emitted early.Possible directions:
EndWorker) instead of racing it.terminateWorkerswait for statistics queries outstanding to that region's workers.How to reproduce?
Not reproduced in the logs I have — all three
worker still has unprocessed messagesoccurrences across the pre-#6960 ubuntu+macosamber-integrationruns wereReturnInvocations, none wereControlInvocations. This is a code-path finding: the interleaving above requires a statistics traversal to still be walking layers when a region completes. The observable symptom would be anEndHandlerwarning whose queued payload is aControlInvocation(queryStatistics, ...)rather than aReturnInvocation, so that string is the thing to grep for in teardown logs.Version/Branch
main (observed at a61702f; found while reviewing #6960, which fixes the trailing-reply half of #6891).