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/18 11:28:24 UTC
[PR] fix(pyamber): stop EndWorker from consuming straggler messages [texera]
aglinxinyuan opened a new pull request, #6522:
URL: https://github.com/apache/texera/pull/6522
### 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 |
Changes:
- `end_worker_handler.py`: log via a non-destructive `peek()` and raise `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` after the queue drains — the exact Python analogue of the Scala `EndHandler`'s `Future.exception`:
```
coordinator worker input queue at EndWorker arrival
| [ReturnInvocation, ..., EndWorker]
|-- EndWorker ---------------->|
|<- ControlError --------------| peek() for the log; nothing consumed
| (retry after delay)
|-- EndWorker ---------------->| queue drained by the main loop
|<- EmptyReturn ---------------| safe to gracefulStop
```
- `internal_queue.py`: expose `InternalQueue.peek()`, delegating to the existing non-destructive `LinkedBlockingMultiQueue.peek()` (mirrors what the Scala `EndHandler` reads via `inputMessageQueue.peek()`).
### 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`.
- `src/test/python/core/models/test_internal_queue.py`: `peek()` on an empty queue returns `None`; `peek()` is non-destructive under repeated calls; `peek()` always surfaces exactly the element the next `get()` returns across sub-queues (deliberately not pinning control-before-data ordering — the registration-order quirk is already documented by this file's `xfail`).
Ran locally: `cd amber && pytest -m "not integration"` on the touched test files 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?
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(pyamber): stop EndWorker from consuming straggler messages [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan commented on code in PR #6522:
URL: https://github.com/apache/texera/pull/6522#discussion_r3649815699
##########
amber/src/test/python/core/architecture/handlers/control/test_end_worker_handler.py:
##########
@@ -0,0 +1,97 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import asyncio
+from types import SimpleNamespace
+
+import pytest
+
+from core.architecture.handlers.control.end_worker_handler import EndWorkerHandler
+from core.models.internal_queue import DCMElement, InternalQueue
+from proto.org.apache.texera.amber.core import ActorVirtualIdentity, ChannelIdentity
+from proto.org.apache.texera.amber.engine.architecture.rpc import (
+ EmptyRequest,
+ EmptyReturn,
+)
+from proto.org.apache.texera.amber.engine.common import DirectControlMessagePayloadV2
+
+
+class TestEndWorkerHandler:
+ @pytest.fixture
+ def input_queue(self):
+ return InternalQueue()
+
+ @pytest.fixture
+ def handler(self, input_queue):
+ # ControlHandler.__init__ just stashes context; bypass the protobuf
+ # base class' __init__ by constructing via __new__.
+ instance = EndWorkerHandler.__new__(EndWorkerHandler)
+ instance.context = SimpleNamespace(input_queue=input_queue)
+ return instance
Review Comment:
Done — switched to direct construction (`EndWorkerHandler(SimpleNamespace(input_queue=input_queue))`) and dropped the stale bypass comment. You are right that there was nothing to bypass — `ControlHandler.__init__` just stashes `context`.
##########
amber/src/main/python/core/architecture/handlers/control/end_worker_handler.py:
##########
@@ -37,13 +37,19 @@ 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.
- input_queue: IQueue = self.context.input_queue
- if not input_queue.is_empty():
+ # 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.
+ input_queue: InternalQueue = self.context.input_queue
+ queued_count = input_queue.size()
+ if queued_count > 0:
logger.warning(
- f"Received EndHandler before all messages are "
- f"processed. Unprocessed messages: {input_queue.get()}"
+ f"Received EndWorker before all {queued_count} queued "
+ f"message(s) were processed; failing the RPC so the coordinator "
+ f"retries once the queue has drained."
Review Comment:
Done — took your wording for the log, and applied the same fixed-delay correction to the comment below it, a comment in the test file, and the PR description, which all carried the "retries once/after the queue drains" causality.
--
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(pyamber): stop EndWorker from consuming straggler messages [texera]
Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on code in PR #6522:
URL: https://github.com/apache/texera/pull/6522#discussion_r3608746669
##########
amber/src/main/python/core/models/internal_queue.py:
##########
@@ -74,6 +74,14 @@ def is_empty(self, key=None) -> bool:
def get(self) -> T:
return self._queue.get()
+ def peek(self) -> Optional[T]:
+ """
+ Peek the element that get() would return next, without removing it.
+
+ :return: Optional[T], the next available element, or None if empty.
+ """
+ return self._queue.peek()
Review Comment:
the encapsulation intention was to let the internal_queue not peakable (same as python Queue interface). Is it possible to solve this issue without exposing the underneath internal api?
--
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(pyamber): stop EndWorker from consuming straggler messages [texera]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #6522:
URL: https://github.com/apache/texera/pull/6522#issuecomment-5011081769
<!-- 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`, `@eugenegujing`
You can notify them by mentioning `@Yicong-Huang`, `@eugenegujing` 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(pyamber): stop EndWorker from consuming straggler messages [texera]
Posted by "Ma77Ball (via GitHub)" <gi...@apache.org>.
Ma77Ball commented on code in PR #6522:
URL: https://github.com/apache/texera/pull/6522#discussion_r3633773931
##########
amber/src/main/python/core/architecture/handlers/control/end_worker_handler.py:
##########
@@ -37,13 +37,19 @@ 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.
- input_queue: IQueue = self.context.input_queue
- if not input_queue.is_empty():
+ # 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.
+ input_queue: InternalQueue = self.context.input_queue
+ queued_count = input_queue.size()
Review Comment:
The PR description documents a peek()-based approach plus internal_queue.py and test_internal_queue.py changes, but this diff uses size() and touches only these two files; please reconcile the description with the actual diff.
--
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(pyamber): stop EndWorker from consuming straggler messages [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan commented on code in PR #6522:
URL: https://github.com/apache/texera/pull/6522#discussion_r3609352824
##########
amber/src/main/python/core/models/internal_queue.py:
##########
@@ -74,6 +74,14 @@ def is_empty(self, key=None) -> bool:
def get(self) -> T:
return self._queue.get()
+ def peek(self) -> Optional[T]:
+ """
+ Peek the element that get() would return next, without removing it.
+
+ :return: Optional[T], the next available element, or None if empty.
+ """
+ return self._queue.peek()
Review Comment:
Good point — dropped `peek()` entirely so `InternalQueue` stays non-peekable like `queue.Queue`. The handler doesn't actually need the next message's content: it now logs the pending count via the existing `len()`/`is_empty()` and raises, which is all that's needed to fail the RPC and let the coordinator retry. No internal API exposed.
--
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(pyamber): stop EndWorker from consuming straggler messages [texera]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #6522:
URL: https://github.com/apache/texera/pull/6522#issuecomment-5011090606
## [Codecov](https://app.codecov.io/gh/apache/texera/pull/6522?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 70.63%. Comparing base ([`39db110`](https://app.codecov.io/gh/apache/texera/commit/39db110e74b4bdcf0e4b0df961c25973e6bbb20c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`b352eb1`](https://app.codecov.io/gh/apache/texera/commit/b352eb101c27978d88afd91ed1e26677f907cfac?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #6522 +/- ##
============================================
+ Coverage 70.57% 70.63% +0.05%
Complexity 3425 3425
============================================
Files 1146 1146
Lines 45062 45064 +2
Branches 4960 4960
============================================
+ Hits 31802 31830 +28
+ Misses 11613 11587 -26
Partials 1647 1647
```
| [Flag](https://app.codecov.io/gh/apache/texera/pull/6522/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/6522/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 [39db110](https://app.codecov.io/gh/apache/texera/commit/39db110e74b4bdcf0e4b0df961c25973e6bbb20c?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/6522/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 [39db110](https://app.codecov.io/gh/apache/texera/commit/39db110e74b4bdcf0e4b0df961c25973e6bbb20c?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/6522/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `66.72% <ø> (ø)` | | Carriedforward from [39db110](https://app.codecov.io/gh/apache/texera/commit/39db110e74b4bdcf0e4b0df961c25973e6bbb20c?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/6522/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `17.72% <ø> (ø)` | | Carriedforward from [39db110](https://app.codecov.io/gh/apache/texera/commit/39db110e74b4bdcf0e4b0df961c25973e6bbb20c?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/6522/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `52.30% <ø> (ø)` | | Carriedforward from [39db110](https://app.codecov.io/gh/apache/texera/commit/39db110e74b4bdcf0e4b0df961c25973e6bbb20c?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/6522/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `66.80% <ø> (ø)` | | Carriedforward from [39db110](https://app.codecov.io/gh/apache/texera/commit/39db110e74b4bdcf0e4b0df961c25973e6bbb20c?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/6522/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `69.36% <ø> (ø)` | | Carriedforward from [39db110](https://app.codecov.io/gh/apache/texera/commit/39db110e74b4bdcf0e4b0df961c25973e6bbb20c?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/6522/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 [39db110](https://app.codecov.io/gh/apache/texera/commit/39db110e74b4bdcf0e4b0df961c25973e6bbb20c?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/6522/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `92.34% <100.00%> (+0.59%)` | :arrow_up: | |
| [workflow-compiling-service](https://app.codecov.io/gh/apache/texera/pull/6522/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 [39db110](https://app.codecov.io/gh/apache/texera/commit/39db110e74b4bdcf0e4b0df961c25973e6bbb20c?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/6522?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(pyamber): stop EndWorker from consuming straggler messages [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan merged PR #6522:
URL: https://github.com/apache/texera/pull/6522
--
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(pyamber): stop EndWorker from consuming straggler messages [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan commented on code in PR #6522:
URL: https://github.com/apache/texera/pull/6522#discussion_r3609353532
##########
amber/src/main/python/core/architecture/handlers/control/end_worker_handler.py:
##########
@@ -38,12 +38,15 @@ async def end_worker(self, req: EmptyRequest) -> EmptyReturn:
of all the control messages.
"""
# Ensure this is really the last message.
- input_queue: IQueue = self.context.input_queue
+ input_queue: InternalQueue = self.context.input_queue
if not input_queue.is_empty():
logger.warning(
f"Received EndHandler before all messages are "
- f"processed. Unprocessed messages: {input_queue.get()}"
+ f"processed. Unprocessed messages: {input_queue.peek()}"
)
Review Comment:
Done — the log now says `EndWorker`, and it reports the pending message count instead of dumping a 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(pyamber): stop EndWorker from consuming straggler messages [texera]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #6522:
URL: https://github.com/apache/texera/pull/6522#issuecomment-5011097501
<!-- texera-benchmarks-comment -->
## ⚠️ Benchmark changes need a look
🟢 4 better · 🔴 4 worse · ⚪ 7 noise (<±5%) · 0 without baseline
> Compared against main `39db110` 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/29642617336)
| | config | throughput | MB/s | latency | max Δ latest / 7d |
|---|---|---:|---:|---:|---:|
| 🔴 | bs=10 sw=10 sl=64 | 386 | 0.236 | 26,306/33,435/33,435 us | 🟢 -36.0% / 🔴 +132.8% |
| 🔴 | bs=100 sw=10 sl=64 | 905 | 0.552 | 108,398/150,082/150,082 us | 🟢 -15.3% / 🔴 +51.4% |
| ⚪ | bs=1000 sw=10 sl=64 | 1,092 | 0.667 | 903,451/1,001,249/1,001,249 us | ⚪ within ±5% / 🔴 +5.3% |
<details><summary>Baseline details</summary>
Latest main `39db110` from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---:|---:|---:|---:|---:|
| bs=10 sw=10 sl=64 | throughput | 386 tuples/sec | 413 tuples/sec | 831.47 tuples/sec | -6.5% | -53.6% |
| bs=10 sw=10 sl=64 | MB/s | 0.236 MB/s | 0.252 MB/s | 0.507 MB/s | -6.3% | -53.5% |
| bs=10 sw=10 sl=64 | p50 | 26,306 us | 19,510 us | 11,844 us | +34.8% | +122.1% |
| bs=10 sw=10 sl=64 | p95 | 33,435 us | 52,233 us | 14,359 us | -36.0% | +132.8% |
| bs=10 sw=10 sl=64 | p99 | 33,435 us | 52,233 us | 18,888 us | -36.0% | +77.0% |
| bs=100 sw=10 sl=64 | throughput | 905 tuples/sec | 923 tuples/sec | 1,089 tuples/sec | -2.0% | -16.9% |
| bs=100 sw=10 sl=64 | MB/s | 0.552 MB/s | 0.564 MB/s | 0.665 MB/s | -2.1% | -16.9% |
| bs=100 sw=10 sl=64 | p50 | 108,398 us | 103,095 us | 92,503 us | +5.1% | +17.2% |
| bs=100 sw=10 sl=64 | p95 | 150,082 us | 177,155 us | 99,121 us | -15.3% | +51.4% |
| bs=100 sw=10 sl=64 | p99 | 150,082 us | 177,155 us | 112,422 us | -15.3% | +33.5% |
| bs=1000 sw=10 sl=64 | throughput | 1,092 tuples/sec | 1,085 tuples/sec | 1,117 tuples/sec | +0.6% | -2.3% |
| bs=1000 sw=10 sl=64 | MB/s | 0.667 MB/s | 0.662 MB/s | 0.682 MB/s | +0.8% | -2.2% |
| bs=1000 sw=10 sl=64 | p50 | 903,451 us | 930,169 us | 910,046 us | -2.9% | -0.7% |
| bs=1000 sw=10 sl=64 | p95 | 1,001,249 us | 957,264 us | 950,716 us | +4.6% | +5.3% |
| bs=1000 sw=10 sl=64 | p99 | 1,001,249 us | 957,264 us | 985,832 us | +4.6% | +1.6% |
</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,518.11,200,128000,386,0.236,26305.75,33434.56,33434.56
1,100,10,64,20,2210.81,2000,1280000,905,0.552,108398.01,150081.54,150081.54
2,1000,10,64,20,18307.29,20000,12800000,1092,0.667,903450.94,1001249.04,1001249.04
```
</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(pyamber): stop EndWorker from consuming straggler messages [texera]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #6522:
URL: https://github.com/apache/texera/pull/6522#discussion_r3610248914
##########
amber/src/main/python/core/architecture/handlers/control/end_worker_handler.py:
##########
@@ -41,9 +41,13 @@ async def end_worker(self, req: EmptyRequest) -> EmptyReturn:
input_queue: IQueue = self.context.input_queue
if not input_queue.is_empty():
logger.warning(
- f"Received EndHandler before all messages are "
- f"processed. Unprocessed messages: {input_queue.get()}"
+ f"Received EndWorker before all {len(input_queue)} queued "
+ f"message(s) were processed; failing the RPC so the coordinator "
Review Comment:
`input_queue` is annotated as `IQueue`, but this interface doesn't define `__len__`, so `len(input_queue)` is inconsistent with the declared type and can trip static analysis / future alternate queue implementations. Also, you can avoid the double-check (`is_empty()` + `len(...)`) by computing the queued count once and branching on 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(pyamber): stop EndWorker from consuming straggler messages [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan commented on code in PR #6522:
URL: https://github.com/apache/texera/pull/6522#discussion_r3643188568
##########
amber/src/main/python/core/architecture/handlers/control/end_worker_handler.py:
##########
@@ -37,13 +37,19 @@ 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.
- input_queue: IQueue = self.context.input_queue
- if not input_queue.is_empty():
+ # 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.
+ input_queue: InternalQueue = self.context.input_queue
+ queued_count = input_queue.size()
Review Comment:
Good catch — the description was stale from the original `peek()`-based version. Updated it to match the current diff: the guard reads the pending count once via `size()` and raises (no `peek()`, so `internal_queue.py` / `test_internal_queue.py` are untouched), and the PR now changes only `end_worker_handler.py` plus its new test.
--
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(pyamber): stop EndWorker from consuming straggler messages [texera]
Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on code in PR #6522:
URL: https://github.com/apache/texera/pull/6522#discussion_r3649537698
##########
amber/src/main/python/core/architecture/handlers/control/end_worker_handler.py:
##########
@@ -37,13 +37,19 @@ 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.
- input_queue: IQueue = self.context.input_queue
- if not input_queue.is_empty():
+ # 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.
+ input_queue: InternalQueue = self.context.input_queue
+ queued_count = input_queue.size()
+ if queued_count > 0:
logger.warning(
- f"Received EndHandler before all messages are "
- f"processed. Unprocessed messages: {input_queue.get()}"
+ f"Received EndWorker before all {queued_count} queued "
+ f"message(s) were processed; failing the RPC so the coordinator "
+ f"retries once the queue has drained."
Review Comment:
The coordinator retries on a fixed delay (`killRetryDelay`, bounded by `maxTerminationAttempts`) — it doesn't watch the queue, so "retries once the queue has drained" can mislead teardown triage into thinking it waits indefinitely. The comment below uses the same phrasing.
```suggestion
f"message(s) were processed; failing the RPC so a later "
f"coordinator retry succeeds once the queue has drained."
```
##########
amber/src/test/python/core/architecture/handlers/control/test_end_worker_handler.py:
##########
@@ -0,0 +1,97 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import asyncio
+from types import SimpleNamespace
+
+import pytest
+
+from core.architecture.handlers.control.end_worker_handler import EndWorkerHandler
+from core.models.internal_queue import DCMElement, InternalQueue
+from proto.org.apache.texera.amber.core import ActorVirtualIdentity, ChannelIdentity
+from proto.org.apache.texera.amber.engine.architecture.rpc import (
+ EmptyRequest,
+ EmptyReturn,
+)
+from proto.org.apache.texera.amber.engine.common import DirectControlMessagePayloadV2
+
+
+class TestEndWorkerHandler:
+ @pytest.fixture
+ def input_queue(self):
+ return InternalQueue()
+
+ @pytest.fixture
+ def handler(self, input_queue):
+ # ControlHandler.__init__ just stashes context; bypass the protobuf
+ # base class' __init__ by constructing via __new__.
+ instance = EndWorkerHandler.__new__(EndWorkerHandler)
+ instance.context = SimpleNamespace(input_queue=input_queue)
+ return instance
Review Comment:
`ControlHandler.__init__` only stashes `context` and never calls `super().__init__()`, so normal construction runs no protobuf-base `__init__` either — there is nothing to bypass. `test_initialize_executor_handler.py` builds its handler this way.
```suggestion
return EndWorkerHandler(SimpleNamespace(input_queue=input_queue))
```
--
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(pyamber): stop EndWorker from consuming straggler messages [texera]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #6522:
URL: https://github.com/apache/texera/pull/6522#discussion_r3608258909
##########
amber/src/main/python/core/architecture/handlers/control/end_worker_handler.py:
##########
@@ -38,12 +38,15 @@ async def end_worker(self, req: EmptyRequest) -> EmptyReturn:
of all the control messages.
"""
# Ensure this is really the last message.
- input_queue: IQueue = self.context.input_queue
+ input_queue: InternalQueue = self.context.input_queue
if not input_queue.is_empty():
logger.warning(
f"Received EndHandler before all messages are "
- f"processed. Unprocessed messages: {input_queue.get()}"
+ f"processed. Unprocessed messages: {input_queue.peek()}"
)
Review Comment:
The warning message says "Received EndHandler" but this is the Python EndWorker handler. This makes logs harder to search/triage and can be confused with the Scala EndHandler; update the log text to reference EndWorker (or EndWorkerHandler).
--
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(pyamber): stop EndWorker from consuming straggler messages [texera]
Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on code in PR #6522:
URL: https://github.com/apache/texera/pull/6522#discussion_r3608747903
##########
amber/src/main/python/core/architecture/handlers/control/end_worker_handler.py:
##########
@@ -38,12 +38,15 @@ async def end_worker(self, req: EmptyRequest) -> EmptyReturn:
of all the control messages.
"""
# Ensure this is really the last message.
- input_queue: IQueue = self.context.input_queue
+ input_queue: InternalQueue = self.context.input_queue
Review Comment:
can we keep IQueue? There are APIs in InternalQueue that not to be exposed..
--
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(pyamber): stop EndWorker from consuming straggler messages [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan commented on code in PR #6522:
URL: https://github.com/apache/texera/pull/6522#discussion_r3609353184
##########
amber/src/main/python/core/architecture/handlers/control/end_worker_handler.py:
##########
@@ -38,12 +38,15 @@ async def end_worker(self, req: EmptyRequest) -> EmptyReturn:
of all the control messages.
"""
# Ensure this is really the last message.
- input_queue: IQueue = self.context.input_queue
+ input_queue: InternalQueue = self.context.input_queue
Review Comment:
Reverted to `IQueue` — the handler only calls `is_empty()` and `len()`, both already on the `IQueue`/`EmtpyCheckable` interface, so nothing `InternalQueue`-specific leaks in here.
--
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(pyamber): stop EndWorker from consuming straggler messages [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan commented on code in PR #6522:
URL: https://github.com/apache/texera/pull/6522#discussion_r3612295352
##########
amber/src/main/python/core/architecture/handlers/control/end_worker_handler.py:
##########
@@ -41,9 +41,13 @@ async def end_worker(self, req: EmptyRequest) -> EmptyReturn:
input_queue: IQueue = self.context.input_queue
if not input_queue.is_empty():
logger.warning(
- f"Received EndHandler before all messages are "
- f"processed. Unprocessed messages: {input_queue.get()}"
+ f"Received EndWorker before all {len(input_queue)} queued "
+ f"message(s) were processed; failing the RPC so the coordinator "
Review Comment:
Good point — typed the local as the actual InternalQueue (which exposes size(); the base IQueue interface does not), and read queued_count = size() once, branching on it instead of the is_empty()+len() double-check.
--
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