Statistics queries can reach a worker after EndWorker, making it refuse termination
- Dominant language
- Scala
- Stars
- 314
- Forks
- 187
- Avg merge
- 1d 21h
- Merged PRs (30d)
- 214
Description
### 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**:
```scala
// 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 `ReturnInvocation`s, none were `ControlInvocation`s. 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 a61702fd10; found while reviewing #6960, which fixes the trailing-reply half of #6891).
Contributor guide
Assessment
This issue has not been assessed yet.