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 91b9f06d478 [multistage] Add gRPC keep-alive to the MSE mailbox
channels (#19383)
91b9f06d478 is described below
commit 91b9f06d4781696d3b65e59dc42d9017e679c859
Author: Gonzalo Ortiz Jaureguizar <[email protected]>
AuthorDate: Mon Aug 31 17:38:15 2026 +0200
[multistage] Add gRPC keep-alive to the MSE mailbox channels (#19383)
---
.../pinot/query/grpc/GrpcKeepAliveConfig.java | 103 ++++++++
.../apache/pinot/query/mailbox/MailboxService.java | 15 +-
.../query/mailbox/channel/ChannelManager.java | 54 +++-
.../query/mailbox/channel/GrpcMailboxServer.java | 32 ++-
.../query/service/dispatch/DispatchClient.java | 52 +---
.../query/service/dispatch/QueryDispatcher.java | 9 +-
.../pinot/query/grpc/GrpcKeepAliveConfigTest.java | 119 +++++++++
.../mailbox/MailboxServiceKeepAliveWiringTest.java | 82 ++++++
.../query/mailbox/channel/ChannelManagerTest.java | 48 +++-
.../channel/MailboxChannelKeepAliveTest.java | 291 +++++++++++++++++++++
.../MailboxServerPermitKeepAliveBehaviorTest.java | 146 +++++++++++
.../channel/MailboxServerPermitKeepAliveTest.java | 77 ++++++
.../apache/pinot/spi/utils/CommonConstants.java | 62 +++++
13 files changed, 1026 insertions(+), 64 deletions(-)
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/grpc/GrpcKeepAliveConfig.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/grpc/GrpcKeepAliveConfig.java
new file mode 100644
index 00000000000..880d9bbff01
--- /dev/null
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/grpc/GrpcKeepAliveConfig.java
@@ -0,0 +1,103 @@
+/**
+ * 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.
+ */
+package org.apache.pinot.query.grpc;
+
+import com.google.common.base.Preconditions;
+import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants;
+
+
+/// Immutable gRPC client keep-alive settings for the multi-stage engine's
internal channels.
+///
+/// Keep-alive makes a *silently* unreachable peer observable. A peer whose
process is gone sends a `RST`, and gRPC
+/// fails the channel immediately; a peer whose kernel is hung, or that sits
behind a one-way partition, sends
+/// nothing at all. Without a transport-level ping such a channel stays in
`READY` indefinitely, and every RPC
+/// issued on it parks until its own deadline. Since channels are cached per
peer, that outlives the query that
+/// first hit it: the cached channel keeps pointing at a dead peer, and
because gRPC only re-resolves DNS when a
+/// transport is dropped, even a peer that has since come back at a new
address is never reached again.
+///
+/// Applies to the MSE's own channels only. The keep-alive settings for the
user-facing gRPC query service live in
+/// [org.apache.pinot.common.config.GrpcConfig].
+///
+/// @param timeMs interval between keep-alive pings, in milliseconds;
keep-alive is disabled when not positive.
+/// gRPC clamps this up to its own 10s floor, so a smaller value
has no effect
+/// @param timeoutMs how long a ping may go unanswered before the transport is
declared dead. Must be positive when
+/// keep-alive is enabled
+/// @param withoutCalls whether to ping while the connection carries no active
RPC. Requires the peer to permit it
+/// (`permitKeepAliveWithoutCalls`), otherwise the peer
answers with `GOAWAY(ENHANCE_YOUR_CALM)`
+public record GrpcKeepAliveConfig(int timeMs, int timeoutMs, boolean
withoutCalls) {
+ /// No keep-alive pings. gRPC's own default, and what every MSE channel did
before keep-alive was configurable.
+ public static final GrpcKeepAliveConfig DISABLED = new
GrpcKeepAliveConfig(-1, 30_000, false);
+
+ /// Reads the policy for the multi-stage engine's **mailbox** channels from
`config`.
+ ///
+ /// The single place these three keys and their defaults are interpreted, so
that anything opening a
+ /// mailbox channel — including an alternative execution engine that
maintains its own channels —
+ /// resolves exactly what
[org.apache.pinot.query.mailbox.channel.ChannelManager] would, and cannot
+ /// drift from it if a default changes.
+ ///
+ /// The broker dispatch channel has its own keys and reads them at its own
call site
+ ///
([org.apache.pinot.broker.requesthandler.MultiStageBrokerRequestHandler]),
because it also has to
+ /// hand them to a [org.apache.pinot.query.service.dispatch.QueryDispatcher]
overload that predates
+ /// this type.
+ public static GrpcKeepAliveConfig forMailboxChannels(PinotConfiguration
config) {
+ return new GrpcKeepAliveConfig(
+
config.getProperty(CommonConstants.MultiStageQueryRunner.KEY_OF_CHANNEL_KEEP_ALIVE_TIME_MS,
+
CommonConstants.MultiStageQueryRunner.DEFAULT_CHANNEL_KEEP_ALIVE_TIME_MS),
+
config.getProperty(CommonConstants.MultiStageQueryRunner.KEY_OF_CHANNEL_KEEP_ALIVE_TIMEOUT_MS,
+
CommonConstants.MultiStageQueryRunner.DEFAULT_CHANNEL_KEEP_ALIVE_TIMEOUT_MS),
+
config.getProperty(CommonConstants.MultiStageQueryRunner.KEY_OF_CHANNEL_KEEP_ALIVE_WITHOUT_CALLS,
+
CommonConstants.MultiStageQueryRunner.DEFAULT_CHANNEL_KEEP_ALIVE_WITHOUT_CALLS));
+ }
+
+ public GrpcKeepAliveConfig {
+ if (timeMs > 0) {
+ Preconditions.checkArgument(timeoutMs > 0,
+ "keepAliveTimeoutMs must be positive when keep-alive is enabled,
got: %s", timeoutMs);
+ }
+ }
+
+ /// Applies these settings to `builder`, leaving it untouched when
keep-alive is disabled.
+ ///
+ /// Centralizing the enabled check here is what keeps a caller from
configuring a `keepAliveTime` of `-1`, which
+ /// gRPC rejects, instead of leaving the ping off.
+ public NettyChannelBuilder configure(NettyChannelBuilder builder) {
+ if (isEnabled()) {
+ builder.keepAliveTime(timeMs, TimeUnit.MILLISECONDS)
+ .keepAliveTimeout(timeoutMs, TimeUnit.MILLISECONDS)
+ .keepAliveWithoutCalls(withoutCalls);
+ }
+ return builder;
+ }
+
+ public boolean isEnabled() {
+ return timeMs > 0;
+ }
+
+ /// Renders as the mailbox and dispatch startup logs show it. Kept instead
of the generated `toString` so a
+ /// disabled policy reads as such rather than as a `-1` interval.
+ @Override
+ public String toString() {
+ return isEnabled()
+ ? "keepAlive[timeMs=" + timeMs + ", timeoutMs=" + timeoutMs + ",
withoutCalls=" + withoutCalls + "]"
+ : "keepAlive[disabled]";
+ }
+}
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/MailboxService.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/MailboxService.java
index 87403ee3a6d..0a130b6692f 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/MailboxService.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/MailboxService.java
@@ -41,6 +41,7 @@ import
org.apache.pinot.core.instance.context.ControllerContext;
import org.apache.pinot.core.instance.context.ServerContext;
import org.apache.pinot.core.transport.grpc.GrpcQueryServer;
import org.apache.pinot.query.access.QueryAccessControlFactory;
+import org.apache.pinot.query.grpc.GrpcKeepAliveConfig;
import org.apache.pinot.query.mailbox.channel.ChannelManager;
import org.apache.pinot.query.mailbox.channel.GrpcMailboxServer;
import org.apache.pinot.query.runtime.operator.MailboxSendOperator;
@@ -131,11 +132,13 @@ public class MailboxService {
int writeBufferLowWaterMarkBytes = config.getProperty(
CommonConstants.MultiStageQueryRunner.KEY_OF_GRPC_WRITE_BUFFER_LOW_WATER_MARK_BYTES,
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_LOW_WATER_MARK_BYTES);
+ GrpcKeepAliveConfig keepAliveConfig =
GrpcKeepAliveConfig.forMailboxChannels(config);
_channelManager = new ChannelManager(_clientSslContext,
_maxInboundMessageSize, getIdleTimeout(config),
- writeBufferHighWaterMarkBytes, writeBufferLowWaterMarkBytes);
+ writeBufferHighWaterMarkBytes, writeBufferLowWaterMarkBytes,
keepAliveConfig);
_accessControlFactory = accessControlFactory;
registerMailboxClientGauges();
- LOGGER.info("Initialized MailboxService with hostname: {}, port: {}",
hostname, port);
+ LOGGER.info("Initialized MailboxService with hostname: {}, port: {},
channel {}", hostname, port,
+ keepAliveConfig);
}
/// Registers gauges exposing the memory used by the gRPC client allocator
@@ -240,6 +243,14 @@ public class MailboxService {
/// [ServerGauge#MAILBOX_CLIENT_USED_DIRECT_MEMORY] — bytes pinned by the
/// shared gRPC client allocator backing every [GrpcSendingMailbox] created
/// from this service.
+ /// The channel manager these mailboxes send through. Exposed so a test can
assert that the policy
+ /// resolved from config actually reaches the transport, which no assertion
on the parsed values can
+ /// show.
+ @VisibleForTesting
+ ChannelManager getChannelManager() {
+ return _channelManager;
+ }
+
@VisibleForTesting
public long getMailboxClientUsedDirectMemoryBytes() {
return _channelManager.usedDirectMemoryBytes();
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/channel/ChannelManager.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/channel/ChannelManager.java
index bc0b0933142..b9b06816935 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/channel/ChannelManager.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/channel/ChannelManager.java
@@ -31,6 +31,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pinot.query.grpc.GrpcKeepAliveConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,6 +51,12 @@ public class ChannelManager {
/// (including TLS negotiation) before sending any message, which increases
the latency of the first query sent after
/// a period of inactivity. In order to achieve that, we set the idle
timeout to a very large value by default.
private final Duration _idleTimeout;
+ /// Transport-level liveness policy for every channel this manager hands out.
+ ///
+ /// The idle timeout above cannot substitute for it. A channel that is being
used is never idle, so on a cluster
+ /// that keeps serving queries the idle timeout never fires — and it is
disabled by default anyway. Keep-alive is
+ /// the only mechanism here that a peer which stopped answering without
closing its socket cannot outlive.
+ private final GrpcKeepAliveConfig _keepAliveConfig;
private final int _maxInboundMessageSize;
/// Buffer allocator configured to prefer direct (off-heap) buffers for
better performance.
/// Using a single allocator instance across all channels allows for better
memory pooling and reduces fragmentation.
@@ -71,11 +78,13 @@ public class ChannelManager {
/// unwritable; it becomes writable
again only when the queue drains below this
/// low watermark. Must satisfy `0 < low
≤ high`; validated eagerly here so
/// misconfiguration surfaces at startup
rather than on the first query.
+ /// @param keepAliveConfig gRPC keep-alive policy applied to every channel;
see [GrpcKeepAliveConfig]
public ChannelManager(@Nullable SslContext clientSslContext, int
maxInboundMessageSize, Duration idleTimeout,
- int writeBufferHighWaterMarkBytes, int writeBufferLowWaterMarkBytes) {
+ int writeBufferHighWaterMarkBytes, int writeBufferLowWaterMarkBytes,
GrpcKeepAliveConfig keepAliveConfig) {
_clientSslContext = clientSslContext;
_maxInboundMessageSize = maxInboundMessageSize;
_idleTimeout = idleTimeout;
+ _keepAliveConfig = keepAliveConfig;
Preconditions.checkArgument(writeBufferLowWaterMarkBytes > 0,
"writeBufferLowWaterMarkBytes must be positive, got: %s",
writeBufferLowWaterMarkBytes);
// The `low <= high` (and `low >= 0`) invariant is also checked by Netty's
WriteBufferWaterMark constructor; by
@@ -96,7 +105,7 @@ public class ChannelManager {
.withOption(ChannelOption.ALLOCATOR, _bufAllocator)
.withOption(ChannelOption.WRITE_BUFFER_WATER_MARK,
_writeBufferWaterMark)
.sslContext(_clientSslContext);
- return decorate(channelBuilder).build();
+ return watchState(decorate(channelBuilder).build(), k.getLeft(),
k.getRight());
}
);
} else {
@@ -108,11 +117,39 @@ public class ChannelManager {
.withOption(ChannelOption.ALLOCATOR, _bufAllocator)
.withOption(ChannelOption.WRITE_BUFFER_WATER_MARK,
_writeBufferWaterMark)
.usePlaintext();
- return decorate(channelBuilder).build();
+ return watchState(decorate(channelBuilder).build(), k.getLeft(),
k.getRight());
});
}
}
+ /// Logs at WARN when `channel` leaves `READY`, and returns it.
+ ///
+ /// Keep-alive makes a silent peer *detectable*; this is what makes it
**legible**. Without it the only
+ /// record of the failure is the absence of one: the startup lines say what
was configured, and a
+ /// channel that later dropped looks exactly like a channel that never had a
problem. The transition is
+ /// the moment worth reading in a log, because it is when a peer stopped
answering — every mailbox send
+ /// queued behind it has already been failing for one keep-alive interval by
then.
+ ///
+ /// Re-arms itself, which is how one callback follows a channel for its
whole life; gRPC's
+ /// `notifyWhenStateChanged` is single-shot. Stops on shutdown, so a
terminated channel cannot keep
+ /// re-registering.
+ private ManagedChannel watchState(ManagedChannel channel, String hostname,
int port) {
+ ConnectivityState state = channel.getState(false);
+ channel.notifyWhenStateChanged(state, () -> {
+ if (channel.isShutdown()) {
+ return;
+ }
+ ConnectivityState next = channel.getState(false);
+ if (state == ConnectivityState.READY && next != ConnectivityState.READY)
{
+ LOGGER.warn("Mailbox channel to {}:{} left READY for {}; sends to that
peer will fail until it "
+ + "reconnects. If keep-alive reported it, the peer stopped
answering about one keep-alive "
+ + "interval ago.", hostname, port, next);
+ }
+ watchState(channel, hostname, port);
+ });
+ return channel;
+ }
+
/// Resets the connection backoff for the channel to the given server if the
channel is in
/// TRANSIENT_FAILURE state. Returns true if a reset was performed, false
otherwise.
///
@@ -128,7 +165,16 @@ public class ChannelManager {
}
private NettyChannelBuilder decorate(NettyChannelBuilder builder) {
- return builder.idleTimeout(_idleTimeout.getSeconds(), TimeUnit.SECONDS);
+ return
_keepAliveConfig.configure(builder.idleTimeout(_idleTimeout.getSeconds(),
TimeUnit.SECONDS));
+ }
+
+ /// The keep-alive policy applied to every channel this manager hands out.
+ ///
+ /// Public because the assertion that matters spans packages: a test of
[MailboxService] has to show
+ /// that the policy resolved from config reached the transport, and no
assertion on parsed values can
+ /// show that. Returns an immutable record, so exposing it grants no control
over the manager.
+ public GrpcKeepAliveConfig getKeepAliveConfig() {
+ return _keepAliveConfig;
}
/// Bytes of direct (off-heap) memory currently pinned by the shared gRPC
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/channel/GrpcMailboxServer.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/channel/GrpcMailboxServer.java
index 285e7804d0d..7f947411f48 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/channel/GrpcMailboxServer.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/channel/GrpcMailboxServer.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.query.mailbox.channel;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import io.grpc.Server;
import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder;
@@ -63,6 +64,8 @@ public class GrpcMailboxServer extends
PinotMailboxGrpc.PinotMailboxImplBase {
private final int _flowControlWindowBytes;
private final int _inboundMessageCredit;
private final boolean _manualInboundFlowControlEnabled;
+ private final int _permitKeepAliveTimeMs;
+ private final boolean _permitKeepAliveWithoutCalls;
/// Constructs a gRPC-based mailbox server.
///
@@ -160,11 +163,25 @@ public class GrpcMailboxServer extends
PinotMailboxGrpc.PinotMailboxImplBase {
"%s (%s) must be >= %s (%s)",
CommonConstants.MultiStageQueryRunner.KEY_OF_GRPC_FLOW_CONTROL_WINDOW_BYTES,
_flowControlWindowBytes,
CommonConstants.MultiStageQueryRunner.KEY_OF_MAX_INBOUND_QUERY_DATA_BLOCK_SIZE_BYTES,
maxInboundMessageSize);
+ // Keep-alive enforcement. A peer configured with a keep-alive time below
permitKeepAliveTime has its pings
+ // counted as "bad" and, past the server's strike threshold, gets
GOAWAY(ENHANCE_YOUR_CALM) — which drops the
+ // mailbox channel mid-query. Defaults match Netty's own so that a peer
left at the default keep-alive time is
+ // never punished; both sides have to be tuned down together.
+ _permitKeepAliveTimeMs = config.getProperty(
+
CommonConstants.MultiStageQueryRunner.KEY_OF_MAILBOX_SERVER_PERMIT_KEEP_ALIVE_TIME_MS,
+
CommonConstants.MultiStageQueryRunner.DEFAULT_OF_MAILBOX_SERVER_PERMIT_KEEP_ALIVE_TIME_MS);
+ _permitKeepAliveWithoutCalls = config.getProperty(
+
CommonConstants.MultiStageQueryRunner.KEY_OF_MAILBOX_SERVER_PERMIT_KEEP_ALIVE_WITHOUT_CALLS,
+
CommonConstants.MultiStageQueryRunner.DEFAULT_OF_MAILBOX_SERVER_PERMIT_KEEP_ALIVE_WITHOUT_CALLS);
+ if (_permitKeepAliveTimeMs > 0) {
+ builder.permitKeepAliveTime(_permitKeepAliveTimeMs,
TimeUnit.MILLISECONDS);
+ }
builder
.addService(this)
.withOption(ChannelOption.ALLOCATOR, bufAllocator)
.withChildOption(ChannelOption.ALLOCATOR, bufAllocator)
.maxInboundMessageSize(maxInboundMessageSize)
+ .permitKeepAliveWithoutCalls(_permitKeepAliveWithoutCalls)
.flowControlWindow(_flowControlWindowBytes);
// Add SSL context only if TLS is configured
@@ -177,10 +194,21 @@ public class GrpcMailboxServer extends
PinotMailboxGrpc.PinotMailboxImplBase {
_server = builder.build();
}
+ @VisibleForTesting
+ int getPermitKeepAliveTimeMs() {
+ return _permitKeepAliveTimeMs;
+ }
+
+ @VisibleForTesting
+ boolean isPermitKeepAliveWithoutCalls() {
+ return _permitKeepAliveWithoutCalls;
+ }
+
public void start() {
LOGGER.info("Starting GrpcMailboxServer with flowControlWindow={} bytes,
inboundMessageCredit={}, "
- + "manualInboundFlowControlEnabled={}",
- _flowControlWindowBytes, _inboundMessageCredit,
_manualInboundFlowControlEnabled);
+ + "manualInboundFlowControlEnabled={}, permitKeepAliveTimeMs={},
permitKeepAliveWithoutCalls={}",
+ _flowControlWindowBytes, _inboundMessageCredit,
_manualInboundFlowControlEnabled, _permitKeepAliveTimeMs,
+ _permitKeepAliveWithoutCalls);
try {
_server.start();
} catch (IOException e) {
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/DispatchClient.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/DispatchClient.java
index a0332f97911..489ab1a3595 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/DispatchClient.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/DispatchClient.java
@@ -18,7 +18,6 @@
*/
package org.apache.pinot.query.service.dispatch;
-import com.google.common.base.Preconditions;
import io.grpc.Deadline;
import io.grpc.ManagedChannel;
import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
@@ -27,7 +26,6 @@ import io.grpc.netty.shaded.io.netty.channel.ChannelOption;
import io.grpc.netty.shaded.io.netty.handler.ssl.SslContext;
import io.grpc.stub.StreamObserver;
import java.util.List;
-import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import javax.annotation.Nullable;
@@ -35,6 +33,7 @@ import org.apache.pinot.common.config.TlsConfig;
import org.apache.pinot.common.proto.PinotQueryWorkerGrpc;
import org.apache.pinot.common.proto.Worker;
import org.apache.pinot.common.utils.grpc.ServerGrpcQueryClient;
+import org.apache.pinot.query.grpc.GrpcKeepAliveConfig;
import org.apache.pinot.query.routing.QueryServerInstance;
import
org.apache.pinot.query.service.dispatch.streaming.StreamingDispatchObserver;
import org.apache.pinot.query.service.dispatch.streaming.StreamingQuerySession;
@@ -59,15 +58,15 @@ class DispatchClient {
private final PinotQueryWorkerGrpc.PinotQueryWorkerStub _dispatchStub;
public DispatchClient(String host, int port, @Nullable TlsConfig tlsConfig) {
- this(host, port, tlsConfig, null, KeepAliveConfig.DISABLED);
+ this(host, port, tlsConfig, null, GrpcKeepAliveConfig.DISABLED);
}
public DispatchClient(String host, int port, @Nullable TlsConfig tlsConfig,
@Nullable SslContext sslContext) {
- this(host, port, tlsConfig, sslContext, KeepAliveConfig.DISABLED);
+ this(host, port, tlsConfig, sslContext, GrpcKeepAliveConfig.DISABLED);
}
DispatchClient(String host, int port, @Nullable TlsConfig tlsConfig,
@Nullable SslContext sslContext,
- KeepAliveConfig keepAliveConfig) {
+ GrpcKeepAliveConfig keepAliveConfig) {
// Always use NettyChannelBuilder to allow setting Netty-specific channel
options like the buffer allocator.
// This ensures we can explicitly configure direct (off-heap) buffers for
better performance.
NettyChannelBuilder channelBuilder = NettyChannelBuilder.forAddress(host,
port)
@@ -81,51 +80,10 @@ class DispatchClient {
}
// Enable gRPC keep-alive when configured so that a silently unreachable
peer transitions the channel out of READY,
// which lets the broker's FailureDetector exclude it from routing.
- if (keepAliveConfig.isEnabled()) {
- channelBuilder.keepAliveTime(keepAliveConfig.getTimeMs(),
TimeUnit.MILLISECONDS)
- .keepAliveTimeout(keepAliveConfig.getTimeoutMs(),
TimeUnit.MILLISECONDS)
- .keepAliveWithoutCalls(keepAliveConfig.isWithoutCalls());
- }
- _channel = channelBuilder.build();
+ _channel = keepAliveConfig.configure(channelBuilder).build();
_dispatchStub = PinotQueryWorkerGrpc.newStub(_channel);
}
- /// Immutable gRPC keep-alive configuration for broker dispatch channels.
Keep-alive is disabled when `timeMs` is not
- /// positive.
- static final class KeepAliveConfig {
- static final KeepAliveConfig DISABLED = new KeepAliveConfig(-1, 30_000,
false);
-
- private final int _timeMs;
- private final int _timeoutMs;
- private final boolean _withoutCalls;
-
- KeepAliveConfig(int timeMs, int timeoutMs, boolean withoutCalls) {
- if (timeMs > 0) {
- Preconditions.checkArgument(timeoutMs > 0,
- "keepAliveTimeoutMs must be positive when keep-alive is enabled,
got: %s", timeoutMs);
- }
- _timeMs = timeMs;
- _timeoutMs = timeoutMs;
- _withoutCalls = withoutCalls;
- }
-
- boolean isEnabled() {
- return _timeMs > 0;
- }
-
- int getTimeMs() {
- return _timeMs;
- }
-
- int getTimeoutMs() {
- return _timeoutMs;
- }
-
- boolean isWithoutCalls() {
- return _withoutCalls;
- }
- }
-
public ManagedChannel getChannel() {
return _channel;
}
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/QueryDispatcher.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/QueryDispatcher.java
index 80d9f1134aa..834d3b008b8 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/QueryDispatcher.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/QueryDispatcher.java
@@ -64,6 +64,7 @@ import org.apache.pinot.core.transport.ServerInstance;
import
org.apache.pinot.core.transport.server.routing.stats.ServerRoutingStatsManager;
import org.apache.pinot.core.util.DataBlockExtractUtils;
import org.apache.pinot.core.util.trace.TracedThreadFactory;
+import org.apache.pinot.query.grpc.GrpcKeepAliveConfig;
import org.apache.pinot.query.mailbox.MailboxService;
import org.apache.pinot.query.planner.PlanFragment;
import org.apache.pinot.query.planner.physical.DispatchablePlanFragment;
@@ -124,7 +125,7 @@ public class QueryDispatcher {
private final TlsConfig _tlsConfig;
@Nullable
private final SslContext _clientGrpcSslContext;
- private final DispatchClient.KeepAliveConfig _keepAliveConfig;
+ private final GrpcKeepAliveConfig _keepAliveConfig;
// maps broker-generated query id to the set of servers that the query was
dispatched to
private final Map<Long, Set<QueryServerInstance>> _serversByQuery;
private final FailureDetector _failureDetector;
@@ -136,7 +137,7 @@ public class QueryDispatcher {
public QueryDispatcher(MailboxService mailboxService, FailureDetector
failureDetector, @Nullable TlsConfig tlsConfig,
boolean enableCancellation, Duration cancelTimeout) {
this(mailboxService, failureDetector, tlsConfig, enableCancellation,
cancelTimeout,
- DispatchClient.KeepAliveConfig.DISABLED, false,
CommonConstants.Broker.DEFAULT_STREAM_STATS_DRAIN_MS);
+ GrpcKeepAliveConfig.DISABLED, false,
CommonConstants.Broker.DEFAULT_STREAM_STATS_DRAIN_MS);
}
/// Overload that accepts gRPC keep-alive settings for broker dispatch
channels. A non-positive `keepAliveTimeMs`
@@ -145,12 +146,12 @@ public class QueryDispatcher {
boolean enableCancellation, Duration cancelTimeout, int keepAliveTimeMs,
int keepAliveTimeoutMs,
boolean keepAliveWithoutCalls, boolean streamStatsDefault, long
statsDrainMs) {
this(mailboxService, failureDetector, tlsConfig, enableCancellation,
cancelTimeout,
- new DispatchClient.KeepAliveConfig(keepAliveTimeMs,
keepAliveTimeoutMs, keepAliveWithoutCalls),
+ new GrpcKeepAliveConfig(keepAliveTimeMs, keepAliveTimeoutMs,
keepAliveWithoutCalls),
streamStatsDefault, statsDrainMs);
}
private QueryDispatcher(MailboxService mailboxService, FailureDetector
failureDetector, @Nullable TlsConfig tlsConfig,
- boolean enableCancellation, Duration cancelTimeout,
DispatchClient.KeepAliveConfig keepAliveConfig,
+ boolean enableCancellation, Duration cancelTimeout, GrpcKeepAliveConfig
keepAliveConfig,
boolean streamStatsDefault, long statsDrainMs) {
_cancelTimeout = cancelTimeout;
_statsDrainMs = statsDrainMs;
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/grpc/GrpcKeepAliveConfigTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/grpc/GrpcKeepAliveConfigTest.java
new file mode 100644
index 00000000000..d520fad322f
--- /dev/null
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/grpc/GrpcKeepAliveConfigTest.java
@@ -0,0 +1,119 @@
+/**
+ * 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.
+ */
+package org.apache.pinot.query.grpc;
+
+import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
+import java.util.Map;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.MultiStageQueryRunner;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertTrue;
+
+
+public class GrpcKeepAliveConfigTest {
+
+ @Test
+ public void testDisabledWhenTimeNotPositive() {
+ assertFalse(GrpcKeepAliveConfig.DISABLED.isEnabled());
+ assertFalse(new GrpcKeepAliveConfig(0, 30_000, false).isEnabled());
+ assertFalse(new GrpcKeepAliveConfig(-1, 30_000, false).isEnabled());
+ }
+
+ @Test
+ public void testEnabledExposesSettings() {
+ GrpcKeepAliveConfig config = new GrpcKeepAliveConfig(300_000, 30_000,
true);
+ assertTrue(config.isEnabled());
+ assertEquals(config.timeMs(), 300_000);
+ assertEquals(config.timeoutMs(), 30_000);
+ assertTrue(config.withoutCalls());
+ }
+
+ /// A zero or negative timeout is only rejected when keep-alive is on: gRPC
would throw on it, while
+ /// [GrpcKeepAliveConfig#DISABLED] must stay constructible whatever the
timeout is.
+ @Test(expectedExceptions = IllegalArgumentException.class,
+ expectedExceptionsMessageRegExp = ".*keepAliveTimeoutMs must be
positive.*")
+ public void testRejectsNonPositiveTimeoutWhenEnabled() {
+ new GrpcKeepAliveConfig(300_000, 0, false);
+ }
+
+ @Test
+ public void testDisabledAllowsAnyTimeout() {
+ assertFalse(new GrpcKeepAliveConfig(-1, 0, false).isEnabled());
+ }
+
+ /// The whole point of routing both call sites through `configure`: a
disabled policy must not reach gRPC, which
+ /// rejects a non-positive `keepAliveTime` with an
`IllegalArgumentException`.
+ @Test
+ public void testConfigureIsNoOpWhenDisabled() {
+ NettyChannelBuilder builder = NettyChannelBuilder.forAddress("localhost",
12345).usePlaintext();
+ assertSame(GrpcKeepAliveConfig.DISABLED.configure(builder), builder);
+ builder.build().shutdownNow();
+ }
+
+ @Test
+ public void testConfigureAppliesWhenEnabled() {
+ NettyChannelBuilder builder = NettyChannelBuilder.forAddress("localhost",
12345).usePlaintext();
+ assertSame(new GrpcKeepAliveConfig(30_000, 5_000,
true).configure(builder), builder);
+ builder.build().shutdownNow();
+ }
+
+ /// Defaults must be the ones an operator reads in the docs, resolved in one
place so both the Java
+ /// mailbox and any other engine maintaining its own mailbox channels see
the same values.
+ @Test
+ public void testForMailboxChannelsDefaults() {
+ GrpcKeepAliveConfig config = GrpcKeepAliveConfig.forMailboxChannels(new
PinotConfiguration(Map.of()));
+ assertTrue(config.isEnabled(), "mailbox channels must ping by default");
+ assertEquals(config.timeMs(),
MultiStageQueryRunner.DEFAULT_CHANNEL_KEEP_ALIVE_TIME_MS);
+ assertEquals(config.timeoutMs(),
MultiStageQueryRunner.DEFAULT_CHANNEL_KEEP_ALIVE_TIMEOUT_MS);
+ // Off by default: pings without an active call are refused by a peer left
at Netty's own default.
+ assertFalse(config.withoutCalls());
+ }
+
+ /// The default interval must not exceed what an un-upgraded peer permits,
or an upgraded instance
+ /// would have its mailbox channels torn down with
`GOAWAY(ENHANCE_YOUR_CALM)` during a rolling
+ /// upgrade.
+ @Test
+ public void testDefaultIntervalIsSafeAgainstNettyServerDefault() {
+ assertTrue(MultiStageQueryRunner.DEFAULT_CHANNEL_KEEP_ALIVE_TIME_MS
+ >=
MultiStageQueryRunner.DEFAULT_OF_MAILBOX_SERVER_PERMIT_KEEP_ALIVE_TIME_MS,
+ "client keep-alive time must be >= the permit enforced by peers");
+ }
+
+ @Test
+ public void testForMailboxChannelsOverrides() {
+ GrpcKeepAliveConfig config = GrpcKeepAliveConfig.forMailboxChannels(new
PinotConfiguration(Map.of(
+ MultiStageQueryRunner.KEY_OF_CHANNEL_KEEP_ALIVE_TIME_MS, 30_000,
+ MultiStageQueryRunner.KEY_OF_CHANNEL_KEEP_ALIVE_TIMEOUT_MS, 5_000,
+ MultiStageQueryRunner.KEY_OF_CHANNEL_KEEP_ALIVE_WITHOUT_CALLS, true)));
+ assertEquals(config.timeMs(), 30_000);
+ assertEquals(config.timeoutMs(), 5_000);
+ assertTrue(config.withoutCalls());
+ }
+
+ @Test
+ public void testForMailboxChannelsCanTurnKeepAliveOff() {
+ GrpcKeepAliveConfig config = GrpcKeepAliveConfig.forMailboxChannels(new
PinotConfiguration(Map.of(
+ MultiStageQueryRunner.KEY_OF_CHANNEL_KEEP_ALIVE_TIME_MS, -1)));
+ assertFalse(config.isEnabled());
+ }
+}
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/MailboxServiceKeepAliveWiringTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/MailboxServiceKeepAliveWiringTest.java
new file mode 100644
index 00000000000..309d57fb1aa
--- /dev/null
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/MailboxServiceKeepAliveWiringTest.java
@@ -0,0 +1,82 @@
+/**
+ * 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.
+ */
+package org.apache.pinot.query.mailbox;
+
+import java.util.Map;
+import org.apache.pinot.query.grpc.GrpcKeepAliveConfig;
+import org.apache.pinot.query.testutils.QueryTestUtils;
+import org.apache.pinot.spi.config.instance.InstanceType;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.MultiStageQueryRunner;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+
+
+/// Pins the hop from configuration to transport: the policy [MailboxService]
resolves must be the one
+/// its [org.apache.pinot.query.mailbox.channel.ChannelManager] hands to every
channel.
+///
+/// Asserting the parsed values alone cannot show this. A `MailboxService`
that resolved the config
+/// correctly and then passed [GrpcKeepAliveConfig#DISABLED] to the channel
manager would satisfy every
+/// parsing test while shipping channels with no keep-alive at all — the exact
defect this change exists
+/// to prevent.
+public class MailboxServiceKeepAliveWiringTest {
+
+ @Test
+ public void testResolvedPolicyReachesTheChannelManager() {
+ PinotConfiguration config = new PinotConfiguration(Map.of(
+ MultiStageQueryRunner.KEY_OF_CHANNEL_KEEP_ALIVE_TIME_MS, 45_000,
+ MultiStageQueryRunner.KEY_OF_CHANNEL_KEEP_ALIVE_TIMEOUT_MS, 7_000,
+ MultiStageQueryRunner.KEY_OF_CHANNEL_KEEP_ALIVE_WITHOUT_CALLS, true));
+ MailboxService mailboxService = newMailboxService(config);
+ GrpcKeepAliveConfig applied =
mailboxService.getChannelManager().getKeepAliveConfig();
+ assertEquals(applied, GrpcKeepAliveConfig.forMailboxChannels(config),
+ "the channel manager must get exactly the policy resolved from
config");
+ assertEquals(applied.timeMs(), 45_000);
+ assertEquals(applied.timeoutMs(), 7_000);
+ assertTrue(applied.withoutCalls());
+ }
+
+ @Test
+ public void testDefaultPolicyReachesTheChannelManager() {
+ MailboxService mailboxService = newMailboxService(new
PinotConfiguration(Map.of()));
+ GrpcKeepAliveConfig applied =
mailboxService.getChannelManager().getKeepAliveConfig();
+ assertTrue(applied.isEnabled(), "mailbox channels must ping by default");
+ assertEquals(applied.timeMs(),
MultiStageQueryRunner.DEFAULT_CHANNEL_KEEP_ALIVE_TIME_MS);
+ assertEquals(applied.timeoutMs(),
MultiStageQueryRunner.DEFAULT_CHANNEL_KEEP_ALIVE_TIMEOUT_MS);
+ assertFalse(applied.withoutCalls());
+ }
+
+ /// Disabling must reach the transport too, so an operator turning
keep-alive off gets channels
+ /// without pings rather than channels with the defaults.
+ @Test
+ public void testDisabledPolicyReachesTheChannelManager() {
+ MailboxService mailboxService = newMailboxService(new
PinotConfiguration(Map.of(
+ MultiStageQueryRunner.KEY_OF_CHANNEL_KEEP_ALIVE_TIME_MS, -1)));
+
assertFalse(mailboxService.getChannelManager().getKeepAliveConfig().isEnabled());
+ }
+
+ /// Constructed but never started: the channel manager and its policy are
resolved in the
+ /// constructor, so no port needs binding.
+ private static MailboxService newMailboxService(PinotConfiguration config) {
+ return new MailboxService("localhost", QueryTestUtils.getAvailablePort(),
InstanceType.SERVER, config);
+ }
+}
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/channel/ChannelManagerTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/channel/ChannelManagerTest.java
index 11faa327dc1..0eab4e2bde2 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/channel/ChannelManagerTest.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/channel/ChannelManagerTest.java
@@ -24,6 +24,7 @@ import java.lang.reflect.Field;
import java.time.Duration;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pinot.query.grpc.GrpcKeepAliveConfig;
import org.apache.pinot.spi.utils.CommonConstants;
import org.testng.annotations.Test;
@@ -41,7 +42,8 @@ public class ChannelManagerTest {
public void testResetConnectBackoffNoOpForUnknownChannel() {
ChannelManager channelManager = new ChannelManager(null, 4_000_000,
Duration.ofDays(365),
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_HIGH_WATER_MARK_BYTES,
-
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_LOW_WATER_MARK_BYTES);
+
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_LOW_WATER_MARK_BYTES,
+ GrpcKeepAliveConfig.DISABLED);
// Should return false and not throw when no channel exists for the given
host/port
assertFalse(channelManager.resetConnectBackoff("unknown-host", 12345));
}
@@ -50,7 +52,8 @@ public class ChannelManagerTest {
public void testResetConnectBackoffNoOpWhenNotInTransientFailure() {
ChannelManager channelManager = new ChannelManager(null, 4_000_000,
Duration.ofDays(365),
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_HIGH_WATER_MARK_BYTES,
-
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_LOW_WATER_MARK_BYTES);
+
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_LOW_WATER_MARK_BYTES,
+ GrpcKeepAliveConfig.DISABLED);
// Create a channel by calling getChannel
ManagedChannel channel = channelManager.getChannel("localhost", 12345);
try {
@@ -70,7 +73,8 @@ public class ChannelManagerTest {
throws Exception {
ChannelManager channelManager = new ChannelManager(null, 4_000_000,
Duration.ofDays(365),
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_HIGH_WATER_MARK_BYTES,
-
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_LOW_WATER_MARK_BYTES);
+
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_LOW_WATER_MARK_BYTES,
+ GrpcKeepAliveConfig.DISABLED);
ManagedChannel mockChannel = mock(ManagedChannel.class);
when(mockChannel.getState(false)).thenReturn(ConnectivityState.TRANSIENT_FAILURE);
@@ -94,7 +98,7 @@ public class ChannelManagerTest {
public void testConstructorRejectsZeroWriteBufferLowWaterMark() {
new ChannelManager(null, 4_000_000, Duration.ofDays(365),
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_HIGH_WATER_MARK_BYTES,
- 0);
+ 0, GrpcKeepAliveConfig.DISABLED);
}
/// Pins the eager `new WriteBufferWaterMark(low, high)` invariant: when
`low > high`, Netty's own
@@ -105,6 +109,40 @@ public class ChannelManagerTest {
public void testConstructorRejectsLowWatermarkAboveHighWatermark() {
new ChannelManager(null, 4_000_000, Duration.ofDays(365),
32 * 1024 * 1024, // high
- 64 * 1024 * 1024); // low > high
+ 64 * 1024 * 1024, // low > high
+ GrpcKeepAliveConfig.DISABLED);
+ }
+
+ /// Pins that the manager hands its keep-alive policy to every channel it
builds.
+ ///
+ /// gRPC exposes nothing on a built `ManagedChannel` to read the keep-alive
settings back, so this asserts the
+ /// policy the manager holds; that the policy reaches the transport is
covered end to end by
+ /// [MailboxChannelKeepAliveTest].
+ @Test
+ public void testKeepAliveConfigRetained() {
+ GrpcKeepAliveConfig keepAlive = new GrpcKeepAliveConfig(30_000, 5_000,
true);
+ ChannelManager channelManager = new ChannelManager(null, 4_000_000,
Duration.ofDays(365),
+
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_HIGH_WATER_MARK_BYTES,
+
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_LOW_WATER_MARK_BYTES,
keepAlive);
+ assertSame(channelManager.getKeepAliveConfig(), keepAlive);
+ // Building a channel with keep-alive enabled must not throw: gRPC rejects
a non-positive keepAliveTime, and the
+ // enabled check that prevents that lives in GrpcKeepAliveConfig rather
than at this call site.
+ ManagedChannel channel = channelManager.getChannel("localhost", 12346);
+ try {
+ assertSame(channelManager.getChannel("localhost", 12346), channel);
+ } finally {
+ channel.shutdownNow();
+ }
+ }
+
+ /// A disabled policy must leave the builder alone rather than passing `-1`
to gRPC, which throws.
+ @Test
+ public void testDisabledKeepAliveStillBuildsChannel() {
+ ChannelManager channelManager = new ChannelManager(null, 4_000_000,
Duration.ofDays(365),
+
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_HIGH_WATER_MARK_BYTES,
+
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_LOW_WATER_MARK_BYTES,
+ GrpcKeepAliveConfig.DISABLED);
+ ManagedChannel channel = channelManager.getChannel("localhost", 12347);
+ channel.shutdownNow();
}
}
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/channel/MailboxChannelKeepAliveTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/channel/MailboxChannelKeepAliveTest.java
new file mode 100644
index 00000000000..0205b590903
--- /dev/null
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/channel/MailboxChannelKeepAliveTest.java
@@ -0,0 +1,291 @@
+/**
+ * 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.
+ */
+package org.apache.pinot.query.mailbox.channel;
+
+import io.grpc.ManagedChannel;
+import io.grpc.Server;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder;
+import io.grpc.stub.StreamObserver;
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.SocketTimeoutException;
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.pinot.common.proto.Mailbox;
+import org.apache.pinot.common.proto.PinotMailboxGrpc;
+import org.apache.pinot.query.grpc.GrpcKeepAliveConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+
+
+/// Proves that a mailbox channel notices a peer that stops answering
*without* closing its socket, which is the
+/// failure keep-alive exists for: no `RST` ever arrives, so gRPC keeps the
cached channel in `READY` and every send
+/// on it parks until the query deadline — for every later query too, since
channels are cached per peer.
+///
+/// The peer is faked with a TCP relay that stops moving bytes in both
directions while holding both sockets open.
+/// That is indistinguishable, from the client's side, from a host whose
kernel is hung or that sits behind a
+/// one-way partition. Killing the server process instead would prove nothing:
it sends a `RST`, and gRPC has always
+/// handled that.
+///
+/// Both channels in this test are black-holed at the same instant and only
the keep-alive one is expected to fail,
+/// so the assertion that the other is still `READY` is what keeps this test
honest — ablate the keep-alive wiring
+/// and the first assertion fails rather than the test passing for the wrong
reason.
+public class MailboxChannelKeepAliveTest {
+ /// gRPC clamps `keepAliveTime` up to its own 10s floor, so this is as fast
as detection can be configured.
+ private static final int KEEP_ALIVE_TIME_MS = 10_000;
+ private static final int KEEP_ALIVE_TIMEOUT_MS = 1_000;
+ /// Generous headroom over the ~11s the settings above imply; a loaded CI
box must not flake this.
+ private static final long DETECTION_BUDGET_MS = 45_000;
+
+ @Test(timeOut = 120_000)
+ public void testKeepAliveDetectsSilentPeer()
+ throws Exception {
+ SilentMailbox silentPeer = new SilentMailbox(2);
+ Server peer =
NettyServerBuilder.forPort(0).addService(silentPeer).build().start();
+ try (BlackHoleRelay relay = new BlackHoleRelay(peer.getPort())) {
+ ChannelManager keepAliveManager = newChannelManager(
+ new GrpcKeepAliveConfig(KEEP_ALIVE_TIME_MS, KEEP_ALIVE_TIMEOUT_MS,
false));
+ ChannelManager noKeepAliveManager =
newChannelManager(GrpcKeepAliveConfig.DISABLED);
+
+ ManagedChannel keepAliveChannel =
keepAliveManager.getChannel("localhost", relay.getPort());
+ ManagedChannel noKeepAliveChannel =
noKeepAliveManager.getChannel("localhost", relay.getPort());
+ try {
+ // Keep-alive with `withoutCalls` off only pings while a call is
active, which is the production default, so
+ // both channels get an open stream. Sending one message is what
forces the RPC onto the transport.
+ OpenStream withKeepAlive = openStream(keepAliveChannel);
+ OpenStream withoutKeepAlive = openStream(noKeepAliveChannel);
+
+ // Wait until the *peer* has received both streams, not merely until
the relay accepted both TCP
+ // connections. Accepting happens before a single HTTP/2 byte is
relayed, so black-holing on that signal can
+ // strand the RPC before it ever starts — and with `withoutCalls` off,
a channel with no active call is
+ // never pinged, so the test would fail with keep-alive working
perfectly.
+ assertTrue(silentPeer.awaitStreams(30_000), "streams did not reach the
peer");
+
+ relay.blackHole();
+
+ assertTrue(withKeepAlive.await(DETECTION_BUDGET_MS),
+ "keep-alive channel never failed: a silent peer must not be able
to park a mailbox send forever");
+ Throwable error = withKeepAlive._error.get();
+ assertNotNull(error);
+ assertEquals(Status.fromThrowable(error).getCode(),
Status.Code.UNAVAILABLE,
+ "expected the transport to be declared dead, got: " + error);
+
+ // Same black hole, same elapsed time, keep-alive off: still believed
healthy. This is the state the
+ // production default used to leave every mailbox channel in.
+ assertNull(withoutKeepAlive._error.get(),
+ "channel without keep-alive should still believe the dead peer is
healthy");
+ } finally {
+ keepAliveChannel.shutdownNow();
+ noKeepAliveChannel.shutdownNow();
+ }
+ } finally {
+ peer.shutdownNow();
+ }
+ }
+
+ private static ChannelManager newChannelManager(GrpcKeepAliveConfig
keepAliveConfig) {
+ return new ChannelManager(null, 4_000_000, Duration.ofDays(365),
+
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_HIGH_WATER_MARK_BYTES,
+
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_LOW_WATER_MARK_BYTES,
keepAliveConfig);
+ }
+
+ private static OpenStream openStream(ManagedChannel channel) {
+ OpenStream stream = new OpenStream();
+ StreamObserver<Mailbox.MailboxContent> sender =
PinotMailboxGrpc.newStub(channel).open(stream);
+ sender.onNext(Mailbox.MailboxContent.getDefaultInstance());
+ return stream;
+ }
+
+ /// Client-side observer of one `open` stream: records the terminal error,
if any.
+ private static final class OpenStream implements
StreamObserver<Mailbox.MailboxStatus> {
+ private final CountDownLatch _terminated = new CountDownLatch(1);
+ private final AtomicReference<Throwable> _error = new AtomicReference<>();
+
+ @Override
+ public void onNext(Mailbox.MailboxStatus value) {
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ _error.compareAndSet(null, t);
+ _terminated.countDown();
+ }
+
+ @Override
+ public void onCompleted() {
+ _error.compareAndSet(null, new
StatusRuntimeException(Status.INTERNAL.withDescription("unexpected
completion")));
+ _terminated.countDown();
+ }
+
+ boolean await(long timeoutMs)
+ throws InterruptedException {
+ return _terminated.await(timeoutMs, TimeUnit.MILLISECONDS);
+ }
+ }
+
+ /// A mailbox peer that accepts `open` and then says nothing at all, so the
only traffic left on the connection is
+ /// the keep-alive ping this test is about.
+ private static final class SilentMailbox extends
PinotMailboxGrpc.PinotMailboxImplBase {
+ private final CountDownLatch _streamsReceived;
+
+ private SilentMailbox(int expectedStreams) {
+ _streamsReceived = new CountDownLatch(expectedStreams);
+ }
+
+ boolean awaitStreams(long timeoutMs)
+ throws InterruptedException {
+ return _streamsReceived.await(timeoutMs, TimeUnit.MILLISECONDS);
+ }
+
+ @Override
+ public StreamObserver<Mailbox.MailboxContent>
open(StreamObserver<Mailbox.MailboxStatus> responseObserver) {
+ return new StreamObserver<>() {
+ private final AtomicBoolean _counted = new AtomicBoolean();
+
+ @Override
+ public void onNext(Mailbox.MailboxContent value) {
+ // Counting the first message rather than `open` itself: a message
proves the client's stream is live on
+ // the transport, which is precisely the precondition for keep-alive
pings.
+ if (_counted.compareAndSet(false, true)) {
+ _streamsReceived.countDown();
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ }
+
+ @Override
+ public void onCompleted() {
+ }
+ };
+ }
+ }
+
+ /// TCP relay that can stop forwarding while keeping every socket open.
+ ///
+ /// Sockets are read with a short `SO_TIMEOUT` rather than blocking
indefinitely, so a relay thread parked in
+ /// `read` still notices [#blackHole] promptly. Nothing is ever closed until
[#close], because closing is exactly
+ /// what this fixture must not do: a closed socket produces the `FIN`/`RST`
that gRPC already detects.
+ private static final class BlackHoleRelay implements Closeable {
+ private final ServerSocket _listener;
+ private final int _upstreamPort;
+ private final ExecutorService _executor = Executors.newCachedThreadPool();
+ private final List<Socket> _sockets = new CopyOnWriteArrayList<>();
+ private final AtomicBoolean _blackHole = new AtomicBoolean();
+ private final AtomicBoolean _closed = new AtomicBoolean();
+
+ BlackHoleRelay(int upstreamPort)
+ throws IOException {
+ _upstreamPort = upstreamPort;
+ _listener = new ServerSocket();
+ _listener.bind(new InetSocketAddress("localhost", 0));
+ _executor.submit(this::acceptLoop);
+ }
+
+ int getPort() {
+ return _listener.getLocalPort();
+ }
+
+ /// Stops moving bytes in both directions, holding all sockets open.
+ void blackHole() {
+ _blackHole.set(true);
+ }
+
+ private void acceptLoop() {
+ while (!_closed.get()) {
+ try {
+ Socket downstream = _listener.accept();
+ Socket upstream = new Socket("localhost", _upstreamPort);
+ downstream.setSoTimeout(100);
+ upstream.setSoTimeout(100);
+ _sockets.add(downstream);
+ _sockets.add(upstream);
+ _executor.submit(() -> relay(downstream, upstream));
+ _executor.submit(() -> relay(upstream, downstream));
+ } catch (IOException e) {
+ return;
+ }
+ }
+ }
+
+ private void relay(Socket from, Socket to) {
+ byte[] buffer = new byte[8192];
+ try {
+ InputStream in = from.getInputStream();
+ OutputStream out = to.getOutputStream();
+ while (!_closed.get()) {
+ if (_blackHole.get()) {
+ // Park instead of returning: returning would let the streams be
garbage collected and the sockets
+ // closed, which would hand the client the RST this fixture must
withhold.
+ Thread.sleep(50);
+ continue;
+ }
+ int read;
+ try {
+ read = in.read(buffer);
+ } catch (SocketTimeoutException e) {
+ continue;
+ }
+ if (read < 0) {
+ return;
+ }
+ out.write(buffer, 0, read);
+ out.flush();
+ }
+ } catch (IOException | InterruptedException e) {
+ // Relay is done; the test asserts on the client's view, not on this
thread.
+ }
+ }
+
+ @Override
+ public void close()
+ throws IOException {
+ _closed.set(true);
+ _listener.close();
+ for (Socket socket : _sockets) {
+ try {
+ socket.close();
+ } catch (IOException e) {
+ // Best effort teardown.
+ }
+ }
+ _executor.shutdownNow();
+ }
+ }
+}
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/channel/MailboxServerPermitKeepAliveBehaviorTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/channel/MailboxServerPermitKeepAliveBehaviorTest.java
new file mode 100644
index 00000000000..7e7cc7b64a7
--- /dev/null
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/channel/MailboxServerPermitKeepAliveBehaviorTest.java
@@ -0,0 +1,146 @@
+/**
+ * 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.
+ */
+package org.apache.pinot.query.mailbox.channel;
+
+import io.grpc.ManagedChannel;
+import io.grpc.Status;
+import io.grpc.stub.StreamObserver;
+import java.time.Duration;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.pinot.common.proto.Mailbox;
+import org.apache.pinot.common.proto.PinotMailboxGrpc;
+import org.apache.pinot.query.grpc.GrpcKeepAliveConfig;
+import org.apache.pinot.query.mailbox.MailboxService;
+import org.apache.pinot.query.testutils.QueryTestUtils;
+import org.apache.pinot.spi.config.instance.InstanceType;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.MultiStageQueryRunner;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+
+
+/// Proves the mailbox server's keep-alive **enforcement** actually changes
behaviour, which the
+/// config-parsing tests in [MailboxServerPermitKeepAliveTest] cannot: delete
both `permitKeepAlive`
+/// calls from [GrpcMailboxServer] and those still pass, because they only
assert the values it read.
+///
+/// The enforcement is the half that hurts a live cluster when it is wrong. A
server whose
+/// `permitKeepAliveTime` is above the client's keep-alive time counts the
pings as "bad" and, past
+/// gRPC's two-strike threshold, answers `GOAWAY(ENHANCE_YOUR_CALM)` with
`too_many_pings` — dropping a
+/// mailbox channel mid-query. That is what makes the client interval
untunable without this knob.
+///
+/// Two servers are driven through one wait window: one permitting a fast
ping, one left at Netty's
+/// default. Only the second may lose its stream. Ablating the permit wiring
makes the *first*
+/// assertion fail, so the test cannot pass for the wrong reason.
+public class MailboxServerPermitKeepAliveBehaviorTest {
+ /// gRPC clamps a client keep-alive time up to its own 10s floor, so this is
the fastest ping a client
+ /// can be made to send, and the enforcement window is sized around it.
+ private static final int CLIENT_KEEP_ALIVE_TIME_MS = 10_000;
+ /// A permit below the client interval: every ping is then acceptable and no
strike is recorded.
+ private static final int PERMISSIVE_PERMIT_MS = 1_000;
+ /// Three pings at 10s plus slack. gRPC allows two bad pings and sends
`GOAWAY` on the third, so a
+ /// server that does not permit the rate gives up at roughly 30s.
+ private static final long WINDOW_MS = 60_000;
+
+ @Test(timeOut = 180_000)
+ public void testPermitKeepAliveTimeDecidesWhetherAFastPingerSurvives()
+ throws Exception {
+ MailboxService permissive = startMailboxService(PERMISSIVE_PERMIT_MS);
+ MailboxService restrictive = startMailboxService(-1); // leaves Netty's
5-minute default in place
+ ChannelManager channelManager = new ChannelManager(null, 4_000_000,
Duration.ofDays(365),
+ MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_HIGH_WATER_MARK_BYTES,
+ MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_LOW_WATER_MARK_BYTES,
+ // `withoutCalls` off, as in production: the open stream below is what
makes the client ping.
+ new GrpcKeepAliveConfig(CLIENT_KEEP_ALIVE_TIME_MS, 5_000, false));
+
+ ManagedChannel toPermissive = channelManager.getChannel("localhost",
permissive.getPort());
+ ManagedChannel toRestrictive = channelManager.getChannel("localhost",
restrictive.getPort());
+ try {
+ // No content is ever sent: `GrpcMailboxServer#open` accepts a stream
without the mailbox-id
+ // header and only reads it when a message arrives, so the stream stays
open with no handshake and
+ // the connection carries nothing but keep-alive pings.
+ OpenStream onPermissive = openStream(toPermissive);
+ OpenStream onRestrictive = openStream(toRestrictive);
+
+ assertTrue(onRestrictive.awaitTermination(WINDOW_MS),
+ "a server that does not permit the client's ping rate must end the
stream; without the "
+ + "permit knob this is what every mailbox channel would do once
the interval is tuned down");
+ Throwable rejection = onRestrictive._error.get();
+ assertNotNull(rejection);
+
assertTrue(String.valueOf(Status.fromThrowable(rejection).getDescription()).contains("too_many_pings"),
+ "expected GOAWAY(ENHANCE_YOUR_CALM) for too_many_pings, got: " +
rejection);
+
+ // Same client, same interval, same window — the only difference is the
permit.
+ assertNull(onPermissive._error.get(),
+ "the permitted server must keep the stream; if this fails the permit
is not reaching the "
+ + "gRPC server builder");
+ } finally {
+ toPermissive.shutdownNow();
+ toRestrictive.shutdownNow();
+ permissive.shutdown();
+ restrictive.shutdown();
+ }
+ }
+
+ private static MailboxService startMailboxService(int permitKeepAliveTimeMs)
{
+ PinotConfiguration config = new PinotConfiguration(Map.of(
+ MultiStageQueryRunner.KEY_OF_MAILBOX_SERVER_PERMIT_KEEP_ALIVE_TIME_MS,
permitKeepAliveTimeMs));
+ MailboxService mailboxService = new MailboxService(
+ "localhost", QueryTestUtils.getAvailablePort(), InstanceType.SERVER,
config);
+ mailboxService.start();
+ return mailboxService;
+ }
+
+ private static OpenStream openStream(ManagedChannel channel) {
+ OpenStream stream = new OpenStream();
+ PinotMailboxGrpc.newStub(channel).open(stream);
+ return stream;
+ }
+
+ /// Client-side observer of one `open` stream: records how it ended, if it
ended.
+ private static final class OpenStream implements
StreamObserver<Mailbox.MailboxStatus> {
+ private final CountDownLatch _terminated = new CountDownLatch(1);
+ private final AtomicReference<Throwable> _error = new AtomicReference<>();
+
+ @Override
+ public void onNext(Mailbox.MailboxStatus value) {
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ _error.compareAndSet(null, t);
+ _terminated.countDown();
+ }
+
+ @Override
+ public void onCompleted() {
+ _terminated.countDown();
+ }
+
+ boolean awaitTermination(long timeoutMs)
+ throws InterruptedException {
+ return _terminated.await(timeoutMs, TimeUnit.MILLISECONDS);
+ }
+ }
+}
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/channel/MailboxServerPermitKeepAliveTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/channel/MailboxServerPermitKeepAliveTest.java
new file mode 100644
index 00000000000..453ace90b0e
--- /dev/null
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/channel/MailboxServerPermitKeepAliveTest.java
@@ -0,0 +1,77 @@
+/**
+ * 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.
+ */
+package org.apache.pinot.query.mailbox.channel;
+
+import java.util.Map;
+import org.apache.pinot.query.mailbox.MailboxService;
+import org.apache.pinot.query.testutils.QueryTestUtils;
+import org.apache.pinot.spi.config.instance.InstanceType;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.MultiStageQueryRunner;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+
+
+/// Pins the keep-alive enforcement the mailbox server applies to its peers.
+///
+/// The two sides are coupled: a peer pinging faster than
`permitKeepAliveTime` has its pings counted as "bad" and,
+/// past gRPC's strike threshold, is answered with `GOAWAY(ENHANCE_YOUR_CALM)`
— dropping the mailbox channel
+/// mid-query. Both halves therefore have to be tunable, and the defaults have
to match Netty's own so that a peer
+/// still on the default client keep-alive time is never punished by an
upgraded one.
+public class MailboxServerPermitKeepAliveTest {
+
+ @Test
+ public void testPermitKeepAliveDefaultsMatchNettyDefaults() {
+ GrpcMailboxServer server = newServer(new PinotConfiguration(Map.of()));
+ assertEquals(server.getPermitKeepAliveTimeMs(),
+
MultiStageQueryRunner.DEFAULT_OF_MAILBOX_SERVER_PERMIT_KEEP_ALIVE_TIME_MS);
+ assertFalse(server.isPermitKeepAliveWithoutCalls());
+ }
+
+ /// Values an operator would use to tune detection of a silent peer down
from minutes to seconds. Both the client
+ /// keep-alive time and this permit have to move together, on every instance.
+ @Test
+ public void testPermitKeepAliveOverridesPickedUpFromConfig() {
+ GrpcMailboxServer server = newServer(new PinotConfiguration(Map.of(
+ MultiStageQueryRunner.KEY_OF_MAILBOX_SERVER_PERMIT_KEEP_ALIVE_TIME_MS,
30_000,
+
MultiStageQueryRunner.KEY_OF_MAILBOX_SERVER_PERMIT_KEEP_ALIVE_WITHOUT_CALLS,
true)));
+ assertEquals(server.getPermitKeepAliveTimeMs(), 30_000);
+ assertTrue(server.isPermitKeepAliveWithoutCalls());
+ }
+
+ /// A non-positive permit must leave Netty's own default in place rather
than being passed to gRPC, which rejects
+ /// it. Reaching construction at all is the assertion.
+ @Test
+ public void testNonPositivePermitLeavesNettyDefault() {
+ GrpcMailboxServer server = newServer(new PinotConfiguration(Map.of(
+ MultiStageQueryRunner.KEY_OF_MAILBOX_SERVER_PERMIT_KEEP_ALIVE_TIME_MS,
-1)));
+ assertEquals(server.getPermitKeepAliveTimeMs(), -1);
+ }
+
+ /// Builds the server without starting it: nothing binds a port until
`start()`, and the values under test are all
+ /// resolved in the constructor.
+ private static GrpcMailboxServer newServer(PinotConfiguration config) {
+ MailboxService mailboxService = new MailboxService(
+ "localhost", QueryTestUtils.getAvailablePort(), InstanceType.BROKER,
config);
+ return new GrpcMailboxServer(mailboxService, config, null, null, null);
+ }
+}
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
index 0ef4615ba20..42f66b844c0 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
@@ -2571,6 +2571,68 @@ public class CommonConstants {
public static final String KEY_OF_CHANNEL_IDLE_TIMEOUT_SECONDS =
"pinot.query.runner.channel.idle.timeout.seconds";
public static final long DEFAULT_CHANNEL_IDLE_TIMEOUT_SECONDS = -1;
+ /// gRPC keep-alive time for mailbox channels, in milliseconds. Values
> 0 enable keep-alive pings on the
+ /// server-to-server (and server-to-broker) data channels, so that a peer
that stopped answering *without*
+ /// closing its socket transitions the channel out of `READY` instead of
being sent to forever.
+ ///
+ /// The failure this addresses is a peer whose host is hung or unreachable
one-way: no `RST` is ever received, so
+ /// gRPC keeps the cached channel `READY`, every mailbox send parks until
the query deadline, and — because gRPC
+ /// only re-resolves DNS when a transport is dropped — a peer that has
come back at a new address is never
+ /// reached again. The symptom is every multi-stage query timing out while
the leaf stages complete in
+ /// milliseconds, for as long as the channel is cached, which is
indefinitely by default (see
+ /// [#KEY_OF_CHANNEL_IDLE_TIMEOUT_SECONDS]).
+ ///
+ /// Defaults are chosen to be safe against a peer running an older
version, whose mailbox server enforces
+ /// Netty's gRPC defaults (`permitKeepAliveTime` = 5 minutes,
`permitKeepAliveWithoutCalls` = false) and answers
+ /// a faster ping with `GOAWAY(ENHANCE_YOUR_CALM)`. Operators wanting
faster detection must lower
+ /// [#KEY_OF_MAILBOX_SERVER_PERMIT_KEEP_ALIVE_TIME_MS] on every peer first.
+ public static final String KEY_OF_CHANNEL_KEEP_ALIVE_TIME_MS =
"pinot.query.runner.channel.keep.alive.time.ms";
+ public static final int DEFAULT_CHANNEL_KEEP_ALIVE_TIME_MS = 300_000;
+
+ /// gRPC keep-alive timeout for mailbox channels, in milliseconds: how
long a keep-alive ping may go unanswered
+ /// before the transport is declared dead. Only applies when keep-alive is
enabled.
+ ///
+ /// Sized to survive a long stop-the-world pause on a loaded peer. A ping
unanswered for a few seconds is a GC
+ /// pause, not a dead host, and dropping the transport there would fail
healthy in-flight queries.
+ public static final String KEY_OF_CHANNEL_KEEP_ALIVE_TIMEOUT_MS =
+ "pinot.query.runner.channel.keep.alive.timeout.ms";
+ public static final int DEFAULT_CHANNEL_KEEP_ALIVE_TIMEOUT_MS = 30_000;
+
+ /// Whether to send gRPC keep-alive pings on mailbox channels even when
there are no active calls. Default is
+ /// `false` because Netty's default gRPC server rejects
pings-without-calls with `GOAWAY(ENHANCE_YOUR_CALM)`;
+ /// enabling it requires
[#KEY_OF_MAILBOX_SERVER_PERMIT_KEEP_ALIVE_WITHOUT_CALLS] on every peer.
+ ///
+ /// Leaving this off does not blind the ping to a dead peer. The ping
deadline is anchored to the last data
+ /// *received* on the transport, not to stream boundaries, so a channel
whose streams keep dying at their query
+ /// deadline still accumulates it; the ping is merely deferred until a
stream is open again, and then fires at
+ /// once. The difference the two settings make is who pays for the
recovery: with pings-without-calls off, the
+ /// transport is torn down by the next query to open a stream, and that
query fails; with it on, the tear-down
+ /// happens in the background between queries and no query sees the dead
channel at all.
+ public static final String KEY_OF_CHANNEL_KEEP_ALIVE_WITHOUT_CALLS =
+ "pinot.query.runner.channel.keep.alive.without.calls";
+ public static final boolean DEFAULT_CHANNEL_KEEP_ALIVE_WITHOUT_CALLS =
false;
+
+ /// Minimum interval, in milliseconds, between client gRPC keep-alive
pings that the mailbox server
+ /// ([org.apache.pinot.query.mailbox.channel.GrpcMailboxServer]) will
accept. Pings arriving more frequently than
+ /// this are counted as "bad pings"; once the server's internal threshold
is exceeded it sends
+ /// `GOAWAY(ENHANCE_YOUR_CALM)` with `too_many_pings` debug data and
closes the connection.
+ ///
+ /// Defaults to 5 minutes to match Netty's gRPC server default. Lowering
+ /// [#KEY_OF_CHANNEL_KEEP_ALIVE_TIME_MS] for faster detection of a silent
peer requires this to be less than or
+ /// equal to the configured client keep-alive time **on every instance the
channel may reach**, otherwise those
+ /// peers will tear the mailbox channel down. A non-positive value leaves
Netty's gRPC server default in place.
+ public static final String KEY_OF_MAILBOX_SERVER_PERMIT_KEEP_ALIVE_TIME_MS
=
+ "pinot.query.runner.mailbox.server.permit.keep.alive.time.ms";
+ public static final int
DEFAULT_OF_MAILBOX_SERVER_PERMIT_KEEP_ALIVE_TIME_MS = 300_000;
+
+ /// Whether the mailbox server permits client gRPC keep-alive pings when
there are no active RPCs on the
+ /// connection. Defaults to `false` to match Netty's gRPC server default.
Must be set to `true` on every peer if
+ /// [#KEY_OF_CHANNEL_KEEP_ALIVE_WITHOUT_CALLS] is enabled, otherwise those
peers will close idle channels with
+ /// `GOAWAY(ENHANCE_YOUR_CALM)`.
+ public static final String
KEY_OF_MAILBOX_SERVER_PERMIT_KEEP_ALIVE_WITHOUT_CALLS =
+ "pinot.query.runner.mailbox.server.permit.keep.alive.without.calls";
+ public static final boolean
DEFAULT_OF_MAILBOX_SERVER_PERMIT_KEEP_ALIVE_WITHOUT_CALLS = false;
+
/// Configuration for server port used to receive query plans.
public static final String KEY_OF_QUERY_SERVER_PORT =
"pinot.query.server.port";
public static final int DEFAULT_QUERY_SERVER_PORT = 0;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
