apache / apache/texera

Statistics queries can reach a worker after EndWorker, making it refuse termination

Open
#7,031 1 comment 0 reactions 1 assignee Claimed by @Neilk1021 View on GitHub
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

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.