aglinxinyuan opened a new pull request, #6892:
URL: https://github.com/apache/texera/pull/6892
### What changes were proposed in this PR?
Fixes the ~1-in-10 `LoopIntegrationSpec` flake in `amber-integration`.
Region termination logs:
```
java.lang.IllegalStateException: worker still has unprocessed messages
at ...promisehandlers.EndHandler.endWorker(EndHandler.scala:51)
[WARN] Failed to terminate region N on attempt 1 of 150. Retrying in 200 ms.
```
**Root cause.** The "unprocessed" message is always a **`ReturnInvocation`**
— the coordinator's ack for an RPC the worker itself sent fire-and-forget
(`workerExecutionCompleted` / `portCompleted`). The race is inherent: that very
RPC is what makes the coordinator decide to end the worker, so the ack and
`EndWorker` are concurrently in flight:
```
worker ──workerExecutionCompleted──▶ coordinator
│ decides region is done
worker ◀───────ack (ReturnInvocation)───┤ (ack still in flight)
worker ◀───────EndWorker────────────────┘
EndWorker sees the queued ack → throws
```
The worker never awaits these acks, so an ack-only backlog carries no work.
Loops amplify the race because every iteration terminates and re-executes
regions — and the suite's single retry doesn't save runs where it strikes twice
(same fail-fast previously seen in the `DataProcessingSpec` flake, #5614).
**Fix (both sides of the engine, mirrored):**
| Side | Before | After |
|---|---|---|
| Scala `EndHandler` | throws on *any* queued message | ack-only backlog →
warn + succeed; anything else (control invocation, data, ECM, actor command)
still fails so the termination retry lets the worker drain it |
| Python `EndWorkerHandler` | **consumed one message as a side effect of
logging it**, then bare `assert empty` (crashed on 2 queued acks; silently
swallowed real work) | same ack-only leniency; non-ack messages are **put back
on the queue** before failing |
### Any related issues, documentation, discussions?
Closes #6891. Related: #5614 (the same fail-fast race in
`DataProcessingSpec`).
### How was this PR tested?
- `EndHandlerSpec` (Scala): new cases — ack-only backlog succeeds; ack +
real work still fails. Existing cases (empty queue succeeds; control message /
actor command fail) unchanged and passing.
- `test_end_worker_handler.py` (Python, new): ack-only backlog succeeds;
empty queue succeeds; ack + control invocation fails **and the control
invocation survives in the queue** (pins the old swallow-one bug).
- Full local battery: test-compile, `scalafmtCheckAll`, `scalafixAll
--check`, ruff format+check, and the loop-adjacent Python suite (68 tests) all
green (Java 17).
- The real-world validation is statistical: the `amber-integration` job on
this PR (and subsequent PRs) should stop hitting the termination-retry flake.
### 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]
