You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by ya...@apache.org on 2026/04/28 23:07:25 UTC
(pinot) branch master updated: Make MSE QueryServer's gRPC permit keep-alive settings configurable (#18354)
This is an automated email from the ASF dual-hosted git repository.
yashmayya 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 5b0b38cf82d Make MSE QueryServer's gRPC permit keep-alive settings configurable (#18354)
5b0b38cf82d is described below
commit 5b0b38cf82dda3d0766f47aeb4dbb3c28fe23c6d
Author: Yash Mayya <ya...@gmail.com>
AuthorDate: Tue Apr 28 16:07:18 2026 -0700
Make MSE QueryServer's gRPC permit keep-alive settings configurable (#18354)
---
.../pinot/query/service/server/QueryServer.java | 41 +++++++++++++++++-----
.../query/service/server/QueryServerTest.java | 25 +++++++++++++
.../apache/pinot/spi/utils/CommonConstants.java | 23 ++++++++++++
3 files changed, 81 insertions(+), 8 deletions(-)
diff --git a/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/server/QueryServer.java b/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/server/QueryServer.java
index 51fa6ba6e42..de88051cfb5 100644
--- a/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/server/QueryServer.java
+++ b/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/server/QueryServer.java
@@ -22,7 +22,6 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.protobuf.ByteString;
import com.google.protobuf.UnsafeByteOperations;
import io.grpc.Server;
-import io.grpc.ServerBuilder;
import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder;
import io.grpc.netty.shaded.io.netty.handler.ssl.SslContext;
import io.grpc.stub.StreamObserver;
@@ -94,6 +93,8 @@ public class QueryServer extends PinotQueryWorkerGrpc.PinotQueryWorkerImplBase {
@Nullable
private final QueryAccessControlFactory _accessControlFactory;
private final ThreadAccountant _threadAccountant;
+ private final int _permitKeepAliveTimeMs;
+ private final boolean _permitKeepAliveWithoutCalls;
// query submission service is only used for plan submission for now.
// TODO: with complex query submission logic we should allow asynchronous query submission return instead of
@@ -150,6 +151,12 @@ public class QueryServer extends PinotQueryWorkerGrpc.PinotQueryWorkerImplBase {
_accessControlFactory = accessControlFactory;
}
_threadAccountant = threadAccountant;
+ _permitKeepAliveTimeMs = serverConf.getProperty(
+ CommonConstants.MultiStageQueryRunner.KEY_OF_QUERY_SERVER_PERMIT_KEEP_ALIVE_TIME_MS,
+ CommonConstants.MultiStageQueryRunner.DEFAULT_OF_QUERY_SERVER_PERMIT_KEEP_ALIVE_TIME_MS);
+ _permitKeepAliveWithoutCalls = serverConf.getProperty(
+ CommonConstants.MultiStageQueryRunner.KEY_OF_QUERY_SERVER_PERMIT_KEEP_ALIVE_WITHOUT_CALLS,
+ CommonConstants.MultiStageQueryRunner.DEFAULT_OF_QUERY_SERVER_PERMIT_KEEP_ALIVE_WITHOUT_CALLS);
ExecutorService baseExecutorService =
ExecutorServiceUtils.create(serverConf, CommonConstants.Server.MULTISTAGE_SUBMISSION_EXEC_CONFIG_PREFIX,
@@ -173,14 +180,16 @@ public class QueryServer extends PinotQueryWorkerGrpc.PinotQueryWorkerImplBase {
LOGGER.info("Starting QueryServer");
try {
if (_server == null) {
- ServerBuilder<?> serverBuilder;
- if (_tlsConfig == null) {
- serverBuilder = ServerBuilder.forPort(_port);
- } else {
- serverBuilder = NettyServerBuilder.forPort(_port).sslContext(getOrCreateServerSslContext());
+ // Always use NettyServerBuilder so we can configure Netty-specific gRPC server options
+ // (e.g. permitKeepAliveTime / permitKeepAliveWithoutCalls).
+ NettyServerBuilder serverBuilder = NettyServerBuilder.forPort(_port);
+ if (_tlsConfig != null) {
+ serverBuilder.sslContext(getOrCreateServerSslContext());
}
_server = buildGrpcServer(serverBuilder);
- LOGGER.info("Initialized QueryServer on port: {}", _port);
+ LOGGER.info(
+ "Initialized QueryServer on port: {} with permitKeepAliveTimeMs: {}, permitKeepAliveWithoutCalls: {}",
+ _port, _permitKeepAliveTimeMs, _permitKeepAliveWithoutCalls);
}
_queryRunner.start();
_server.start();
@@ -189,12 +198,28 @@ public class QueryServer extends PinotQueryWorkerGrpc.PinotQueryWorkerImplBase {
}
}
- private <T extends ServerBuilder<T>> Server buildGrpcServer(ServerBuilder<T> builder) {
+ @VisibleForTesting
+ int getPermitKeepAliveTimeMs() {
+ return _permitKeepAliveTimeMs;
+ }
+
+ @VisibleForTesting
+ boolean isPermitKeepAliveWithoutCalls() {
+ return _permitKeepAliveWithoutCalls;
+ }
+
+ private Server buildGrpcServer(NettyServerBuilder builder) {
// By using directExecutor, GRPC doesn't need to manage its own thread pool
builder.directExecutor();
if (_accessControlFactory != null) {
builder.intercept(new AuthorizationInterceptor(_accessControlFactory));
}
+ // Configure server-side keep-alive enforcement so that operators tuning down the broker dispatch keep-alive time
+ // (or enabling pings-without-calls) do not get their channels torn down with GOAWAY(ENHANCE_YOUR_CALM).
+ if (_permitKeepAliveTimeMs > 0) {
+ builder.permitKeepAliveTime(_permitKeepAliveTimeMs, TimeUnit.MILLISECONDS);
+ }
+ builder.permitKeepAliveWithoutCalls(_permitKeepAliveWithoutCalls);
return builder.addService(this).maxInboundMessageSize(MAX_INBOUND_MESSAGE_SIZE).build();
}
diff --git a/pinot-query-runtime/src/test/java/org/apache/pinot/query/service/server/QueryServerTest.java b/pinot-query-runtime/src/test/java/org/apache/pinot/query/service/server/QueryServerTest.java
index af84711fadf..fd8020942f6 100644
--- a/pinot-query-runtime/src/test/java/org/apache/pinot/query/service/server/QueryServerTest.java
+++ b/pinot-query-runtime/src/test/java/org/apache/pinot/query/service/server/QueryServerTest.java
@@ -56,6 +56,7 @@ import org.apache.pinot.query.routing.WorkerMetadata;
import org.apache.pinot.query.runtime.QueryRunner;
import org.apache.pinot.query.testutils.QueryTestUtils;
import org.apache.pinot.spi.accounting.ThreadAccountantUtils;
+import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.metrics.PinotMetricUtils;
import org.apache.pinot.spi.query.QueryExecutionContext;
import org.apache.pinot.spi.query.QueryThreadContext;
@@ -73,6 +74,7 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.Mockito.*;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
@@ -123,6 +125,29 @@ public class QueryServerTest extends QueryTestSet {
MDC.clear();
}
+ @Test
+ public void testPermitKeepAliveDefaultsMatchNettyDefaults() {
+ // Defaults must match Netty's gRPC server defaults so that operators not configuring keep-alive on the broker
+ // dispatch side keep observing the same behavior as before this configurability was added.
+ QueryServer server = new QueryServer(QueryTestUtils.getAvailablePort(), mock(QueryRunner.class));
+ assertEquals(server.getPermitKeepAliveTimeMs(),
+ CommonConstants.MultiStageQueryRunner.DEFAULT_OF_QUERY_SERVER_PERMIT_KEEP_ALIVE_TIME_MS);
+ assertFalse(server.isPermitKeepAliveWithoutCalls());
+ }
+
+ @Test
+ public void testPermitKeepAliveOverridesPickedUpFromConfig() {
+ // Verify the config keys are wired to the server fields. Using values that mirror what an operator would set
+ // when tuning the broker-side dispatch keep-alive down (e.g. keepAliveTime=30s, keepAliveWithoutCalls=true).
+ Map<String, Object> overrides = new HashMap<>();
+ overrides.put(CommonConstants.MultiStageQueryRunner.KEY_OF_QUERY_SERVER_PERMIT_KEEP_ALIVE_TIME_MS, 30_000);
+ overrides.put(CommonConstants.MultiStageQueryRunner.KEY_OF_QUERY_SERVER_PERMIT_KEEP_ALIVE_WITHOUT_CALLS, true);
+ QueryServer server = new QueryServer(new PinotConfiguration(overrides), "serverId",
+ QueryTestUtils.getAvailablePort(), mock(QueryRunner.class), null, ThreadAccountantUtils.getNoOpAccountant());
+ assertEquals(server.getPermitKeepAliveTimeMs(), 30_000);
+ assertTrue(server.isPermitKeepAliveWithoutCalls());
+ }
+
@Test
public void testException()
throws Exception {
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 7a056576112..916ca208a4a 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
@@ -2237,6 +2237,29 @@ public class CommonConstants {
public static final String KEY_OF_DISPATCH_CHANNEL_KEEP_ALIVE_WITHOUT_CALLS =
"pinot.query.multistage.dispatch.channel.keep.alive.without.calls";
public static final boolean DEFAULT_OF_DISPATCH_CHANNEL_KEEP_ALIVE_WITHOUT_CALLS = false;
+
+ /// Minimum interval, in milliseconds, between client gRPC keep-alive pings that the MSE
+ /// [org.apache.pinot.query.service.server.QueryServer] 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. Operators tuning down the broker dispatch
+ /// `keepAliveTime` (see [#KEY_OF_DISPATCH_CHANNEL_KEEP_ALIVE_TIME_MS]) for faster silent-peer detection MUST set
+ /// this to a value less than or equal to the configured client keep-alive time, otherwise the server will tear
+ /// down the dispatch channel. A non-positive value leaves Netty's gRPC server default (currently 5 minutes) in
+ /// place.
+ public static final String KEY_OF_QUERY_SERVER_PERMIT_KEEP_ALIVE_TIME_MS =
+ "pinot.query.multistage.query.server.permit.keep.alive.time.ms";
+ public static final int DEFAULT_OF_QUERY_SERVER_PERMIT_KEEP_ALIVE_TIME_MS = 300_000;
+
+ /// Whether the MSE [org.apache.pinot.query.service.server.QueryServer] 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` if brokers configure
+ /// [#KEY_OF_DISPATCH_CHANNEL_KEEP_ALIVE_WITHOUT_CALLS] to `true`, otherwise the server will close idle channels
+ /// with `GOAWAY(ENHANCE_YOUR_CALM)`.
+ public static final String KEY_OF_QUERY_SERVER_PERMIT_KEEP_ALIVE_WITHOUT_CALLS =
+ "pinot.query.multistage.query.server.permit.keep.alive.without.calls";
+ public static final boolean DEFAULT_OF_QUERY_SERVER_PERMIT_KEEP_ALIVE_WITHOUT_CALLS = false;
}
public static class NullValuePlaceHolder {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org