You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@texera.apache.org by "aglinxinyuan (via GitHub)" <gi...@apache.org> on 2026/07/25 08:02:38 UTC
[PR] fix(amber): tolerate in-flight RPC acks when EndWorker terminates a worker [texera]
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(amber): tolerate in-flight RPC acks when EndWorker terminates a worker [texera]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #6892:
URL: https://github.com/apache/texera/pull/6892#issuecomment-5077631760
## [Codecov](https://app.codecov.io/gh/apache/texera/pull/6892?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 77.96%. Comparing base ([`52584a8`](https://app.codecov.io/gh/apache/texera/commit/52584a80cef1c9b9ee67b8083b617bf654be806e?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`dc535e1`](https://app.codecov.io/gh/apache/texera/commit/dc535e1b1f622fb09f73384b96c677232e57f40b?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 1 commits behind head on main.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #6892 +/- ##
============================================
+ Coverage 77.94% 77.96% +0.01%
+ Complexity 3663 3661 -2
============================================
Files 1161 1161
Lines 46034 46044 +10
Branches 5099 5099
============================================
+ Hits 35879 35896 +17
+ Misses 8535 8528 -7
Partials 1620 1620
```
| [Flag](https://app.codecov.io/gh/apache/texera/pull/6892/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | | *Carryforward flag |
|---|---|---|---|
| [access-control-service](https://app.codecov.io/gh/apache/texera/pull/6892/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.00% <ø> (ø)` | | Carriedforward from [52584a8](https://app.codecov.io/gh/apache/texera/commit/52584a80cef1c9b9ee67b8083b617bf654be806e?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [agent-service](https://app.codecov.io/gh/apache/texera/pull/6892/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `76.76% <ø> (ø)` | | Carriedforward from [52584a8](https://app.codecov.io/gh/apache/texera/commit/52584a80cef1c9b9ee67b8083b617bf654be806e?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [amber](https://app.codecov.io/gh/apache/texera/pull/6892/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.69% <ø> (-0.01%)` | :arrow_down: | Carriedforward from [52584a8](https://app.codecov.io/gh/apache/texera/commit/52584a80cef1c9b9ee67b8083b617bf654be806e?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [computing-unit-managing-service](https://app.codecov.io/gh/apache/texera/pull/6892/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `20.49% <ø> (ø)` | | Carriedforward from [52584a8](https://app.codecov.io/gh/apache/texera/commit/52584a80cef1c9b9ee67b8083b617bf654be806e?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [config-service](https://app.codecov.io/gh/apache/texera/pull/6892/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `66.66% <ø> (ø)` | | Carriedforward from [52584a8](https://app.codecov.io/gh/apache/texera/commit/52584a80cef1c9b9ee67b8083b617bf654be806e?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [file-service](https://app.codecov.io/gh/apache/texera/pull/6892/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `67.21% <ø> (ø)` | | Carriedforward from [52584a8](https://app.codecov.io/gh/apache/texera/commit/52584a80cef1c9b9ee67b8083b617bf654be806e?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [frontend](https://app.codecov.io/gh/apache/texera/pull/6892/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `82.58% <ø> (ø)` | | Carriedforward from [52584a8](https://app.codecov.io/gh/apache/texera/commit/52584a80cef1c9b9ee67b8083b617bf654be806e?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [notebook-migration-service](https://app.codecov.io/gh/apache/texera/pull/6892/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `78.94% <ø> (ø)` | | Carriedforward from [52584a8](https://app.codecov.io/gh/apache/texera/commit/52584a80cef1c9b9ee67b8083b617bf654be806e?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [pyamber](https://app.codecov.io/gh/apache/texera/pull/6892/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `92.41% <100.00%> (+0.17%)` | :arrow_up: | |
| [workflow-compiling-service](https://app.codecov.io/gh/apache/texera/pull/6892/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.14% <ø> (ø)` | | Carriedforward from [52584a8](https://app.codecov.io/gh/apache/texera/commit/52584a80cef1c9b9ee67b8083b617bf654be806e?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
*This pull request uses carry forward flags. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to find out more.
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/texera/pull/6892?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(amber): tolerate in-flight RPC acks when EndWorker terminates a worker [texera]
Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on code in PR #6892:
URL: https://github.com/apache/texera/pull/6892#discussion_r3661381292
##########
amber/src/main/python/core/architecture/handlers/control/end_worker_handler.py:
##########
@@ -37,19 +39,45 @@ async def end_worker(self, req: EmptyRequest) -> EmptyReturn:
has finished not only the data processing logic, but also the processing
of all the control messages.
"""
- # Ensure this is really the last message. Read the queued count once (InternalQueue
- # exposes size(); the base IQueue interface does not) and branch on it.
+ # Ensure this is really the last message that carries work. Read the
+ # queued count once (InternalQueue exposes size(); the base IQueue
+ # interface does not) and branch on it.
input_queue: InternalQueue = self.context.input_queue
- queued_count = input_queue.size()
- if queued_count > 0:
+ if input_queue.size() == 0:
+ # Now we can safely acknowledge that this worker can be terminated.
+ return EmptyReturn()
+
+ # RPC acks (ReturnInvocations for this worker's own fire-and-forget
+ # calls, e.g. worker_execution_completed) race with EndWorker by
+ # design: the coordinator decides to end the worker while its acks are
+ # still in flight. The worker never awaits those acks, so an ack-only
+ # backlog is safe to drop at termination. Anything else fails loudly
+ # AND is put back on the queue so the coordinator's retried EndWorker
+ # finds it processed (mirrors the Scala EndHandler).
+ pending = []
+ while not input_queue.is_empty():
+ pending.append(input_queue.get())
+
+ def is_ack(element) -> bool:
+ return isinstance(element, DCMElement) and isinstance(
+ get_one_of(element.payload, sealed=False), ReturnInvocation
+ )
+
+ if all(is_ack(element) for element in pending):
logger.warning(
- f"Received EndWorker before all {queued_count} queued "
- f"message(s) were processed; failing the RPC so a later "
- f"coordinator retry succeeds once the queue has drained."
+ f"Received EndWorker with only RPC acks left in the queue; "
+ f"proceeding with termination. Pending acks: {pending}"
)
- # Fail this RPC (the counterpart of the Scala EndHandler's
- # Future.exception) so a later coordinator retry succeeds once
- # the queue has drained, instead of dropping the pending message.
- raise RuntimeError("worker still has unprocessed messages")
- # Now we can safely acknowledge that this worker can be terminated.
- return EmptyReturn()
+ return EmptyReturn()
+
+ for element in pending:
Review Comment:
this would change the thread to be a producer (by putting elements into the queue). can we avoid it?
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(amber): tolerate in-flight RPC acks when EndWorker terminates a worker [texera]
Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on code in PR #6892:
URL: https://github.com/apache/texera/pull/6892#discussion_r3661454852
##########
amber/src/main/python/core/architecture/handlers/control/end_worker_handler.py:
##########
@@ -37,19 +39,45 @@ async def end_worker(self, req: EmptyRequest) -> EmptyReturn:
has finished not only the data processing logic, but also the processing
of all the control messages.
"""
- # Ensure this is really the last message. Read the queued count once (InternalQueue
- # exposes size(); the base IQueue interface does not) and branch on it.
+ # Ensure this is really the last message that carries work. Read the
+ # queued count once (InternalQueue exposes size(); the base IQueue
+ # interface does not) and branch on it.
input_queue: InternalQueue = self.context.input_queue
- queued_count = input_queue.size()
- if queued_count > 0:
+ if input_queue.size() == 0:
+ # Now we can safely acknowledge that this worker can be terminated.
+ return EmptyReturn()
+
+ # RPC acks (ReturnInvocations for this worker's own fire-and-forget
+ # calls, e.g. worker_execution_completed) race with EndWorker by
+ # design: the coordinator decides to end the worker while its acks are
+ # still in flight. The worker never awaits those acks, so an ack-only
+ # backlog is safe to drop at termination. Anything else fails loudly
+ # AND is put back on the queue so the coordinator's retried EndWorker
+ # finds it processed (mirrors the Scala EndHandler).
+ pending = []
+ while not input_queue.is_empty():
+ pending.append(input_queue.get())
+
+ def is_ack(element) -> bool:
+ return isinstance(element, DCMElement) and isinstance(
+ get_one_of(element.payload, sealed=False), ReturnInvocation
+ )
+
+ if all(is_ack(element) for element in pending):
logger.warning(
- f"Received EndWorker before all {queued_count} queued "
- f"message(s) were processed; failing the RPC so a later "
- f"coordinator retry succeeds once the queue has drained."
+ f"Received EndWorker with only RPC acks left in the queue; "
+ f"proceeding with termination. Pending acks: {pending}"
)
- # Fail this RPC (the counterpart of the Scala EndHandler's
- # Future.exception) so a later coordinator retry succeeds once
- # the queue has drained, instead of dropping the pending message.
- raise RuntimeError("worker still has unprocessed messages")
- # Now we can safely acknowledge that this worker can be terminated.
Review Comment:
as discussed offline,
`Received EndWorker before all {queued_count} queued message(s) were processed` is a hard requirement for EndWorker. if there are flaky scenarios causing this failure, then that means there are some uncleared message do not meet this requirement.
I think we should fix those scenarios, instead of relaxing this requirement.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(amber): tolerate in-flight RPC acks when EndWorker terminates a worker [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan closed pull request #6892: fix(amber): tolerate in-flight RPC acks when EndWorker terminates a worker
URL: https://github.com/apache/texera/pull/6892
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(amber): tolerate in-flight RPC acks when EndWorker terminates a worker [texera]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #6892:
URL: https://github.com/apache/texera/pull/6892#issuecomment-5077622098
<!-- texera-reviewer-suggestion -->
### Automated Reviewer Suggestions
Based on the `git blame` history of the changed files, we recommend the following reviewers:
- **Contributors with relevant context:** `@Yicong-Huang`
You can notify them by mentioning `@Yicong-Huang` in a comment.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(amber): tolerate in-flight RPC acks when EndWorker terminates a worker [texera]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #6892:
URL: https://github.com/apache/texera/pull/6892#issuecomment-5077639307
<!-- texera-benchmarks-comment -->
## ⚠️ Benchmark changes need a look
🟢 2 better · 🔴 5 worse · ⚪ 8 noise (<±5%) · 0 without baseline
> Compared against main `81d4612` benchmarked on **this same runner**, so the delta is largely free of cross-runner hardware noise. The "7d avg" column still reflects the gh-pages dashboard. Treat <±5% as noise unless repeated.
[Dashboard](https://apache.github.io/texera/dev/bench/) · [Run](https://github.com/apache/texera/actions/runs/30150398150)
| | config | throughput | MB/s | latency | max Δ latest / 7d |
|---|---|---:|---:|---:|---:|
| 🔴 | bs=10 sw=10 sl=64 | 389 | 0.237 | 25,774/35,475/35,475 us | 🔴 +26.4% / 🔴 +113.8% |
| 🟢 | bs=100 sw=10 sl=64 | 846 | 0.516 | 116,788/131,065/131,065 us | 🟢 -6.2% / 🔴 +20.1% |
| ⚪ | bs=1000 sw=10 sl=64 | 933 | 0.57 | 1,076,946/1,115,838/1,115,838 us | ⚪ within ±5% / 🔴 +7.4% |
<details><summary>Baseline details</summary>
Latest main `81d4612` from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---:|---:|---:|---:|---:|
| bs=10 sw=10 sl=64 | throughput | 389 tuples/sec | 460 tuples/sec | 754.55 tuples/sec | -15.4% | -48.4% |
| bs=10 sw=10 sl=64 | MB/s | 0.237 MB/s | 0.28 MB/s | 0.461 MB/s | -15.4% | -48.5% |
| bs=10 sw=10 sl=64 | p50 | 25,774 us | 21,991 us | 12,816 us | +17.2% | +101.1% |
| bs=10 sw=10 sl=64 | p95 | 35,475 us | 28,056 us | 16,594 us | +26.4% | +113.8% |
| bs=10 sw=10 sl=64 | p99 | 35,475 us | 28,056 us | 19,806 us | +26.4% | +79.1% |
| bs=100 sw=10 sl=64 | throughput | 846 tuples/sec | 863 tuples/sec | 969.38 tuples/sec | -2.0% | -12.7% |
| bs=100 sw=10 sl=64 | MB/s | 0.516 MB/s | 0.527 MB/s | 0.592 MB/s | -2.1% | -12.8% |
| bs=100 sw=10 sl=64 | p50 | 116,788 us | 115,340 us | 103,584 us | +1.3% | +12.7% |
| bs=100 sw=10 sl=64 | p95 | 131,065 us | 139,753 us | 109,097 us | -6.2% | +20.1% |
| bs=100 sw=10 sl=64 | p99 | 131,065 us | 139,753 us | 117,304 us | -6.2% | +11.7% |
| bs=1000 sw=10 sl=64 | throughput | 933 tuples/sec | 954 tuples/sec | 1,004 tuples/sec | -2.2% | -7.0% |
| bs=1000 sw=10 sl=64 | MB/s | 0.57 MB/s | 0.582 MB/s | 0.613 MB/s | -2.1% | -6.9% |
| bs=1000 sw=10 sl=64 | p50 | 1,076,946 us | 1,051,650 us | 1,002,357 us | +2.4% | +7.4% |
| bs=1000 sw=10 sl=64 | p95 | 1,115,838 us | 1,076,220 us | 1,046,463 us | +3.7% | +6.6% |
| bs=1000 sw=10 sl=64 | p99 | 1,115,838 us | 1,076,220 us | 1,073,661 us | +3.7% | +3.9% |
</details>
<details><summary>Raw CSV</summary>
```csv
config_idx,batch_size,schema_width,string_len,num_batches,total_ms,total_tuples,total_bytes,tuples_per_sec,mb_per_sec,lat_p50_us,lat_p95_us,lat_p99_us
0,10,10,64,20,514.17,200,128000,389,0.237,25773.69,35474.99,35474.99
1,100,10,64,20,2364.77,2000,1280000,846,0.516,116787.80,131064.93,131064.93
2,1000,10,64,20,21427.65,20000,12800000,933,0.570,1076946.36,1115838.07,1115838.07
```
</details>
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(amber): tolerate in-flight RPC acks when EndWorker terminates a worker [texera]
Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on code in PR #6892:
URL: https://github.com/apache/texera/pull/6892#discussion_r3661458677
##########
amber/src/main/python/core/architecture/handlers/control/end_worker_handler.py:
##########
@@ -37,19 +39,45 @@ async def end_worker(self, req: EmptyRequest) -> EmptyReturn:
has finished not only the data processing logic, but also the processing
of all the control messages.
"""
- # Ensure this is really the last message. Read the queued count once (InternalQueue
- # exposes size(); the base IQueue interface does not) and branch on it.
+ # Ensure this is really the last message that carries work. Read the
+ # queued count once (InternalQueue exposes size(); the base IQueue
+ # interface does not) and branch on it.
input_queue: InternalQueue = self.context.input_queue
- queued_count = input_queue.size()
- if queued_count > 0:
+ if input_queue.size() == 0:
+ # Now we can safely acknowledge that this worker can be terminated.
+ return EmptyReturn()
+
+ # RPC acks (ReturnInvocations for this worker's own fire-and-forget
+ # calls, e.g. worker_execution_completed) race with EndWorker by
+ # design: the coordinator decides to end the worker while its acks are
+ # still in flight. The worker never awaits those acks, so an ack-only
+ # backlog is safe to drop at termination. Anything else fails loudly
+ # AND is put back on the queue so the coordinator's retried EndWorker
+ # finds it processed (mirrors the Scala EndHandler).
+ pending = []
+ while not input_queue.is_empty():
+ pending.append(input_queue.get())
+
+ def is_ack(element) -> bool:
+ return isinstance(element, DCMElement) and isinstance(
+ get_one_of(element.payload, sealed=False), ReturnInvocation
+ )
+
+ if all(is_ack(element) for element in pending):
logger.warning(
- f"Received EndWorker before all {queued_count} queued "
- f"message(s) were processed; failing the RPC so a later "
- f"coordinator retry succeeds once the queue has drained."
+ f"Received EndWorker with only RPC acks left in the queue; "
+ f"proceeding with termination. Pending acks: {pending}"
)
- # Fail this RPC (the counterpart of the Scala EndHandler's
- # Future.exception) so a later coordinator retry succeeds once
- # the queue has drained, instead of dropping the pending message.
- raise RuntimeError("worker still has unprocessed messages")
- # Now we can safely acknowledge that this worker can be terminated.
- return EmptyReturn()
+ return EmptyReturn()
+
+ for element in pending:
Review Comment:
for actor model, we don't allow peek into the next message.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(amber): tolerate in-flight RPC acks when EndWorker terminates a worker [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan commented on PR #6892:
URL: https://github.com/apache/texera/pull/6892#issuecomment-5099722699
Superseded by #6960, which takes the direction suggested in review: instead of tolerating in-flight acks on the worker side, it fixes the send-order scenario — portCompleted now replies before its continuation fans out EndWorker (the shape workerExecutionCompleted/consoleMessageTriggered/workerStateUpdated already have), so the ack can never trail EndWorker and the strict worker-side requirement is kept unchanged. Thanks @Yicong-Huang for pushing back on the diagnosis here; the ordering question you raised is answered in detail in #6960's description.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org