aglinxinyuan opened a new pull request, #6916:
URL: https://github.com/apache/texera/pull/6916
### What changes were proposed in this PR?
**Root cause.** On every normal region termination, the coordinator sends
`EndWorker` from *inside* the `portCompleted` handler, so the
`ReturnInvocation` replying to that same `portCompleted` is always emitted
**after** `EndWorker` on the same FIFO control channel:
```
Coordinator — portCompleted(W) handler, one synchronous continuation:
1. advanceRegionExecutions() ──> sends EndWorker to every worker (seq N
on W's control channel)
2. returns EmptyReturn() ──> sends ReturnInvocation to W (seq
N+1, same channel)
Worker W — processes EndWorker while ReturnInvocation(N+1) may already sit
in its arrival queue:
old check: ANY queued element => IllegalStateException("worker still has
unprocessed messages")
=> attempt 1 of 150 fails => 200 ms retry succeeds
```
Whether the trailing reply has landed when `EndHandler` runs is a
thread-scheduling race, so healthy teardowns routinely fail the first
termination attempt and log an alarming storm. A queued `ReturnInvocation` is
not work: processing it only fulfills a promise for a request the worker
already issued, and every worker-to-coordinator call discards its future
(`DataProcessor.scala:179,187,224,312`), so no continuation can be lost.
**Fix.** Both workers now exclude queued `ReturnInvocation`s from the
unprocessed-messages check. Everything that *is* work — `ControlInvocation`s,
data, ECMs, timer-based controls, actor commands — still blocks termination,
and the 150 × 200 ms retry/give-up contract is untouched.
| | Before | After |
|---|---|---|
| Normal teardown logs | 2 ERROR + 2 WARN + 2 stack traces | none |
| First `EndWorker` attempt | fails on a benign race | succeeds |
| Teardown latency | +200 ms retry | no retry |
| Real queued work | blocks termination | still blocks termination |
| Python: queued reply | silently **dropped** (destructive `get()` inside
the warning f-string), then a bare `assert` passes; with 2+ elements the
coordinator gets a blank-message `ControlError` | kept (non-destructive
`peek()`); failure raises `RuntimeError` with the same message string as the
Scala side |
Changes:
- **`EndHandler.scala`**: scan the arrival queue for pending *work* instead
of testing bare emptiness; the warning names the payload type and channel
instead of dumping the element; scaladoc no longer claims `EndWorker` is the
last message a worker receives.
- **`end_worker_handler.py` / `internal_queue.py`**: new non-destructive
`InternalQueue.peek()` and a reply-aware `has_unprocessed_work()` (an
`AtomicInteger` work counter maintained in `put`/`get`, since the multi-queue
offers no iteration; it also sees work in paused/backpressured sub-queues that
`size()`/`get()` hide).
- **`RegionExecutionManager.scala`**: fixed the false doc ("This will be the
last message each worker receives") and demoted the detail-free duplicate
`"Error when terminating region X."` WARN to debug —
`terminateWorkersWithRetry` logs the same failure with attempt context (WARN)
or gives up loudly (ERROR), unchanged.
One deliberate behavior change to be aware of: a Python worker that used to
"succeed" `EndWorker` by silently swallowing a queued message now honestly
reports unprocessed work and lets the coordinator retry.
### Any related issues, documentation, discussions?
Related: #6796 — this addresses the `EndWorker` retry-storm item
behaviorally (the expected race no longer fails the first attempt) rather than
by adjusting log levels. Follow-up to the CI log-verbosity work in #6797.
### How was this PR tested?
- **`EndHandlerSpec`** (the three existing tests are untouched — control
invocations and actor commands still fail `endWorker`): new regression tests
for the exact CI payload (`ReturnInvocation(2, EmptyReturn())` queued →
success), multiple queued replies, a `ControlInvocation` queued *behind* a
reply (the scan must not stop at the head), and queue-size preservation on both
paths. Written test-first: the two new success tests fail on `main`.
- **`test_end_worker_handler.py`** (new; this handler previously had zero
Python coverage): success keeps the queued reply (`queue.size() == 1`
afterwards — fails on `main`, where the element was destructively consumed),
`RuntimeError` with the shared message string for queued invocations/data, and
idempotence across coordinator retries.
- **`test_internal_queue.py`**: `peek()` (None on empty, non-destructive)
and `has_unprocessed_work()` (ignores replies; counts invocations, data, ECMs,
SYSTEM items; sees data in a disabled sub-queue; not counted for rejected
elements).
- Untouched and green as evidence the retry contract is unchanged:
`RegionExecutionManagerSpec`, `WorkflowExecutionManagerSpec`, `DPThreadSpec`,
`WorkerSpec`, `AsyncRPCClientSpec` (43/43).
- Lint: `scalafmtCheck`, `scalafixAll --check`, `ruff check` + `ruff format
--check` all pass.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Fable 5)
--
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]
