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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
