You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "gortiz (via GitHub)" <gi...@apache.org> on 2025/09/29 08:15:20 UTC
[PR] Remove SendingMailbox.complete method and add that logic when EOS is sent [pinot]
gortiz opened a new pull request, #16915:
URL: https://github.com/apache/pinot/pull/16915
The current API for SendingMailbox is designed to call `send(Data)` several times, followed by a call to `send(EOS)` and finally a call to `complete()`. At any time, `cancel` can be called to report an uncontrolled error.
Having to call `complete` after `send(EOS)` makes no sense and it is error-prone. For example, on https://github.com/apache/pinot/pull/16899 we detected that we were not calling `complete` when using pipeline breaker.
This PR changes the code to remove the `complete()` method. Instead, the code executed by that former method is added to `send(EOS)`.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Remove SendingMailbox.complete method and add that logic when EOS is sent [pinot]
Posted by "gortiz (via GitHub)" <gi...@apache.org>.
gortiz commented on code in PR #16915:
URL: https://github.com/apache/pinot/pull/16915#discussion_r2402316960
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/GrpcSendingMailbox.java:
##########
@@ -98,6 +98,8 @@ public void send(MseBlock.Data data)
public void send(MseBlock.Eos block, List<DataBuffer> serializedStats)
throws IOException, TimeoutException {
sendInternal(block, serializedStats);
+ LOGGER.debug("Completing mailbox: {}", _id);
+ _contentObserver.onCompleted();
Review Comment:
Notice that we need to call onCompleted even when an error is received.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Remove SendingMailbox.complete method and add that logic when EOS is sent [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #16915:
URL: https://github.com/apache/pinot/pull/16915#discussion_r2387085202
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/exchange/BlockExchange.java:
##########
@@ -139,16 +139,28 @@ public boolean send(MseBlock.Eos eosBlock, List<DataBuffer> serializedStats)
// this may happen when the block exchange is itself used as a sending mailbox, like when using spools
mailboxIdToSendMetadata = -1;
}
+ RuntimeException firstException = null;
for (int i = 0; i < numMailboxes; i++) {
- SendingMailbox sendingMailbox = _sendingMailboxes.get(i);
- List<DataBuffer> statsToSend = i == mailboxIdToSendMetadata ? serializedStats : Collections.emptyList();
+ try {
+ SendingMailbox sendingMailbox = _sendingMailboxes.get(i);
+ List<DataBuffer> statsToSend = i == mailboxIdToSendMetadata ? serializedStats : Collections.emptyList();
- sendingMailbox.send(eosBlock, statsToSend);
- sendingMailbox.complete();
- if (LOGGER.isTraceEnabled()) {
- LOGGER.trace("Block sent: {} {} to {}", eosBlock, System.identityHashCode(eosBlock), sendingMailbox);
+ sendingMailbox.send(eosBlock, statsToSend);
+ if (LOGGER.isTraceEnabled()) {
+ LOGGER.trace("Block sent: {} {} to {}", eosBlock, System.identityHashCode(eosBlock), sendingMailbox);
+ }
+ } catch (IOException | TimeoutException e) {
+ // We want to try to send EOS to all mailboxes, so we catch the exception and rethrow it at the end.
+ if (firstException == null) {
+ firstException = new RuntimeException("Failed to send EOS block to mailbox #" + i, e);
+ } else {
+ firstException.addSuppressed(e);
+ }
}
}
+ if (firstException != null) {
+ throw firstException;
+ }
Review Comment:
[nitpick] The exception handling logic could be extracted into a helper method to reduce complexity in the main method. Consider creating a `sendEosToAllMailboxes` method that encapsulates the loop and exception aggregation logic.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/SendingMailbox.java:
##########
@@ -28,6 +28,16 @@
/**
* Mailbox that's used to send data.
+ *
+ * Usages of this interface should follow the pattern:
+ *
+ * <ol>
+ * <li>Zero or more calls to {@link #send(MseBlock.Data)}</li>
+ * <ul>
+ * <li>And exactly one call to {@link #send(MseBlock.Eos, List)} if the receiver is not early terminated</li>
+ * <li>Or exactly one call to {@link #cancel(Throwable)} if the sender wants to cancel the receiver</li>
+ * </ul>
Review Comment:
The nested list structure is incorrect HTML. The `<ul>` should contain `<li>` elements, not be nested inside another `<li>`. Consider restructuring as: `<li>Then exactly one of: <ul><li>One call to {@link #send(MseBlock.Eos, List)} if the receiver is not early terminated</li><li>One call to {@link #cancel(Throwable)} if the sender wants to cancel the receiver</li></ul></li>`
```suggestion
* <li>Then exactly one of:
* <ul>
* <li>One call to {@link #send(MseBlock.Eos, List)} if the receiver is not early terminated</li>
* <li>One call to {@link #cancel(Throwable)} if the sender wants to cancel the receiver</li>
* </ul>
* </li>
```
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Remove SendingMailbox.complete method and add that logic when EOS is sent [pinot]
Posted by "gortiz (via GitHub)" <gi...@apache.org>.
gortiz commented on code in PR #16915:
URL: https://github.com/apache/pinot/pull/16915#discussion_r2409819958
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/exchange/BlockExchange.java:
##########
@@ -237,14 +249,10 @@ private void sendPrivate(MseBlock block, List<DataBuffer> serializedStats)
_earlyTerminated = BlockExchange.this.send(((MseBlock.Data) block));
} else {
_earlyTerminated = BlockExchange.this.send(((MseBlock.Eos) block), serializedStats);
+ _completed = true;
Review Comment:
We can also remove this method.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Remove SendingMailbox.complete method and add that logic when EOS is sent [pinot]
Posted by "gortiz (via GitHub)" <gi...@apache.org>.
gortiz commented on code in PR #16915:
URL: https://github.com/apache/pinot/pull/16915#discussion_r2402283879
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/GrpcSendingMailbox.java:
##########
@@ -97,7 +97,12 @@ public void send(MseBlock.Data data)
@Override
public void send(MseBlock.Eos block, List<DataBuffer> serializedStats)
throws IOException, TimeoutException {
- sendInternal(block, serializedStats);
+ try {
+ sendInternal(block, serializedStats);
+ } finally {
+ LOGGER.debug("Completing mailbox: {}", _id);
+ _contentObserver.onCompleted();
+ }
Review Comment:
Notice that we need to call onCompleted even when an error is received
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Remove SendingMailbox.complete method and add that logic when EOS is sent [pinot]
Posted by "gortiz (via GitHub)" <gi...@apache.org>.
gortiz commented on code in PR #16915:
URL: https://github.com/apache/pinot/pull/16915#discussion_r2409804119
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/InMemorySendingMailbox.java:
##########
@@ -68,7 +68,11 @@ public void send(MseBlock.Data data)
@Override
public void send(MseBlock.Eos block, List<DataBuffer> serializedStats)
throws IOException, TimeoutException {
- sendPrivate(block, serializedStats);
+ try {
+ sendPrivate(block, serializedStats);
+ } finally {
+ _isTerminated = true;
Review Comment:
Good catch. I'll change 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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Remove SendingMailbox.complete method and add that logic when EOS is sent [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #16915:
URL: https://github.com/apache/pinot/pull/16915#issuecomment-3345810201
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/16915?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `55.55556%` with `8 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 56.41%. Comparing base ([`9baf132`](https://app.codecov.io/gh/apache/pinot/commit/9baf132185cdccea83acb6fef453530bd4edb4ef?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`91c1c35`](https://app.codecov.io/gh/apache/pinot/commit/91c1c35a5ee3ff85ea82be653a8d5d4e8628dbc9?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/pinot/pull/16915?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...query/runtime/operator/exchange/BlockExchange.java](https://app.codecov.io/gh/apache/pinot/pull/16915?src=pr&el=tree&filepath=pinot-query-runtime%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fquery%2Fruntime%2Foperator%2Fexchange%2FBlockExchange.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtcXVlcnktcnVudGltZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcXVlcnkvcnVudGltZS9vcGVyYXRvci9leGNoYW5nZS9CbG9ja0V4Y2hhbmdlLmphdmE=) | 38.46% | [6 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/pinot/pull/16915?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
> :exclamation: There is a different number of reports uploaded between BASE (9baf132) and HEAD (91c1c35). Click for more details.
>
> <details><summary>HEAD has 32 uploads less than BASE</summary>
>
>| Flag | BASE (9baf132) | HEAD (91c1c35) |
>|------|------|------|
>|java-21|5|1|
>|unittests1|2|1|
>|unittests|4|1|
>|temurin|9|1|
>|java-11|4|0|
>|unittests2|2|0|
>|integration|5|0|
>|integration1|2|0|
>|custom-integration1|2|0|
>|integration2|1|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #16915 +/- ##
============================================
- Coverage 63.51% 56.41% -7.10%
+ Complexity 1411 701 -710
============================================
Files 3073 2407 -666
Lines 180742 133889 -46853
Branches 27628 21276 -6352
============================================
- Hits 114802 75540 -39262
+ Misses 57084 52056 -5028
+ Partials 8856 6293 -2563
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/16915/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/16915/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration](https://app.codecov.io/gh/apache/pinot/pull/16915/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/16915/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/16915/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-11](https://app.codecov.io/gh/apache/pinot/pull/16915/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-21](https://app.codecov.io/gh/apache/pinot/pull/16915/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.41% <55.55%> (-7.07%)` | :arrow_down: |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/16915/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.41% <55.55%> (-7.10%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/16915/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.41% <55.55%> (-7.10%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/16915/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.41% <55.55%> (-0.03%)` | :arrow_down: |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/16915/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
Flags with carried forward coverage won't be shown. [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#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/pinot/pull/16915?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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Remove SendingMailbox.complete method and add that logic when EOS is sent [pinot]
Posted by "gortiz (via GitHub)" <gi...@apache.org>.
gortiz merged PR #16915:
URL: https://github.com/apache/pinot/pull/16915
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Remove SendingMailbox.complete method and add that logic when EOS is sent [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #16915:
URL: https://github.com/apache/pinot/pull/16915#discussion_r2408976983
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/exchange/BlockExchange.java:
##########
@@ -237,14 +249,10 @@ private void sendPrivate(MseBlock block, List<DataBuffer> serializedStats)
_earlyTerminated = BlockExchange.this.send(((MseBlock.Data) block));
} else {
_earlyTerminated = BlockExchange.this.send(((MseBlock.Eos) block), serializedStats);
+ _completed = true;
Review Comment:
Should we set it in `public void send(MseBlock.Eos block, List<DataBuffer> serializedStats)`?
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/InMemorySendingMailbox.java:
##########
@@ -68,7 +68,11 @@ public void send(MseBlock.Data data)
@Override
public void send(MseBlock.Eos block, List<DataBuffer> serializedStats)
throws IOException, TimeoutException {
- sendPrivate(block, serializedStats);
+ try {
+ sendPrivate(block, serializedStats);
+ } finally {
+ _isTerminated = true;
Review Comment:
Should we put this in the `finally` block? This could prevent error block being set in `cancel()` when `sendPrivate()` throws exception
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/exchange/BlockExchange.java:
##########
@@ -139,16 +138,29 @@ public boolean send(MseBlock.Eos eosBlock, List<DataBuffer> serializedStats)
// this may happen when the block exchange is itself used as a sending mailbox, like when using spools
mailboxIdToSendMetadata = -1;
}
+ RuntimeException firstException = null;
+ int numMailboxes = _sendingMailboxes.size();
for (int i = 0; i < numMailboxes; i++) {
- SendingMailbox sendingMailbox = _sendingMailboxes.get(i);
- List<DataBuffer> statsToSend = i == mailboxIdToSendMetadata ? serializedStats : Collections.emptyList();
+ try {
+ SendingMailbox sendingMailbox = _sendingMailboxes.get(i);
+ List<DataBuffer> statsToSend = i == mailboxIdToSendMetadata ? serializedStats : Collections.emptyList();
- sendingMailbox.send(eosBlock, statsToSend);
- sendingMailbox.complete();
- if (LOGGER.isTraceEnabled()) {
- LOGGER.trace("Block sent: {} {} to {}", eosBlock, System.identityHashCode(eosBlock), sendingMailbox);
+ sendingMailbox.send(eosBlock, statsToSend);
+ if (LOGGER.isTraceEnabled()) {
+ LOGGER.trace("Block sent: {} {} to {}", eosBlock, System.identityHashCode(eosBlock), sendingMailbox);
+ }
+ } catch (IOException | TimeoutException | RuntimeException e) {
Review Comment:
Is this the same as `catch(Exception e)`?
##########
pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/exchange/BlockExchangeTest.java:
##########
@@ -76,11 +76,11 @@ public void shouldSendEosBlockToAllDestinations()
// Then:
ArgumentCaptor<MseBlock.Eos> captor = ArgumentCaptor.forClass(MseBlock.Eos.class);
- Mockito.verify(_mailbox1).complete();
+ Mockito.verify(_mailbox1).send(SuccessMseBlock.INSTANCE, Collections.emptyList());
Review Comment:
(nit) Let's use `List.of()`
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/exchange/BlockExchange.java:
##########
@@ -139,16 +138,29 @@ public boolean send(MseBlock.Eos eosBlock, List<DataBuffer> serializedStats)
// this may happen when the block exchange is itself used as a sending mailbox, like when using spools
mailboxIdToSendMetadata = -1;
}
+ RuntimeException firstException = null;
+ int numMailboxes = _sendingMailboxes.size();
for (int i = 0; i < numMailboxes; i++) {
- SendingMailbox sendingMailbox = _sendingMailboxes.get(i);
- List<DataBuffer> statsToSend = i == mailboxIdToSendMetadata ? serializedStats : Collections.emptyList();
+ try {
+ SendingMailbox sendingMailbox = _sendingMailboxes.get(i);
+ List<DataBuffer> statsToSend = i == mailboxIdToSendMetadata ? serializedStats : Collections.emptyList();
- sendingMailbox.send(eosBlock, statsToSend);
- sendingMailbox.complete();
- if (LOGGER.isTraceEnabled()) {
- LOGGER.trace("Block sent: {} {} to {}", eosBlock, System.identityHashCode(eosBlock), sendingMailbox);
+ sendingMailbox.send(eosBlock, statsToSend);
+ if (LOGGER.isTraceEnabled()) {
+ LOGGER.trace("Block sent: {} {} to {}", eosBlock, System.identityHashCode(eosBlock), sendingMailbox);
+ }
+ } catch (IOException | TimeoutException | RuntimeException e) {
+ // We want to try to send EOS to all mailboxes, so we catch the exception and rethrow it at the end.
+ if (firstException == null) {
+ firstException = new RuntimeException("Failed to send EOS block to mailbox #" + i, e);
Review Comment:
This prevents checked exceptions being thrown. We check exception type in `QueryRunner` and handle them differently
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Remove SendingMailbox.complete method and add that logic when EOS is sent [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #16915:
URL: https://github.com/apache/pinot/pull/16915#discussion_r2402296568
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/exchange/BlockExchange.java:
##########
@@ -139,16 +139,28 @@ public boolean send(MseBlock.Eos eosBlock, List<DataBuffer> serializedStats)
// this may happen when the block exchange is itself used as a sending mailbox, like when using spools
mailboxIdToSendMetadata = -1;
}
+ RuntimeException firstException = null;
for (int i = 0; i < numMailboxes; i++) {
- SendingMailbox sendingMailbox = _sendingMailboxes.get(i);
- List<DataBuffer> statsToSend = i == mailboxIdToSendMetadata ? serializedStats : Collections.emptyList();
+ try {
+ SendingMailbox sendingMailbox = _sendingMailboxes.get(i);
+ List<DataBuffer> statsToSend = i == mailboxIdToSendMetadata ? serializedStats : Collections.emptyList();
- sendingMailbox.send(eosBlock, statsToSend);
- sendingMailbox.complete();
- if (LOGGER.isTraceEnabled()) {
- LOGGER.trace("Block sent: {} {} to {}", eosBlock, System.identityHashCode(eosBlock), sendingMailbox);
+ sendingMailbox.send(eosBlock, statsToSend);
+ if (LOGGER.isTraceEnabled()) {
+ LOGGER.trace("Block sent: {} {} to {}", eosBlock, System.identityHashCode(eosBlock), sendingMailbox);
+ }
+ } catch (IOException | TimeoutException e) {
+ // We want to try to send EOS to all mailboxes, so we catch the exception and rethrow it at the end.
+ if (firstException == null) {
+ firstException = new RuntimeException("Failed to send EOS block to mailbox #" + i, e);
+ } else {
+ firstException.addSuppressed(e);
+ }
}
Review Comment:
The exception handling only catches `IOException` and `TimeoutException`, but the `send()` method can potentially throw `RuntimeException` (as seen in the new completion logic). Any uncaught `RuntimeException` from individual mailbox sends will prevent EOS from being sent to remaining mailboxes, which violates the stated goal of trying to send EOS to all mailboxes.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/GrpcSendingMailbox.java:
##########
@@ -97,7 +97,12 @@ public void send(MseBlock.Data data)
@Override
public void send(MseBlock.Eos block, List<DataBuffer> serializedStats)
throws IOException, TimeoutException {
- sendInternal(block, serializedStats);
+ try {
+ sendInternal(block, serializedStats);
+ } finally {
+ LOGGER.debug("Completing mailbox: {}", _id);
+ _contentObserver.onCompleted();
+ }
Review Comment:
The `_contentObserver.onCompleted()` will be called even if `sendInternal()` throws an exception. This could lead to calling `onCompleted()` on a gRPC observer that may already be in an error state, potentially causing undefined behavior. Consider checking if the observer is still in a valid state before calling `onCompleted()`, or move the completion logic to execute only on successful send.
```suggestion
sendInternal(block, serializedStats);
LOGGER.debug("Completing mailbox: {}", _id);
_contentObserver.onCompleted();
```
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Remove SendingMailbox.complete method and add that logic when EOS is sent [pinot]
Posted by "gortiz (via GitHub)" <gi...@apache.org>.
gortiz commented on code in PR #16915:
URL: https://github.com/apache/pinot/pull/16915#discussion_r2414330929
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/exchange/BlockExchange.java:
##########
@@ -139,14 +138,36 @@ public boolean send(MseBlock.Eos eosBlock, List<DataBuffer> serializedStats)
// this may happen when the block exchange is itself used as a sending mailbox, like when using spools
mailboxIdToSendMetadata = -1;
}
+ Exception firstException = null;
+ int numMailboxes = _sendingMailboxes.size();
for (int i = 0; i < numMailboxes; i++) {
- SendingMailbox sendingMailbox = _sendingMailboxes.get(i);
- List<DataBuffer> statsToSend = i == mailboxIdToSendMetadata ? serializedStats : Collections.emptyList();
+ try {
+ SendingMailbox sendingMailbox = _sendingMailboxes.get(i);
+ List<DataBuffer> statsToSend = i == mailboxIdToSendMetadata ? serializedStats : Collections.emptyList();
- sendingMailbox.send(eosBlock, statsToSend);
- sendingMailbox.complete();
- if (LOGGER.isTraceEnabled()) {
- LOGGER.trace("Block sent: {} {} to {}", eosBlock, System.identityHashCode(eosBlock), sendingMailbox);
+ sendingMailbox.send(eosBlock, statsToSend);
+ if (LOGGER.isTraceEnabled()) {
+ LOGGER.trace("Block sent: {} {} to {}", eosBlock, System.identityHashCode(eosBlock), sendingMailbox);
+ }
+ } catch (IOException | TimeoutException | RuntimeException e) {
+ // We want to try to send EOS to all mailboxes, so we catch the exception and rethrow it at the end.
+ if (firstException == null) {
+ firstException = e;
+ } else {
+ firstException.addSuppressed(e);
+ }
+ }
+ }
+ if (firstException != null) {
+ // This is ugly, but necessary to be sure we throw the right exception, which is later caught by the
Review Comment:
No, that would produce a compilation error because `firstException` static type is Exception, and we don't declare that in the throws section.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Remove SendingMailbox.complete method and add that logic when EOS is sent [pinot]
Posted by "gortiz (via GitHub)" <gi...@apache.org>.
gortiz commented on code in PR #16915:
URL: https://github.com/apache/pinot/pull/16915#discussion_r2409844037
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/exchange/BlockExchange.java:
##########
@@ -139,16 +138,29 @@ public boolean send(MseBlock.Eos eosBlock, List<DataBuffer> serializedStats)
// this may happen when the block exchange is itself used as a sending mailbox, like when using spools
mailboxIdToSendMetadata = -1;
}
+ RuntimeException firstException = null;
+ int numMailboxes = _sendingMailboxes.size();
for (int i = 0; i < numMailboxes; i++) {
- SendingMailbox sendingMailbox = _sendingMailboxes.get(i);
- List<DataBuffer> statsToSend = i == mailboxIdToSendMetadata ? serializedStats : Collections.emptyList();
+ try {
+ SendingMailbox sendingMailbox = _sendingMailboxes.get(i);
+ List<DataBuffer> statsToSend = i == mailboxIdToSendMetadata ? serializedStats : Collections.emptyList();
- sendingMailbox.send(eosBlock, statsToSend);
- sendingMailbox.complete();
- if (LOGGER.isTraceEnabled()) {
- LOGGER.trace("Block sent: {} {} to {}", eosBlock, System.identityHashCode(eosBlock), sendingMailbox);
+ sendingMailbox.send(eosBlock, statsToSend);
+ if (LOGGER.isTraceEnabled()) {
+ LOGGER.trace("Block sent: {} {} to {}", eosBlock, System.identityHashCode(eosBlock), sendingMailbox);
+ }
+ } catch (IOException | TimeoutException | RuntimeException e) {
+ // We want to try to send EOS to all mailboxes, so we catch the exception and rethrow it at the end.
+ if (firstException == null) {
+ firstException = new RuntimeException("Failed to send EOS block to mailbox #" + i, e);
Review Comment:
Changed. The code is ugly now, but it should be correct.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/exchange/BlockExchange.java:
##########
@@ -139,16 +138,29 @@ public boolean send(MseBlock.Eos eosBlock, List<DataBuffer> serializedStats)
// this may happen when the block exchange is itself used as a sending mailbox, like when using spools
mailboxIdToSendMetadata = -1;
}
+ RuntimeException firstException = null;
+ int numMailboxes = _sendingMailboxes.size();
for (int i = 0; i < numMailboxes; i++) {
- SendingMailbox sendingMailbox = _sendingMailboxes.get(i);
- List<DataBuffer> statsToSend = i == mailboxIdToSendMetadata ? serializedStats : Collections.emptyList();
+ try {
+ SendingMailbox sendingMailbox = _sendingMailboxes.get(i);
+ List<DataBuffer> statsToSend = i == mailboxIdToSendMetadata ? serializedStats : Collections.emptyList();
- sendingMailbox.send(eosBlock, statsToSend);
- sendingMailbox.complete();
- if (LOGGER.isTraceEnabled()) {
- LOGGER.trace("Block sent: {} {} to {}", eosBlock, System.identityHashCode(eosBlock), sendingMailbox);
+ sendingMailbox.send(eosBlock, statsToSend);
+ if (LOGGER.isTraceEnabled()) {
+ LOGGER.trace("Block sent: {} {} to {}", eosBlock, System.identityHashCode(eosBlock), sendingMailbox);
+ }
+ } catch (IOException | TimeoutException | RuntimeException e) {
Review Comment:
Right now it is the same, but it is more specific. If in the future we throw more (or fewer) exceptions, the compiler will fail, which will force us to verify this code to be sure the newly added exception has to be treated this way
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Remove SendingMailbox.complete method and add that logic when EOS is sent [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #16915:
URL: https://github.com/apache/pinot/pull/16915#discussion_r2411999288
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/exchange/BlockExchange.java:
##########
@@ -139,14 +138,36 @@ public boolean send(MseBlock.Eos eosBlock, List<DataBuffer> serializedStats)
// this may happen when the block exchange is itself used as a sending mailbox, like when using spools
mailboxIdToSendMetadata = -1;
}
+ Exception firstException = null;
+ int numMailboxes = _sendingMailboxes.size();
for (int i = 0; i < numMailboxes; i++) {
- SendingMailbox sendingMailbox = _sendingMailboxes.get(i);
- List<DataBuffer> statsToSend = i == mailboxIdToSendMetadata ? serializedStats : Collections.emptyList();
+ try {
+ SendingMailbox sendingMailbox = _sendingMailboxes.get(i);
+ List<DataBuffer> statsToSend = i == mailboxIdToSendMetadata ? serializedStats : Collections.emptyList();
- sendingMailbox.send(eosBlock, statsToSend);
- sendingMailbox.complete();
- if (LOGGER.isTraceEnabled()) {
- LOGGER.trace("Block sent: {} {} to {}", eosBlock, System.identityHashCode(eosBlock), sendingMailbox);
+ sendingMailbox.send(eosBlock, statsToSend);
+ if (LOGGER.isTraceEnabled()) {
+ LOGGER.trace("Block sent: {} {} to {}", eosBlock, System.identityHashCode(eosBlock), sendingMailbox);
+ }
+ } catch (IOException | TimeoutException | RuntimeException e) {
+ // We want to try to send EOS to all mailboxes, so we catch the exception and rethrow it at the end.
+ if (firstException == null) {
+ firstException = e;
+ } else {
+ firstException.addSuppressed(e);
+ }
+ }
+ }
+ if (firstException != null) {
+ // This is ugly, but necessary to be sure we throw the right exception, which is later caught by the
Review Comment:
Does it work if we simply do `throw firstException;`?
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org