This is an automated email from the ASF dual-hosted git repository.
gortiz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new a8476f988cf Stop a lingering receiving mailbox from pinning the
receive operator (#19465)
a8476f988cf is described below
commit a8476f988cfd823b36e0069f6d00f3bfd391a531
Author: Yash Mayya <[email protected]>
AuthorDate: Mon Sep 7 06:47:25 2026 -0400
Stop a lingering receiving mailbox from pinning the receive operator
(#19465)
---
.../operator/BaseMailboxReceiveOperator.java | 21 +++++++----
.../operator/MailboxReceiveOperatorTest.java | 41 ++++++++++++++++++++++
2 files changed, 56 insertions(+), 6 deletions(-)
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/BaseMailboxReceiveOperator.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/BaseMailboxReceiveOperator.java
index 4e19b8a3a47..da5afb98b9e 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/BaseMailboxReceiveOperator.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/BaseMailboxReceiveOperator.java
@@ -73,7 +73,7 @@ public abstract class BaseMailboxReceiveOperator extends
MultiStageOperator {
for (String mailboxId : _mailboxIds) {
ReceivingMailbox receivingMailbox =
_mailboxService.getReceivingMailbox(mailboxId);
receivingMailbox.registerReceiveOperatorThreadContext(QueryThreadContext.getIfAvailable());
- ReadMailboxAsyncStream asyncStream = new
ReadMailboxAsyncStream(receivingMailbox, this);
+ ReadMailboxAsyncStream asyncStream = new
ReadMailboxAsyncStream(receivingMailbox, _mailboxService);
asyncStreams.add(asyncStream);
_receivingStats.add(asyncStream._mailbox.getStatMap());
}
@@ -164,13 +164,20 @@ public abstract class BaseMailboxReceiveOperator extends
MultiStageOperator {
_statMap.merge(StatKey.UPSTREAM_WAIT_MS,
from.getLong(ReceivingMailbox.StatKey.WAIT_CPU_TIME_MS));
}
+ /// Adapts a [ReceivingMailbox] to the [AsyncStream] the
[BlockingMultiStreamConsumer] reads from.
+ ///
+ /// Deliberately holds the [MailboxService] and not the operator. The
mailbox keeps this stream reachable through
+ /// the reader callback registered on it, and after a failed query the
mailbox itself stays in the service's cache
+ /// until it expires (see [#poll()] for why it is not released earlier). A
reference back to the operator from here
+ /// would keep the operator, its [OpChainExecutionContext] and everything
reachable from them alive for that whole
+ /// time.
private static class ReadMailboxAsyncStream implements
AsyncStream<ReceivingMailbox.MseBlockWithStats> {
final ReceivingMailbox _mailbox;
- final BaseMailboxReceiveOperator _operator;
+ final MailboxService _mailboxService;
- ReadMailboxAsyncStream(ReceivingMailbox mailbox,
BaseMailboxReceiveOperator operator) {
+ ReadMailboxAsyncStream(ReceivingMailbox mailbox, MailboxService
mailboxService) {
_mailbox = mailbox;
- _operator = operator;
+ _mailboxService = mailboxService;
}
@Override
@@ -186,9 +193,11 @@ public abstract class BaseMailboxReceiveOperator extends
MultiStageOperator {
if (blockWithStats != null) {
MseBlock block = blockWithStats.getBlock();
- // TODO: Check if we should also release mailbox on not successful EOS.
+ // Only a successful EOS releases the mailbox from the service's
cache. After an error the senders may still
+ // be running, and they have to find the cancelled mailbox rather than
recreate a fresh one that nobody reads.
+ // The cache expiry takes care of it, and this stream holds nothing
that makes that wait expensive.
if (block.isSuccess()) {
- _operator._mailboxService.releaseReceivingMailbox(_mailbox);
+ _mailboxService.releaseReceivingMailbox(_mailbox);
}
}
return blockWithStats;
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/MailboxReceiveOperatorTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/MailboxReceiveOperatorTest.java
index 0af5a807b3d..b2efc3652df 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/MailboxReceiveOperatorTest.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/MailboxReceiveOperatorTest.java
@@ -19,6 +19,8 @@
package org.apache.pinot.query.runtime.operator;
import java.io.IOException;
+import java.lang.ref.ReferenceQueue;
+import java.lang.ref.WeakReference;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
@@ -244,6 +246,45 @@ public class MailboxReceiveOperatorTest {
}
}
+ /// After an error the mailbox stays in the service's cache until it
expires, and the reader callback it holds keeps
+ /// the stream that adapts it reachable. That stream must not point back at
the operator, or every failed query
+ /// would pin its operator tree and execution context for as long as the
mailbox lingers.
+ @Test
+ public void shouldNotStayReachableFromTheMailboxAfterAnError()
+ throws InterruptedException {
+ ReceivingMailbox mailbox = new ReceivingMailbox(MAILBOX_ID_1);
+
when(_mailboxService.getReceivingMailbox(eq(MAILBOX_ID_1))).thenReturn(mailbox);
+ MailboxReceiveOperator operator = getOperator(_stageMetadata1,
RelDistribution.Type.SINGLETON);
+ mailbox.offer(ErrorMseBlock.fromException(new RuntimeException("TEST
ERROR")), List.of(), 1000L);
+ assertTrue(operator.nextBlock().isError());
+ operator.cancel(new RuntimeException("TEST ERROR"));
+ operator.close();
+
+ ReferenceQueue<MailboxReceiveOperator> queue = new ReferenceQueue<>();
+ WeakReference<MailboxReceiveOperator> operatorRef = new
WeakReference<>(operator, queue);
+ operator = null;
+ assertTrue(awaitCleared(operatorRef, queue), "the mailbox must not keep
the closed operator reachable");
+ // The mailbox is the retention path under test, so it has to stay
strongly reachable until this point.
+ assertEquals(mailbox.getId(), MAILBOX_ID_1);
+ }
+
+ /// Waits for `ref` to be cleared, prompting collection as it goes. The
reference queue does the waiting rather than a
+ /// sleep loop over [WeakReference#get()], so the expected case costs one
collection instead of a fixed delay, and the
+ /// failing case still gives the collector several attempts before giving up.
+ private static boolean awaitCleared(WeakReference<?> ref, ReferenceQueue<?>
queue)
+ throws InterruptedException {
+ long deadlineMs = System.currentTimeMillis() + 5_000L;
+ do {
+ System.gc();
+ if (queue.remove(200L) != null) {
+ return true;
+ }
+ } while (System.currentTimeMillis() < deadlineMs);
+ // The collector clears the reference before enqueueing it, so check
directly rather than call a late enqueue a
+ // failure.
+ return ref.get() == null;
+ }
+
@Test
public void shouldEarlyTerminateMailboxesWhenIndicated() {
Object[] row1 = new Object[]{1, 1};
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
