fix(pyamber, v1.2): stop EndWorker from consuming straggler messages - #6995
Open
Yicong-Huang wants to merge 1 commit into
Open
fix(pyamber, v1.2): stop EndWorker from consuming straggler messages #6995Yicong-Huang wants to merge 1 commit into
Yicong-Huang wants to merge 1 commit into
Conversation
### What changes were proposed in this PR?
**Root cause.** The Python worker's `EndWorkerHandler` guard logs its
"unprocessed messages" warning through `input_queue.get()` — a
destructive, blocking read that removes a pending message — and then
`assert input_queue.is_empty()`. With exactly one straggler message the
straggler is silently dropped and `EndWorker` is acknowledged as
success; with two or more, one message is still destroyed and the RPC
fails with a bare `AssertionError` — and because the coordinator retries
`EndWorker`, every retry that hits the guard eats another queued message
until the last one is dropped under a success ack.
| queue state at `EndWorker` | before | after |
|---|---|---|
| empty | ack | ack (unchanged) |
| 1 straggler | straggler **dropped**, then ack | RPC fails, straggler
kept |
| ≥ 2 stragglers | 1 dropped, then `AssertionError` | RPC fails, all
kept |
**Fix.** `end_worker_handler.py` reads the queued count once via
`input_queue.size()` and branches on it, so nothing is consumed. When
the queue is non-empty it logs the pending count and raises
`RuntimeError("worker still has unprocessed messages")` instead of
consuming a message and asserting. The raise rides the existing failure
path — `AsyncRPCServer.receive` converts a handler exception into a
`ControlError` reply, `AsyncRPCClient.fulfillPromise` on the coordinator
turns it into a failed future, and
`RegionExecutionManager.terminateWorkersWithRetry` re-sends `EndWorker`
on a fixed `killRetryDelay` (bounded by `maxTerminationAttempts`),
succeeding once the queue has drained — the exact Python analogue of the
Scala `EndHandler`'s `Future.exception`:
```
coordinator worker input queue at EndWorker arrival
| [ReturnInvocation, ..., EndWorker]
|-- EndWorker ---------------->|
|<- ControlError --------------| size() for the log; nothing consumed
| (retry after delay)
|-- EndWorker ---------------->| queue drained by the main loop
|<- EmptyReturn ---------------| safe to gracefulStop
```
The local is annotated as `InternalQueue` because `size()` lives on
`InternalQueue`, not the base `IQueue` interface. No new inspection API
(e.g. `peek()`) is added — `InternalQueue` stays non-peekable, like
`queue.Queue` — so the change is confined to the handler plus its new
test.
### Any related issues, documentation, discussions?
Closes #6521
### How was this PR tested?
TDD — the tests were written first and fail against the unfixed handler
(with 1 straggler: no exception is raised and the message vanishes; with
≥ 2: `AssertionError` instead of a clean RPC failure):
- New
`src/test/python/core/architecture/handlers/control/test_end_worker_handler.py`:
acks on an empty queue; fails the RPC with one straggler; does **not**
consume the straggler; keeps all messages with two stragglers; acks
again once the queue drains (the retry protocol end-to-end). This is the
Python analogue of the Scala `EndHandlerSpec`.
Ran locally: `cd amber && pytest -m "not integration"` on the touched
test file plus the full unit suite, and `ruff check src/main/python
src/test/python && ruff format --check src/main/python src/test/python`.
### Was this PR authored or co-authored using generative AI tooling?
(backported from commit 7e4a9b4)
Generated-by: Claude Code (Fable 5)
Contributor
Automated Reviewer SuggestionsBased on the
|
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## release/v1.2 #6995 +/- ##
==================================================
+ Coverage 53.95% 53.96% +0.01%
Complexity 1441 1441
==================================================
Files 809 809
Lines 34144 34145 +1
Branches 3448 3448
==================================================
+ Hits 18421 18428 +7
+ Misses 14815 14809 -6
Partials 908 908
*This pull request uses carry forward flags. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this PR?
Backport of #6522 to
release/v1.2, cherry-picked from 7e4a9b4. The cherry-pick applied cleanly.Follows the Direct Backport Push convention; opened as a PR (rather than a direct push) per a backport-coverage audit.
Any related issues, documentation, discussions?
Backport of #6522. Originally linked #6521.
How was this PR tested?
Release-branch CI runs on this PR. Cherry-pick applied cleanly onto
release/v1.2; no manual conflict resolution was needed.Was this PR authored or co-authored using generative AI tooling?
Yes — backport prepared with Claude Code (mechanical cherry-pick; the change itself is #6522 by its original author).
🤖 Generated with Claude Code