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 6aaa0405566 Add broker startup pre-connect for broker-to-server
channels (SSE) (#19407)
6aaa0405566 is described below
commit 6aaa0405566c576a51b64cb4a031033bb2409eba
Author: Jinesh Parakh <[email protected]>
AuthorDate: Mon Sep 7 20:34:44 2026 +0530
Add broker startup pre-connect for broker-to-server channels (SSE) (#19407)
---
.../broker/broker/helix/BaseBrokerStarter.java | 126 ++++++-
.../requesthandler/BrokerRequestHandler.java | 10 +
.../BrokerRequestHandlerDelegate.java | 7 +
.../broker/requesthandler/ServerPreConnector.java | 196 ++++++++++
.../SingleConnectionBrokerRequestHandler.java | 46 +++
.../routing/manager/BaseBrokerRoutingManager.java | 5 +
.../manager/MultiClusterRoutingManager.java | 12 +
.../requesthandler/ServerPreConnectorTest.java | 406 +++++++++++++++++++++
.../manager/MultiClusterRoutingManagerTest.java | 14 +
.../apache/pinot/common/metrics/BrokerTimer.java | 9 +
.../apache/pinot/core/routing/RoutingManager.java | 11 +
.../apache/pinot/core/transport/QueryRouter.java | 26 ++
.../pinot/core/transport/ServerChannels.java | 136 ++++++-
.../pinot/core/transport/ServerChannelsTest.java | 95 +++++
.../BrokerServerPreConnectIntegrationTest.java | 142 +++++++
.../integration/tests/TlsIntegrationTest.java | 29 ++
.../apache/pinot/spi/utils/CommonConstants.java | 15 +
17 files changed, 1271 insertions(+), 14 deletions(-)
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java
index 5ed947953e3..e02549da2e9 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java
@@ -142,6 +142,10 @@ public abstract class BaseBrokerStarter implements
ServiceStartable {
/// desynchronize brokers on shared storage.
private static final int RESPONSE_STORE_CLEANUP_INITIAL_DELAY_JITTER_DIVISOR
= 4;
+ /// How often the pre-connect thread re-checks whether Helix has converged.
Short enough not to add
+ /// meaningful delay to a fast startup, long enough not to hammer the Helix
data accessor.
+ private static final long HELIX_CONVERGENCE_POLL_INTERVAL_MS = 200L;
+
protected PinotConfiguration _brokerConf;
protected List<ListenerConfig> _listenerConfigs;
protected String _clusterName;
@@ -189,6 +193,17 @@ public abstract class BaseBrokerStarter implements
ServiceStartable {
protected BrokerGrpcServer _brokerGrpcServer;
protected FailureDetector _failureDetector;
protected ThreadAccountant _threadAccountant;
+ /// The Helix-convergence half of the service-status composite, held so
startup pre-connect can wait
+ /// on exactly that signal -- convergence is the point at which routing, and
the servers it
+ /// references, first exist.
+ @Nullable
+ private volatile ServiceStatus.ServiceStatusCallback
_helixConvergenceCallback;
+ /// The background pre-connect thread, tracked so shutdown can interrupt it.
+ @Nullable
+ private volatile Thread _preConnectThread;
+ /// Whether startup pre-connect is enabled, and its budget. Read once in
`start()`.
+ private boolean _preConnectEnabled;
+ private long _preConnectTimeoutMs;
@Override
public void init(PinotConfiguration brokerConf)
@@ -669,10 +684,25 @@ public abstract class BaseBrokerStarter implements
ServiceStartable {
_brokerConf.getProperty(CommonConstants.Groovy.GROOVY_QUERY_STATIC_ANALYZER_CONFIG,
_brokerConf.getProperty(CommonConstants.Groovy.GROOVY_ALL_STATIC_ANALYZER_CONFIG)));
- // Register the service status handler
+ // Only the Netty single-stage transport has broker-to-server channels to
open; the gRPC single-stage
+ // handler, the multi-stage engine and the time-series path all use
different transports and are
+ // unaffected by this flag.
+ _preConnectEnabled =
_brokerConf.getProperty(Broker.CONFIG_OF_BROKER_STARTUP_PRECONNECT_ENABLED,
+ Broker.DEFAULT_BROKER_STARTUP_PRECONNECT_ENABLED);
+ _preConnectTimeoutMs =
_brokerConf.getProperty(Broker.CONFIG_OF_BROKER_STARTUP_PRECONNECT_TIMEOUT_MS,
+ Broker.DEFAULT_BROKER_STARTUP_PRECONNECT_TIMEOUT_MS);
registerServiceStatusHandler();
-
- _isStarting = false;
+ if (_preConnectEnabled) {
+ // Startup is not finished until the broker-to-server channels are open,
so `_isStarting` stays set
+ // and the existing lifecycle callback keeps reporting STARTING -- no
query is routed here before the
+ // connect and TLS handshake have been paid. "Still pre-connecting" is
not a new kind of statement,
+ // it is the same one, so it reuses the same flag rather than a second
parallel gate. The flag was
+ // set before the status handler was registered, so there is
structurally no window in which
+ // readiness is granted un-gated. The pre-connect thread clears it when
it finishes.
+ startPreConnect();
+ } else {
+ _isStarting = false;
+ }
_brokerMetrics.addTimedValue(BrokerTimer.STARTUP_SUCCESS_DURATION_MS,
System.currentTimeMillis() - startTimeMs, TimeUnit.MILLISECONDS);
@@ -852,13 +882,90 @@ public abstract class BaseBrokerStarter implements
ServiceStartable {
Broker.DEFAULT_BROKER_MIN_RESOURCE_PERCENT_FOR_START);
LOGGER.info("Registering service status handler");
- ServiceStatus.setServiceStatusCallback(_instanceId, new
ServiceStatus.MultipleCallbackServiceStatusCallback(
- List.of(
+ // The two Helix callbacks are grouped into their own composite so startup
pre-connect can wait on
+ // exactly the "Helix has converged" signal. CurrentState only reports
ONLINE once the
+ // OFFLINE->ONLINE transition has returned, and that transition is what
builds routing -- so
+ // convergence is the precondition for routing entries, and the servers
they reference, existing.
+ // Behaviour is unchanged: MultipleCallbackServiceStatusCallback surfaces
the first non-GOOD
+ // callback, so nesting the two Helix callbacks reports the same status as
listing them flat.
+ _helixConvergenceCallback = new
ServiceStatus.MultipleCallbackServiceStatusCallback(List.of(
+ new
ServiceStatus.IdealStateAndCurrentStateMatchServiceStatusCallback(_participantHelixManager,
+ _clusterName, _instanceId, resourcesToMonitor,
minResourcePercentForStartup),
+ new
ServiceStatus.IdealStateAndExternalViewMatchServiceStatusCallback(_participantHelixManager,
+ _clusterName, _instanceId, resourcesToMonitor,
minResourcePercentForStartup)));
+
+ ServiceStatus.setServiceStatusCallback(_instanceId,
+ new ServiceStatus.MultipleCallbackServiceStatusCallback(List.of(
new ServiceStatus.LifecycleServiceStatusCallback(this::isStarting,
this::isShuttingDown),
- new
ServiceStatus.IdealStateAndCurrentStateMatchServiceStatusCallback(_participantHelixManager,
- _clusterName, _instanceId, resourcesToMonitor,
minResourcePercentForStartup),
- new
ServiceStatus.IdealStateAndExternalViewMatchServiceStatusCallback(_participantHelixManager,
- _clusterName, _instanceId, resourcesToMonitor,
minResourcePercentForStartup))));
+ _helixConvergenceCallback)));
+ }
+
+ /// Runs startup server pre-connect on a background thread and ends startup
([#_isStarting]) when it
+ /// finishes. Asynchronous so `start()` still returns promptly -- readiness
is withheld through
+ /// `ServiceStatus`, not by blocking startup. The flag is cleared in a
`finally` so startup ends even if
+ /// pre-connect throws or is interrupted: readiness withheld indefinitely
would stall a rolling restart,
+ /// a worse failure than serving a broker whose channels are not yet warm.
+ ///
+ /// Only called when pre-connect is enabled; otherwise `start()` ends
startup itself and readiness
+ /// behaves exactly as before.
+ private void startPreConnect() {
+ _preConnectThread = new Thread(() -> {
+ // Set once Helix converges. Both the budget and the duration metric are
measured from here, not
+ // from thread start, so the deliberately unbounded convergence wait is
charged against neither: the
+ // Helix callbacks withhold readiness until convergence anyway, so it
costs nothing.
+ long preConnectStartMs = 0L;
+ try {
+ long threadStartMs = System.currentTimeMillis();
+ awaitHelixConvergence();
+ preConnectStartMs = System.currentTimeMillis();
+ LOGGER.info("Helix converged after {} ms; pre-connecting server
channels",
+ preConnectStartMs - threadStartMs);
+ int connected =
_brokerRequestHandler.preConnectServers(preConnectStartMs +
_preConnectTimeoutMs);
+ LOGGER.info("Startup server pre-connect opened {} channel(s); ending
startup", connected);
+ } catch (InterruptedException e) {
+ // Normal on shutdown; stopPreConnect() interrupts us.
+ Thread.currentThread().interrupt();
+ LOGGER.info("Startup server pre-connect interrupted before completion;
ending startup");
+ } catch (Throwable t) {
+ LOGGER.warn("Startup server pre-connect threw; ending startup anyway",
t);
+ } finally {
+ _isStarting = false;
+ // Record the duration only if convergence was reached, so the metric
measures the pre-connect work
+ // itself and never the (unbounded) convergence wait -- e.g. when
shutdown interrupts the wait.
+ if (preConnectStartMs > 0L) {
+
_brokerMetrics.addTimedValue(BrokerTimer.STARTUP_PRECONNECT_DURATION_MS,
+ System.currentTimeMillis() - preConnectStartMs,
TimeUnit.MILLISECONDS);
+ }
+ }
+ }, "broker-startup-preconnect");
+ _preConnectThread.setDaemon(true);
+ _preConnectThread.start();
+ }
+
+ /// Blocks until the Helix-convergence callbacks report GOOD -- the point at
which routing entries and
+ /// the servers they reference exist. Deliberately **unbounded** and
interruptible: a broker that never
+ /// converges is never Ready regardless of pre-connect, and shutdown
interrupts this thread. Monitors
+ /// `brokerResource` only (partitions in {OFFLINE, ONLINE, DROPPED});
segment states live in the table
+ /// resources that servers monitor and cannot hold this up.
+ private void awaitHelixConvergence()
+ throws InterruptedException {
+ ServiceStatus.ServiceStatusCallback callback = _helixConvergenceCallback;
+ if (callback == null) {
+ return;
+ }
+ while (callback.getServiceStatus() != ServiceStatus.Status.GOOD) {
+ Thread.sleep(HELIX_CONVERGENCE_POLL_INTERVAL_MS);
+ }
+ }
+
+ /// Interrupts an in-flight pre-connect so shutdown never waits on it. Best
effort: the thread is a
+ /// daemon and records its metric in a `finally` regardless.
+ private void stopPreConnect() {
+ Thread thread = _preConnectThread;
+ if (thread != null && thread.isAlive()) {
+ LOGGER.info("Interrupting in-flight startup server pre-connect for
shutdown");
+ thread.interrupt();
+ }
}
private String getDefaultBrokerId() {
@@ -893,6 +1000,7 @@ public abstract class BaseBrokerStarter implements
ServiceStartable {
public void stop() {
LOGGER.info("Shutting down Pinot broker");
_isShuttingDown = true;
+ stopPreConnect();
LOGGER.info("Disconnecting participant Helix manager");
_participantHelixManager.disconnect();
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BrokerRequestHandler.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BrokerRequestHandler.java
index dc4a5624f42..c9dcc497277 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BrokerRequestHandler.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BrokerRequestHandler.java
@@ -47,6 +47,16 @@ public interface BrokerRequestHandler {
void shutDown();
+ /// Opens broker-to-server channels ahead of traffic so the first real query
does not pay the blocking
+ /// connect -- and, when broker-to-server TLS is on, the handshake -- on its
critical path. Called once
+ /// at startup after Helix has converged, when
`pinot.broker.startup.preconnect.enabled` is set.
+ ///
+ /// Only the single-connection SSE handler opens Netty channels, so the
default is a no-op. Returns the
+ /// number of channels connected before `deadlineMs` (an absolute
[System#currentTimeMillis] value).
+ default int preConnectServers(long deadlineMs) {
+ return 0;
+ }
+
BrokerResponse handleRequest(JsonNode request, @Nullable SqlNodeAndOptions
sqlNodeAndOptions,
@Nullable RequesterIdentity requesterIdentity, RequestContext
requestContext, @Nullable HttpHeaders httpHeaders)
throws Exception;
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BrokerRequestHandlerDelegate.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BrokerRequestHandlerDelegate.java
index 1835030436e..bca292de883 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BrokerRequestHandlerDelegate.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BrokerRequestHandlerDelegate.java
@@ -76,6 +76,13 @@ public class BrokerRequestHandlerDelegate implements
BrokerRequestHandler {
}
}
+ @Override
+ public int preConnectServers(long deadlineMs) {
+ // Only the single-stage handler owns the broker-to-server Netty channels;
the multi-stage (gRPC)
+ // and time-series paths have nothing to pre-connect here.
+ return _singleStageBrokerRequestHandler.preConnectServers(deadlineMs);
+ }
+
@Override
public void shutDown() {
_singleStageBrokerRequestHandler.shutDown();
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/ServerPreConnector.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/ServerPreConnector.java
new file mode 100644
index 00000000000..c90b65e9f0b
--- /dev/null
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/ServerPreConnector.java
@@ -0,0 +1,196 @@
+/**
+ * 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.broker.requesthandler;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.CompletionService;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorCompletionService;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+import javax.annotation.concurrent.ThreadSafe;
+import org.apache.pinot.core.transport.ServerInstance;
+import org.apache.pinot.spi.config.table.TableType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Opens broker-to-server Netty channels ahead of query traffic, so the first
real query does not pay
+/// the blocking `connect()` -- and, when broker-to-server TLS is on, the
handshake -- on its critical
+/// path.
+///
+/// A channel's identity includes the table type, so OFFLINE and REALTIME are
**separate** channels
+/// (separate sockets, separate handshakes) to the same physical server. The
caller supplies the exact
+/// (server, table type) pairs to open, derived from what this broker actually
routes -- so an offline-only
+/// cluster opens no REALTIME channels, and a broker serving one tenant does
not connect to another
+/// tenant's servers. A table that lands on a server later is left to the lazy
connect path (one query pays
+/// the connect) rather than pre-warmed here on the chance it appears: the
would-be second channel shares
+/// nothing with the first, so pre-warming it amortizes nothing. Connecting an
already-active channel is a
+/// no-op, so this is safe to call more than once.
+///
+/// Bounded so it can never stall startup: a capped thread pool, a per-channel
connect bound derived from
+/// the remaining budget, a per-channel wait clamped to the caller's deadline,
and a straggler grace window
+/// ([#STRAGGLER_GRACE_MS]) that, once at least one channel is up, releases
the caller when the rest stop
+/// arriving rather than waiting out the whole budget on one stuck server. A
server that is unreachable or
+/// itself restarting is logged and skipped -- the existing lazy-connect path
still serves it. This class
+/// is stateless and thread-safe.
+///
+/// It takes its dependencies as functions rather than concrete
`RoutingManager`/`QueryRouter` types so
+/// the parallelism, budget and failure handling can be unit-tested without a
live broker.
+@ThreadSafe
+public class ServerPreConnector {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(ServerPreConnector.class);
+
+ /// Cap on the connect thread pool: a large tenant must not spawn a thread
per server. Safe to exceed
+ /// the core count even on a 2- or 4-vCPU broker: each task is blocking
connect + TLS handshake (mostly
+ /// network wait, with the actual I/O on Netty's event loop), and this runs
during startup before any
+ /// query load, so the threads are almost entirely parked rather than
contending for CPU.
+ ///
+ /// It is a throughput cap, not a safety bound: with more channels than
threads the surplus queues
+ /// behind the workers, so the budget alone must not be what stops a stuck
connect. That is why
+ /// [ChannelConnector] takes a per-channel timeout.
+ @VisibleForTesting
+ static final int MAX_CONNECT_THREADS = 16;
+
+ /// How long to keep waiting, once at least one channel is up, for the next
one before concluding the
+ /// rest are stuck. A quiet window this long means what is left is stuck
rather than merely slow, so the
+ /// caller is released and the stragglers finish -- or time out -- on their
own daemon threads. Until the
+ /// first *successful* connect the whole budget is available: with nothing
up yet there is no way to tell
+ /// "every server is slow" from "a few are stuck", and a fast failure must
not start the clock.
+ @VisibleForTesting
+ static final long STRAGGLER_GRACE_MS = 2_000L;
+
+ /// Opens one broker-to-server channel. Implementations must bound their own
wait by `timeoutMs` and
+ /// must not throw; the return value reports whether the channel is
connected.
+ @FunctionalInterface
+ public interface ChannelConnector {
+ boolean connect(ServerInstance serverInstance, TableType tableType, long
timeoutMs);
+ }
+
+ /// A (server, table type) channel to open. The table type is part of the
channel identity: OFFLINE and
+ /// REALTIME are separate channels (separate sockets) to the same physical
server.
+ public record ChannelTarget(ServerInstance serverInstance, TableType
tableType) {
+ }
+
+ private final Supplier<Collection<ChannelTarget>> _targetsSupplier;
+ private final ChannelConnector _connector;
+
+ /// @param targetsSupplier supplies the (server, table type) channels to
open, evaluated once per
+ /// [#preConnect] call after the caller has ensured routing is built.
Derive these from routing so
+ /// only channels this broker actually uses are opened. Must return a
non-null collection and must
+ /// not throw (it is evaluated before the failure-handling loop); the
production supplier reads
+ /// routing, which cannot do either.
+ /// @param connector opens the channel for one (server, table type) within a
timeout
+ public ServerPreConnector(Supplier<Collection<ChannelTarget>>
targetsSupplier, ChannelConnector connector) {
+ _targetsSupplier = targetsSupplier;
+ _connector = connector;
+ }
+
+ /// Opens the supplied (server, table type) channels in parallel, bounded by
`deadlineMs` (an absolute
+ /// [System#currentTimeMillis] value). Returns the number of channels
connected **before the caller was
+ /// released** -- so a straggler that connects after the grace window (see
[#STRAGGLER_GRACE_MS]) is not
+ /// counted, even though its channel is still published for the first query
to reuse. Never throws: a
+ /// channel that fails or times out is logged and skipped.
+ public int preConnect(long deadlineMs) {
+ // Snapshot the target view once. The supplier may derive from a live
routing view that another thread
+ // updates during startup; snapshotting keeps the channel count consistent
with the tasks actually
+ // submitted below, so we never poll for phantom channels or under-count
real ones.
+ List<ChannelTarget> targets = new ArrayList<>(_targetsSupplier.get());
+ if (targets.isEmpty() || System.currentTimeMillis() >= deadlineMs) {
+ return 0;
+ }
+ long startMs = System.currentTimeMillis();
+ int channelCount = targets.size();
+ ExecutorService executor =
Executors.newFixedThreadPool(Math.min(channelCount, MAX_CONNECT_THREADS),
+ new
ThreadFactoryBuilder().setNameFormat("broker-preconnect-%d").setDaemon(true).build());
+ // A completion service hands channels back in the order they finish, not
the order submitted, so a
+ // slow or unreachable server never delays the counting of faster ones
that finished behind it. This
+ // removes head-of-line blocking from the *counting* only: with more
channels than workers the surplus
+ // still queues for a worker, which is what the per-channel timeout bounds.
+ CompletionService<Boolean> completionService = new
ExecutorCompletionService<>(executor);
+ int connected = 0;
+ boolean releasedEarly = false;
+ try {
+ for (ChannelTarget target : targets) {
+ completionService.submit(() ->
_connector.connect(target.serverInstance(), target.tableType(),
+ Math.max(0L, deadlineMs - System.currentTimeMillis())));
+ }
+ for (int i = 0; i < channelCount; i++) {
+ long remainingMs = deadlineMs - System.currentTimeMillis();
+ if (remainingMs <= 0) {
+ break;
+ }
+ // Keep the whole budget available until the first *successful*
connect: a quiet window only means
+ // "the rest are stuck" once at least one channel has actually come
up. Keying the exemption on the
+ // first completion instead would let a single fast event -- an
instantly refused connect, or one
+ // nearby server -- start the grace clock before the
healthy-but-slower channels return, abandoning
+ // them. Once one channel is up, a quiet grace window is the signal
that what is left is stuck rather
+ // than slow, and the caller is released -- one unreachable server
otherwise holds the gate for the
+ // entire budget. (A cluster where no server ever connects still exits
promptly when connects fail
+ // fast, and waits the budget only when every connect black-holes,
which is the correct thing to do
+ // for a broker that can reach nothing.)
+ long waitMs = connected == 0 ? remainingMs : Math.min(remainingMs,
STRAGGLER_GRACE_MS);
+ try {
+ Future<Boolean> future = completionService.poll(waitMs,
TimeUnit.MILLISECONDS);
+ if (future == null) {
+ // The grace cap was binding (we could have waited longer but
chose not to) only when waitMs was
+ // clamped below the remaining budget; otherwise this is plain
budget exhaustion.
+ releasedEarly = waitMs < remainingMs;
+ LOGGER.info("No pre-connect channel completed in {} ms with {}/{}
still outstanding; releasing startup "
+ + "and leaving them to the lazy connect path", waitMs,
channelCount - i, channelCount);
+ break;
+ }
+ if (Boolean.TRUE.equals(future.get())) {
+ connected++;
+ }
+ } catch (InterruptedException e) {
+ // Shutdown: stopPreConnect() interrupts us. Restore the flag and
stop promptly.
+ Thread.currentThread().interrupt();
+ break;
+ } catch (ExecutionException e) {
+ // A server that is unreachable or itself restarting must not block
startup.
+ LOGGER.debug("Pre-connect did not complete for one channel", e);
+ }
+ }
+ } finally {
+ // shutdown(), not shutdownNow(): a channel we stopped waiting on is
still connecting on a daemon
+ // thread. Interrupting a worker parked in connect().sync() abandons a
ChannelFuture that can still
+ // complete, leaking a socket nobody references or closes. Letting the
workers run means a late
+ // channel is still published for the first query to reuse, and each
task is already bounded by its
+ // own deadline-derived timeout, so none outlives the budget.
+ executor.shutdown();
+ }
+ long elapsedMs = System.currentTimeMillis() - startMs;
+ if (connected < channelCount) {
+ LOGGER.warn("Broker pre-connected {}/{} channel(s) in {} ms ({}); the
rest fall back to the lazy connect "
+ + "path", connected, channelCount, elapsedMs, releasedEarly ?
"released early on a straggler grace window"
+ : "budget elapsed");
+ } else {
+ LOGGER.info("Broker pre-connected {}/{} channel(s) in {} ms", connected,
channelCount, elapsedMs);
+ }
+ return connected;
+ }
+}
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/SingleConnectionBrokerRequestHandler.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/SingleConnectionBrokerRequestHandler.java
index b9d3ad23e19..4fd4ef47bac 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/SingleConnectionBrokerRequestHandler.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/SingleConnectionBrokerRequestHandler.java
@@ -21,9 +21,12 @@ package org.apache.pinot.broker.requesthandler;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Maps;
import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
import java.util.IdentityHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;
@@ -53,6 +56,7 @@ import org.apache.pinot.core.transport.ServerRoutingInstance;
import
org.apache.pinot.core.transport.server.routing.stats.ServerRoutingStatsManager;
import org.apache.pinot.materializedview.handler.MaterializedViewHandler;
import org.apache.pinot.spi.accounting.ThreadAccountant;
+import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.exception.QueryErrorCode;
import org.apache.pinot.spi.exception.QueryException;
@@ -113,6 +117,48 @@ public class SingleConnectionBrokerRequestHandler extends
BaseSingleStageBrokerR
_brokerReduceService.shutDown();
}
+ /// Opens the broker-to-server channels this broker actually routes to,
taking the blocking connect --
+ /// and, when broker-to-server TLS is on, the handshake -- off the first
real query's critical path. The
+ /// channels opened are derived from routing (see
[#routableChannelTargets]), so an offline-only cluster
+ /// opens no REALTIME channels and a broker serving one tenant does not
connect to another tenant's
+ /// servers. The caller guarantees Helix has converged, so routing reflects
the tables and servers this
+ /// broker serves. Never throws; returns the number of channels connected
before `deadlineMs`.
+ @Override
+ public int preConnectServers(long deadlineMs) {
+ return new ServerPreConnector(() ->
routableChannelTargets(_routingManager),
+ _queryRouter::preConnect).preConnect(deadlineMs);
+ }
+
+ /// Derives the (server, table type) channels this broker actually routes
to, so pre-connect opens
+ /// exactly those. Iterates the broker's routable tables (each name carries
its type) and resolves each
+ /// table's serving instances to `ServerInstance`s, deduping across tables.
This deliberately does **not**
+ /// use `getRoutableServerInstanceMap()` as the server set: that is every
enabled server in the whole
+ /// cluster, not this broker's, and crossing it with both table types would
open a duplicate socket per
+ /// server (OFFLINE and REALTIME are separate channels) plus channels to
servers this broker never
+ /// queries. Static and package-private for unit testing against a mocked
[RoutingManager].
+ @VisibleForTesting
+ static Collection<ServerPreConnector.ChannelTarget>
routableChannelTargets(RoutingManager routingManager) {
+ Map<String, ServerInstance> serverInstanceMap =
routingManager.getRoutableServerInstanceMap();
+ Set<ServerPreConnector.ChannelTarget> targets = new HashSet<>();
+ for (String tableNameWithType : routingManager.getRoutableTables()) {
+ TableType tableType =
TableNameBuilder.getTableTypeFromTableName(tableNameWithType);
+ if (tableType == null) {
+ continue;
+ }
+ Set<String> servingInstances =
routingManager.getServingInstances(tableNameWithType);
+ if (servingInstances == null) {
+ continue;
+ }
+ for (String instanceId : servingInstances) {
+ ServerInstance serverInstance = serverInstanceMap.get(instanceId);
+ if (serverInstance != null) {
+ targets.add(new ServerPreConnector.ChannelTarget(serverInstance,
tableType));
+ }
+ }
+ }
+ return targets;
+ }
+
@Override
protected BrokerResponseNative processBrokerRequest(long requestId,
BrokerRequest originalBrokerRequest,
BrokerRequest serverBrokerRequest, TableRouteInfo route, long timeoutMs,
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java
index dfde3e9c950..db3f106b2e8 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java
@@ -1269,6 +1269,11 @@ public abstract class BaseBrokerRoutingManager
implements RoutingManager, Cluste
return _routableServerInstanceMap;
}
+ @Override
+ public Set<String> getRoutableTables() {
+ return Set.copyOf(_routingEntryMap.keySet());
+ }
+
/// Must be called under `_globalLock.writeLock()` when rebuilding
`_routableServerInstanceMap` from a freshly
/// computed set of routable server IDs.
@GuardedBy("_globalLock.writeLock()")
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/MultiClusterRoutingManager.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/MultiClusterRoutingManager.java
index 9f04a4ca3d4..538b85dd416 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/MultiClusterRoutingManager.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/MultiClusterRoutingManager.java
@@ -167,6 +167,18 @@ public class MultiClusterRoutingManager implements
RoutingManager {
return combined;
}
+ @Override
+ public Set<String> getRoutableTables() {
+ // Union across clusters so startup pre-connect derives channels for every
cluster this broker routes
+ // to; without this override the RoutingManager default returns empty and
pre-connect would silently do
+ // nothing on a multi-cluster broker.
+ Set<String> combined = new
HashSet<>(_localClusterRoutingManager.getRoutableTables());
+ for (BaseBrokerRoutingManager remoteCluster :
_remoteClusterRoutingManagers) {
+ combined.addAll(remoteCluster.getRoutableTables());
+ }
+ return combined;
+ }
+
@Override
public TablePartitionInfo getTablePartitionInfo(String tableNameWithType) {
return findFirst(mgr -> mgr.getTablePartitionInfo(tableNameWithType),
tableNameWithType);
diff --git
a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/ServerPreConnectorTest.java
b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/ServerPreConnectorTest.java
new file mode 100644
index 00000000000..00db6cad8fc
--- /dev/null
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/ServerPreConnectorTest.java
@@ -0,0 +1,406 @@
+/**
+ * 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.broker.requesthandler;
+
+import java.util.ArrayList;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.pinot.broker.requesthandler.ServerPreConnector.ChannelTarget;
+import org.apache.pinot.core.routing.RoutingManager;
+import org.apache.pinot.core.transport.ServerInstance;
+import org.apache.pinot.spi.config.table.TableType;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+
+
+public class ServerPreConnectorTest {
+ private static final long ONE_MINUTE_MS = 60_000L;
+
+ private static List<ServerInstance> mockServers(int count) {
+ List<ServerInstance> servers = new ArrayList<>(count);
+ for (int i = 0; i < count; i++) {
+ servers.add(mock(ServerInstance.class));
+ }
+ return servers;
+ }
+
+ /// The (server, table type) channels the caller would derive from routing:
here just the cross product
+ /// of the given servers and types, so the connector-behaviour tests can
exercise a known channel count.
+ private static List<ChannelTarget> targets(List<ServerInstance> servers,
TableType... types) {
+ List<ChannelTarget> targets = new ArrayList<>(servers.size() *
types.length);
+ for (ServerInstance server : servers) {
+ for (TableType type : types) {
+ targets.add(new ChannelTarget(server, type));
+ }
+ }
+ return targets;
+ }
+
+ private static long farDeadline() {
+ return System.currentTimeMillis() + ONE_MINUTE_MS;
+ }
+
+ @Test
+ public void connectsEverySuppliedTarget() {
+ List<ServerInstance> servers = mockServers(3);
+ Set<TableType> tableTypesSeen = ConcurrentHashMap.newKeySet();
+ Set<Integer> serversSeen = ConcurrentHashMap.newKeySet();
+ AtomicInteger calls = new AtomicInteger();
+
+ int connected = new ServerPreConnector(() -> targets(servers,
TableType.OFFLINE, TableType.REALTIME),
+ (server, tableType, timeoutMs) -> {
+ calls.incrementAndGet();
+ tableTypesSeen.add(tableType);
+ serversSeen.add(System.identityHashCode(server));
+ return true;
+ }).preConnect(farDeadline());
+
+ // 3 servers x 2 table types.
+ assertEquals(connected, 6);
+ assertEquals(calls.get(), 6);
+ assertEquals(serversSeen.size(), 3);
+ assertEquals(tableTypesSeen, EnumSet.of(TableType.OFFLINE,
TableType.REALTIME));
+ }
+
+ @Test
+ public void emptyTargetsReturnsZeroWithoutConnecting() {
+ AtomicInteger calls = new AtomicInteger();
+ int connected = new ServerPreConnector(List::of, (server, tableType,
timeoutMs) -> {
+ calls.incrementAndGet();
+ return true;
+ }).preConnect(farDeadline());
+
+ assertEquals(connected, 0);
+ assertEquals(calls.get(), 0);
+ }
+
+ @Test
+ public void deadlineAlreadyPassedReturnsZeroWithoutConnecting() {
+ List<ServerInstance> servers = mockServers(2);
+ AtomicInteger calls = new AtomicInteger();
+ int connected = new ServerPreConnector(() -> targets(servers,
TableType.OFFLINE, TableType.REALTIME),
+ (server, tableType, timeoutMs) -> {
+ calls.incrementAndGet();
+ return true;
+ }).preConnect(System.currentTimeMillis() - 1);
+
+ assertEquals(connected, 0);
+ assertEquals(calls.get(), 0);
+ }
+
+ @Test
+ public void countsOnlySuccessfulConnects() {
+ List<ServerInstance> servers = mockServers(4);
+ // OFFLINE succeeds, REALTIME fails: exactly one successful channel per
server.
+ int connected = new ServerPreConnector(() -> targets(servers,
TableType.OFFLINE, TableType.REALTIME),
+ (server, tableType, timeoutMs) -> tableType ==
TableType.OFFLINE).preConnect(farDeadline());
+
+ assertEquals(connected, 4);
+ }
+
+ @Test
+ public void connectFailureIsSwallowedAndOthersStillConnect() {
+ List<ServerInstance> servers = mockServers(5);
+ AtomicInteger attempts = new AtomicInteger();
+ // Every REALTIME attempt throws; the method must not propagate it and
must still connect OFFLINE.
+ int connected = new ServerPreConnector(() -> targets(servers,
TableType.OFFLINE, TableType.REALTIME),
+ (server, tableType, timeoutMs) -> {
+ attempts.incrementAndGet();
+ if (tableType == TableType.REALTIME) {
+ throw new RuntimeException("connect blew up");
+ }
+ return true;
+ }).preConnect(farDeadline());
+
+ assertEquals(connected, 5); // only the 5 OFFLINE channels
+ assertEquals(attempts.get(), 10); // all 10 were still attempted
+ }
+
+ @Test
+ public void respectsBudgetAndDoesNotWaitForSlowConnects() {
+ List<ServerInstance> servers = mockServers(4);
+ // Every connect is far slower than the budget; preConnect must return
near the budget, not wait for
+ // the connects, and must not throw. Nothing completes, so the first poll
(which gets the whole budget)
+ // returns empty and releases.
+ long budgetMs = 400L;
+ long startMs = System.currentTimeMillis();
+ int connected = new ServerPreConnector(() -> targets(servers,
TableType.OFFLINE, TableType.REALTIME),
+ (server, tableType, timeoutMs) -> {
+ try {
+ Thread.sleep(5_000L);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ return true;
+ }).preConnect(System.currentTimeMillis() + budgetMs);
+ long elapsedMs = System.currentTimeMillis() - startMs;
+
+ assertEquals(connected, 0);
+ // Comfortably below the 5s connect: proves the budget bounded the wait
rather than blocking on
+ // the slow connects.
+ assertTrue(elapsedMs < 3_000L, "preConnect took " + elapsedMs + " ms,
expected it to honor the budget");
+ }
+
+ /// One unreachable server must not hold startup for the whole budget: once
the healthy channels are back
+ /// and nothing more arrives for the grace window, preConnect returns and
leaves the straggler to finish
+ /// (or time out) on its own daemon thread. Without the grace window the
final poll would block for the
+ /// rest of the budget.
+ @Test
+ public void oneStuckChannelDoesNotHoldStartupForTheWholeBudget() {
+ List<ServerInstance> servers = mockServers(4); // 4 x 2 table types = 8
channels
+ long budgetMs = 30_000L;
+ AtomicInteger n = new AtomicInteger();
+ long startMs = System.currentTimeMillis();
+
+ int connected = new ServerPreConnector(() -> targets(servers,
TableType.OFFLINE, TableType.REALTIME),
+ (server, tableType, timeoutMs) -> {
+ if (n.getAndIncrement() == 0) {
+ try {
+ Thread.sleep(timeoutMs);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ return false;
+ }
+ return true;
+ }).preConnect(startMs + budgetMs);
+ long elapsedMs = System.currentTimeMillis() - startMs;
+
+ assertEquals(connected, 7, "the seven healthy channels must still be
counted");
+ assertTrue(elapsedMs < 5 * ServerPreConnector.STRAGGLER_GRACE_MS,
+ "one stuck channel held startup for " + elapsedMs + " ms of a " +
budgetMs + " ms budget");
+ }
+
+ /// The whole budget is available until the first successful connect: a
cluster whose channels are all
+ /// slower than the grace window (but faster than the budget) must still
connect every one, not bail at
+ /// the grace window having connected nothing. If the exemption were
missing, the first poll would time
+ /// out at the grace window before any channel returned, and connected would
be 0.
+ @Test
+ public void slowFirstChannelIsStillCountedAndNotAbandonedByGraceWindow() {
+ List<ServerInstance> servers = mockServers(3); // 3 x 2 = 6 channels
+ long slowMs = ServerPreConnector.STRAGGLER_GRACE_MS + 500L; // slower
than grace, faster than budget
+ long budgetMs = 30_000L;
+ long startMs = System.currentTimeMillis();
+
+ int connected = new ServerPreConnector(() -> targets(servers,
TableType.OFFLINE, TableType.REALTIME),
+ (server, tableType, timeoutMs) -> {
+ try {
+ Thread.sleep(slowMs);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ return true;
+ }).preConnect(startMs + budgetMs);
+ long elapsedMs = System.currentTimeMillis() - startMs;
+
+ assertEquals(connected, 6, "every channel must be counted even though all
are slower than the grace window");
+ assertTrue(elapsedMs >= slowMs, "the first channel must be waited for past
the grace window, not abandoned");
+ assertTrue(elapsedMs < budgetMs, "must not wait the whole budget once the
channels are back");
+ }
+
+ /// A fast failure (an instantly refused connect) that completes before the
healthy channels must NOT
+ /// consume the whole-budget exemption and cause the grace window to abandon
the healthy-but-slower
+ /// channels. Since the exemption keys on the first *successful* connect,
the fast failure does not start
+ /// the grace clock, and the five healthy channels are all waited for and
counted. Regression test for a
+ /// grace-window bug where keying on the first *completion* undercounted to
0 here.
+ @Test
+ public void healthyButSlowChannelsNotAbandonedAfterFastFailure() {
+ List<ServerInstance> servers = mockServers(3); // 3 x 2 = 6 channels
+ long slowMs = ServerPreConnector.STRAGGLER_GRACE_MS + 500L; // slower
than grace, faster than budget
+ long budgetMs = 30_000L;
+ AtomicInteger n = new AtomicInteger();
+
+ int connected = new ServerPreConnector(() -> targets(servers,
TableType.OFFLINE, TableType.REALTIME),
+ (server, tableType, timeoutMs) -> {
+ if (n.getAndIncrement() == 0) {
+ return false; // one instant failure, completes first, must not
start the grace clock
+ }
+ try {
+ Thread.sleep(slowMs);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ return true;
+ }).preConnect(System.currentTimeMillis() + budgetMs);
+
+ assertEquals(connected, 5,
+ "the five healthy channels must be counted; a fast failure must not
trigger the grace window");
+ }
+
+ /// More channels than worker threads: the surplus queues behind the pool
and still all connect. Exercises
+ /// the `min(channelCount, MAX_CONNECT_THREADS)` pool sizing and the queue
draining that the completion
+ /// loop depends on.
+ @Test
+ public void moreTargetsThanThreadsAllConnect() {
+ int count = ServerPreConnector.MAX_CONNECT_THREADS * 3; // 48 servers ->
96 channels, pool caps at 16
+ List<ServerInstance> servers = mockServers(count);
+ AtomicInteger calls = new AtomicInteger();
+
+ int connected = new ServerPreConnector(() -> targets(servers,
TableType.OFFLINE, TableType.REALTIME),
+ (server, tableType, timeoutMs) -> {
+ calls.incrementAndGet();
+ return true;
+ }).preConnect(farDeadline());
+
+ assertEquals(connected, count * 2, "every queued channel must eventually
connect");
+ assertEquals(calls.get(), count * 2, "every channel must be attempted");
+ }
+
+ /// The thread pool is a throughput cap, not a safety bound, so each connect
has to carry its own
+ /// deadline-derived timeout. Without it a channel queued behind a stuck
worker could outlive the
+ /// budget entirely.
+ @Test
+ public void passesRemainingBudgetToEachConnect() {
+ List<ServerInstance> servers = mockServers(2);
+ long budgetMs = 5_000L;
+ AtomicLong maxTimeoutSeen = new AtomicLong(Long.MIN_VALUE);
+ AtomicLong minTimeoutSeen = new AtomicLong(Long.MAX_VALUE);
+
+ int connected = new ServerPreConnector(() -> targets(servers,
TableType.OFFLINE, TableType.REALTIME),
+ (server, tableType, timeoutMs) -> {
+ maxTimeoutSeen.accumulateAndGet(timeoutMs, Math::max);
+ minTimeoutSeen.accumulateAndGet(timeoutMs, Math::min);
+ return true;
+ }).preConnect(System.currentTimeMillis() + budgetMs);
+
+ assertEquals(connected, 4);
+ assertTrue(maxTimeoutSeen.get() <= budgetMs,
+ "connect timeout " + maxTimeoutSeen.get() + " ms must never exceed the
budget " + budgetMs + " ms");
+ assertTrue(minTimeoutSeen.get() >= 0, "connect timeout must never be
negative");
+ }
+
+ /// A connector may legitimately be handed a zero timeout when the budget
runs out mid-flight. It must
+ /// be treated as "no budget", never as "wait forever" -- Netty reads
+ /// `ChannelOption.CONNECT_TIMEOUT_MILLIS <= 0` as *no* connect timeout, so
a zero leaking through to
+ /// the bootstrap would park a worker indefinitely, which is the opposite of
what the bound is for.
+ @Test
+ public void connectTimeoutIsNeverNegative() {
+ List<ServerInstance> servers = mockServers(8);
+ AtomicLong minTimeoutSeen = new AtomicLong(Long.MAX_VALUE);
+
+ new ServerPreConnector(() -> targets(servers, TableType.OFFLINE,
TableType.REALTIME),
+ (server, tableType, timeoutMs) -> {
+ minTimeoutSeen.accumulateAndGet(timeoutMs, Math::min);
+ try {
+ Thread.sleep(20L);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ return true;
+ }).preConnect(System.currentTimeMillis() + 50L);
+
+ assertTrue(minTimeoutSeen.get() >= 0,
+ "connect timeout " + minTimeoutSeen.get() + " ms must never be
negative");
+ }
+
+ // ---- routing-derived target selection
(SingleConnectionBrokerRequestHandler#routableChannelTargets) ----
+
+ private static RoutingManager routing(Map<String, ServerInstance>
serverInstanceMap,
+ Map<String, Set<String>> tableToServingInstances) {
+ RoutingManager routingManager = mock(RoutingManager.class);
+
when(routingManager.getRoutableServerInstanceMap()).thenReturn(serverInstanceMap);
+
when(routingManager.getRoutableTables()).thenReturn(tableToServingInstances.keySet());
+ tableToServingInstances.forEach(
+ (table, servingInstances) ->
when(routingManager.getServingInstances(table)).thenReturn(servingInstances));
+ return routingManager;
+ }
+
+ private static Map<String, ServerInstance> serverInstances(String... ids) {
+ Map<String, ServerInstance> map = new HashMap<>();
+ for (String id : ids) {
+ map.put(id, mock(ServerInstance.class));
+ }
+ return map;
+ }
+
+ /// Only the (server, table type) pairs routing actually uses are targeted:
a server gets a channel for
+ /// each type that routes to it (s2 hybrid -> both), never the cross
product, and a server no table routes
+ /// to (s4) gets nothing -- even though it is in the cluster-wide
`getRoutableServerInstanceMap()`.
+ @Test
+ public void routableTargetsAreDerivedFromRoutingNotCrossProduct() {
+ Map<String, ServerInstance> serverMap = serverInstances("s1", "s2", "s3",
"s4");
+ RoutingManager routingManager = routing(serverMap, Map.of(
+ "a_OFFLINE", Set.of("s1", "s2"),
+ "b_REALTIME", Set.of("s2", "s3")));
+
+ Set<ChannelTarget> targets =
+ new
HashSet<>(SingleConnectionBrokerRequestHandler.routableChannelTargets(routingManager));
+
+ assertEquals(targets, Set.of(
+ new ChannelTarget(serverMap.get("s1"), TableType.OFFLINE),
+ new ChannelTarget(serverMap.get("s2"), TableType.OFFLINE),
+ new ChannelTarget(serverMap.get("s2"), TableType.REALTIME),
+ new ChannelTarget(serverMap.get("s3"), TableType.REALTIME)));
+ }
+
+ /// An offline-only cluster opens no REALTIME channels (the
wasted-duplicate-socket case).
+ @Test
+ public void offlineOnlyClusterTargetsNoRealtimeChannels() {
+ Map<String, ServerInstance> serverMap = serverInstances("s1", "s2");
+ RoutingManager routingManager = routing(serverMap, Map.of(
+ "a_OFFLINE", Set.of("s1", "s2"),
+ "b_OFFLINE", Set.of("s1")));
+
+ Set<ChannelTarget> targets =
+ new
HashSet<>(SingleConnectionBrokerRequestHandler.routableChannelTargets(routingManager));
+
+ // Two tables on s1 dedupe to a single (s1, OFFLINE); no REALTIME anywhere.
+ assertEquals(targets, Set.of(
+ new ChannelTarget(serverMap.get("s1"), TableType.OFFLINE),
+ new ChannelTarget(serverMap.get("s2"), TableType.OFFLINE)));
+ }
+
+ /// No routable tables -> no targets (a broker that converged with an empty
routing table, e.g. a tenant
+ /// with no tables yet). preConnect then connects nothing.
+ @Test
+ public void routableTargetsEmptyWhenNoRoutableTables() {
+ RoutingManager routingManager = routing(serverInstances("s1", "s2"),
Map.of());
+
assertTrue(SingleConnectionBrokerRequestHandler.routableChannelTargets(routingManager).isEmpty());
+ }
+
+ /// A serving instance not present in the routable-server map (e.g. just
disabled) is skipped rather than
+ /// producing a null-server target, and a table with no routing is ignored.
+ @Test
+ public void routableTargetsSkipUnknownServersAndUnroutedTables() {
+ Map<String, ServerInstance> serverMap = serverInstances("s1");
+ Map<String, Set<String>> tables = new HashMap<>();
+ tables.put("a_OFFLINE", Set.of("s1", "gone")); // "gone" is not in the
routable-server map
+ tables.put("b_OFFLINE", null); // no routing yet
+ RoutingManager routingManager = routing(serverMap, tables);
+
+ Set<ChannelTarget> targets =
+ new
HashSet<>(SingleConnectionBrokerRequestHandler.routableChannelTargets(routingManager));
+
+ assertEquals(targets, Set.of(new ChannelTarget(serverMap.get("s1"),
TableType.OFFLINE)));
+ }
+}
diff --git
a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/manager/MultiClusterRoutingManagerTest.java
b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/manager/MultiClusterRoutingManagerTest.java
index 0ea105a56e1..98131d62716 100644
---
a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/manager/MultiClusterRoutingManagerTest.java
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/manager/MultiClusterRoutingManagerTest.java
@@ -285,6 +285,20 @@ public class MultiClusterRoutingManagerTest {
assertTrue(result.containsKey("server2"));
}
+ /// Routable tables union across local and remote clusters. Without this
override the RoutingManager
+ /// default returns empty and startup pre-connect would derive no channels
on a multi-cluster broker.
+ @Test
+ public void testGetRoutableTablesCombinesAll() {
+
when(_localClusterRoutingManager.getRoutableTables()).thenReturn(Set.of("a_OFFLINE",
"shared_OFFLINE"));
+
when(_remoteClusterRoutingManager1.getRoutableTables()).thenReturn(Set.of("b_REALTIME",
"shared_OFFLINE"));
+
when(_remoteClusterRoutingManager2.getRoutableTables()).thenReturn(Set.of());
+
+ Set<String> result = _multiClusterRoutingManager.getRoutableTables();
+
+ // Union with the duplicate ("shared_OFFLINE") collapsed.
+ assertEquals(result, Set.of("a_OFFLINE", "b_REALTIME", "shared_OFFLINE"));
+ }
+
// Helper methods
private BrokerRequest createMockBrokerRequest(String tableName) {
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerTimer.java
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerTimer.java
index b7b8cc338c6..b049e51f66c 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerTimer.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerTimer.java
@@ -54,6 +54,15 @@ public enum BrokerTimer implements AbstractMetrics.Timer {
REALTIME_TOTAL_CPU_TIME_NS(false),
// How long it took the server to start.
STARTUP_SUCCESS_DURATION_MS(true),
+ // How long the startup server pre-connect (open broker->server channels,
including TLS handshake)
+ // took, from Helix convergence to the last channel connecting or the budget
expiring.
+ STARTUP_PRECONNECT_DURATION_MS(true),
+ // Time to establish a single broker->server Netty channel (TCP connect,
plus the TLS handshake on the
+ // startup pre-connect path). Complements the
NETTY_CONNECTION_CONNECT_TIME_MS gauge, which only keeps
+ // the last value, by keeping a burst of connections on a cold broker
analyzable across the whole
+ // distribution. NOTE: must not be named after that gauge -- both derive
their metric name from the
+ // enum constant, and a gauge and a timer sharing one name collide in the
metrics registry.
+ NETTY_CONNECTION_CONNECT_LATENCY_MS(true),
// GRPC query execution time
GRPC_QUERY_EXECUTION_MS(true),
// Audit logging timers
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/routing/RoutingManager.java
b/pinot-core/src/main/java/org/apache/pinot/core/routing/RoutingManager.java
index 53198ede031..fbd6fa27f37 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/routing/RoutingManager.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/routing/RoutingManager.java
@@ -127,6 +127,17 @@ public interface RoutingManager {
/// @return true if the route table exists.
boolean routingExists(String tableNameWithType);
+ /// Returns the tables this instance currently has routing for.
+ ///
+ /// Used by broker startup pre-connect to derive the (server, table type)
channels this broker actually
+ /// routes to, so it opens exactly those rather than the cross product of
every enabled server with every
+ /// table type. The default returns an empty set for implementations that do
not track routing per table.
+ ///
+ /// @return table names with type; empty when unknown.
+ default Set<String> getRoutableTables() {
+ return Set.of();
+ }
+
/// Acquire the time boundary info. Useful for hybrid logical table queries
that needs to split between
/// realtime and offline.
/// @param offlineTableName offline table name
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/transport/QueryRouter.java
b/pinot-core/src/main/java/org/apache/pinot/core/transport/QueryRouter.java
index e885a14788a..7d58a2a3413 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/transport/QueryRouter.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/transport/QueryRouter.java
@@ -157,6 +157,9 @@ public class QueryRouter {
}
/// Connects to the given server, returns `true` if the server is
successfully connected.
+ ///
+ /// Unchanged: this is the reachability probe the failure detector uses,
opening the OFFLINE channel
+ /// with a TCP connect only. Startup pre-connect uses [#preConnect] instead.
public boolean connect(ServerInstance serverInstance) {
try {
if (_serverChannelsTls != null) {
@@ -173,6 +176,29 @@ public class QueryRouter {
}
}
+ /// Opens the channel used for the given table type ahead of query traffic,
including the TLS
+ /// handshake, bounded by `timeoutMs`. Returns `true` if it is connected.
+ ///
+ /// [ServerRoutingInstance] includes the table type in its
`equals`/`hashCode`, so OFFLINE and REALTIME
+ /// map to **separate** channels for the same physical server; the caller
decides which of them this
+ /// broker actually routes to. Whatever is left out is still established
lazily by the first query that
+ /// needs it.
+ public boolean preConnect(ServerInstance serverInstance, TableType
tableType, long timeoutMs) {
+ try {
+ if (_serverChannelsTls != null) {
+ _serverChannelsTls.preConnect(
+ serverInstance.toServerRoutingInstance(tableType,
ServerInstance.RoutingType.NETTY_TLS), timeoutMs);
+ } else {
+ _serverChannels.preConnect(
+ serverInstance.toServerRoutingInstance(tableType,
ServerInstance.RoutingType.NETTY), timeoutMs);
+ }
+ return true;
+ } catch (Exception e) {
+ LOGGER.debug("Failed to pre-connect to server: {} for table type: {}",
serverInstance, tableType, e);
+ return false;
+ }
+ }
+
public void shutDown() {
_serverChannels.shutDown();
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java
b/pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java
index 2bf34540cf3..d711c120337 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java
@@ -36,6 +36,8 @@ import io.netty.channel.kqueue.KQueueSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.Future;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -141,12 +143,25 @@ public class ServerChannels {
ServerRoutingInstance serverRoutingInstance, InstanceRequest
instanceRequest, long timeoutMs)
throws Exception {
byte[] requestBytes =
THREAD_LOCAL_T_SERIALIZER.get().serialize(instanceRequest);
- _serverToChannelMap.computeIfAbsent(serverRoutingInstance,
ServerChannel::new)
- .sendRequest(rawTableName, asyncQueryResponse, serverRoutingInstance,
requestBytes, timeoutMs);
+ ServerChannel serverChannel =
_serverToChannelMap.computeIfAbsent(serverRoutingInstance, ServerChannel::new);
+ // This server is now query-carrying, whoever opened the channel. See
[#hasChannel].
+ serverChannel._openedByQuery = true;
+ serverChannel.sendRequest(rawTableName, asyncQueryResponse,
serverRoutingInstance, requestBytes, timeoutMs);
}
+ /// Whether this broker has sent, or tried to send, a single-stage query to
the given server.
+ ///
+ /// Deliberately **not** "is there an entry in the channel map". Startup
pre-connect opens channels
+ /// before any query, so a plain `containsKey` would flip true for every
reachable server merely because
+ /// pre-connect ran.
`SingleConnectionBrokerRequestHandler#retryUnhealthyServer` uses this to decide
+ /// whether the single-stage transport has any opinion on a server's health
at all -- it reports
+ /// `UNKNOWN` when it does not -- so on a cluster serving only multi-stage
queries the answer has to stay
+ /// `false`, exactly as it was before pre-connect existed. Otherwise a
server that multi-stage can reach
+ /// over gRPC but single-stage cannot reach over Netty would be voted
`UNHEALTHY` and dropped from
+ /// routing by a transport that never carries its queries.
public boolean hasChannel(ServerRoutingInstance serverRoutingInstance) {
- return _serverToChannelMap.containsKey(serverRoutingInstance);
+ ServerChannel serverChannel =
_serverToChannelMap.get(serverRoutingInstance);
+ return serverChannel != null && serverChannel._openedByQuery;
}
public void connect(ServerRoutingInstance serverRoutingInstance)
@@ -154,6 +169,16 @@ public class ServerChannels {
_serverToChannelMap.computeIfAbsent(serverRoutingInstance,
ServerChannel::new).connect();
}
+ /// Opens a channel ahead of query traffic, awaiting the TLS handshake so
that neither the connect nor
+ /// the handshake lands on the first query's critical path. Both waits are
bounded by `timeoutMs`.
+ ///
+ /// The channel is entered into the same map the query path uses, so the
first query reuses it rather
+ /// than connecting again -- but it is not marked query-carrying, so
[#hasChannel] is unaffected.
+ public void preConnect(ServerRoutingInstance serverRoutingInstance, long
timeoutMs)
+ throws InterruptedException, TimeoutException {
+ _serverToChannelMap.computeIfAbsent(serverRoutingInstance,
ServerChannel::new).preConnect(timeoutMs);
+ }
+
public void shutDown() {
// Shut down immediately
_eventLoopGroup.shutdownGracefully(0, 0, TimeUnit.SECONDS);
@@ -171,6 +196,11 @@ public class ServerChannels {
// lock to protect channel as requests must be written into channel
sequentially
final ReentrantLock _channelLock = new ReentrantLock();
Channel _channel;
+ // Set once a query has been sent, or attempted, through this channel;
startup pre-connect leaves it
+ // false. Read by hasChannel(), which the failure detector uses to decide
whether the single-stage
+ // transport has any opinion on this server's health. Volatile: written on
a query thread, read on the
+ // failure-detector retry thread.
+ volatile boolean _openedByQuery;
ServerChannel(ServerRoutingInstance serverRoutingInstance) {
_serverRoutingInstance = serverRoutingInstance;
@@ -234,13 +264,90 @@ public class ServerChannels {
}
}
+ /// Lazy query path: opens the TCP connection only. Any TLS handshake is
left to proceed
+ /// asynchronously so the channel lock is released as soon as the socket
is up, keeping the first
+ /// query's critical section short. Startup pre-connect uses
[#preConnectWithoutLocking(long)]
+ /// instead, which additionally pays the handshake.
void connectWithoutLocking()
throws InterruptedException {
if (_channel == null || !_channel.isActive()) {
long startTime = System.currentTimeMillis();
_channel = _bootstrap.connect().sync().channel();
-
_brokerMetrics.setValueOfGlobalGauge(BrokerGauge.NETTY_CONNECTION_CONNECT_TIME_MS,
- System.currentTimeMillis() - startTime);
+ recordConnectTime(System.currentTimeMillis() - startTime);
+ }
+ }
+
+ /// Like [#connectWithoutLocking()] but additionally waits out the TLS
handshake, and bounds both
+ /// waits by `timeoutMs`.
+ ///
+ /// Used only by startup pre-connect ([#preConnect(long)]), never by the
lazy query path or by the
+ /// failure detector's reconnect probe ([#connect()]) -- both of those
keep the shorter critical
+ /// section they had before this feature existed.
+ ///
+ /// `_channel` is assigned only after the handshake succeeds. Netty fails
the handshake promise
+ /// *before* it closes the channel (`SslHandler#setHandshakeFailure` calls
`Promise#tryFailure` and
+ /// only then `SslUtils#handleHandshakeFailure` -> `ctx.close()`, which is
itself asynchronous), so a
+ /// channel assigned up front would briefly still report `isActive()`, and
a query written into it
+ /// would fail.
+ void preConnectWithoutLocking(long timeoutMs)
+ throws InterruptedException, TimeoutException {
+ if (_channel != null && _channel.isActive()) {
+ return;
+ }
+ if (timeoutMs <= 0) {
+ // No budget left. Must return before touching CONNECT_TIMEOUT_MILLIS:
Netty only schedules its
+ // connect-timeout task when the value is > 0, so passing 0 would mean
"wait forever".
+ throw new TimeoutException("No pre-connect budget left for server: " +
_serverRoutingInstance);
+ }
+ long startTime = System.currentTimeMillis();
+ // The shared bootstrap leaves ChannelOption.CONNECT_TIMEOUT_MILLIS at
Netty's 30s default, which is
+ // the whole pre-connect budget -- one server whose SYN is dropped
rather than refused would consume
+ // it alone and occupy a worker for the entire startup. Clone the
bootstrap so the tighter bound
+ // applies to pre-connect only and the query path is untouched.
+ Channel channel = _bootstrap.clone()
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, (int)
Math.min(timeoutMs, Integer.MAX_VALUE))
+ .connect().sync().channel();
+ try {
+ awaitTlsHandshake(channel, Math.max(0L, timeoutMs -
(System.currentTimeMillis() - startTime)));
+ } catch (Throwable t) {
+ channel.close();
+ throw t;
+ }
+ _channel = channel;
+ recordConnectTime(System.currentTimeMillis() - startTime);
+ }
+
+ private void recordConnectTime(long connectTimeMs) {
+
_brokerMetrics.setValueOfGlobalGauge(BrokerGauge.NETTY_CONNECTION_CONNECT_TIME_MS,
connectTimeMs);
+
_brokerMetrics.addTimedValue(BrokerTimer.NETTY_CONNECTION_CONNECT_LATENCY_MS,
connectTimeMs,
+ TimeUnit.MILLISECONDS);
+ }
+
+ /// Blocks until the TLS handshake on a freshly-connected channel
completes, for at most `timeoutMs`.
+ ///
+ /// `bootstrap.connect().sync()` returns once the TCP connection is up;
the client-mode [SslHandler]
+ /// then drives the handshake asynchronously on the event loop. Awaiting
its future here pays the
+ /// handshake -- two round trips plus certificate validation -- on the
connecting thread rather than
+ /// on the first query that writes to the channel. On a plaintext channel
there is no [SslHandler] in
+ /// the pipeline and this is a no-op. The calling thread is never an
event-loop thread, so this cannot
+ /// deadlock.
+ ///
+ /// The wait is bounded by the caller's remaining budget rather than by
[SslHandler]'s own
+ /// `handshakeTimeoutMillis`, which defaults to 10s and is not set on this
pipeline -- otherwise a
+ /// single hung TLS peer could outlive the pre-connect budget.
+ private void awaitTlsHandshake(Channel channel, long timeoutMs)
+ throws InterruptedException, TimeoutException {
+ SslHandler sslHandler = channel.pipeline().get(SslHandler.class);
+ if (sslHandler == null) {
+ return;
+ }
+ Future<Channel> handshakeFuture = sslHandler.handshakeFuture();
+ if (!handshakeFuture.await(timeoutMs, TimeUnit.MILLISECONDS)) {
+ throw new TimeoutException("Timed out waiting for the TLS handshake to
server: " + _serverRoutingInstance);
+ }
+ if (!handshakeFuture.isSuccess()) {
+ throw new RuntimeException("Failed the TLS handshake to server: " +
_serverRoutingInstance,
+ handshakeFuture.cause());
}
}
@@ -267,6 +374,11 @@ public class ServerChannels {
_brokerMetrics.addMeteredGlobalValue(BrokerMeter.NETTY_CONNECTION_BYTES_SENT,
requestBytes.length);
}
+ /// Opens the TCP connection, as the failure detector's reconnect probe
+ /// (`SingleConnectionBrokerRequestHandler#retryUnhealthyServer`) has
always done. Deliberately does
+ /// **not** await the TLS handshake: this runs at steady state under live
traffic, and holding
+ /// `_channelLock` across a handshake would make concurrent queries to a
recovering server queue
+ /// behind it -- the very serialization startup pre-connect exists to
remove.
void connect()
throws InterruptedException, TimeoutException {
if (_channelLock.tryLock(TRY_CONNECT_CHANNEL_LOCK_TIMEOUT_MS,
TimeUnit.MILLISECONDS)) {
@@ -279,5 +391,19 @@ public class ServerChannels {
throw new TimeoutException(CHANNEL_LOCK_TIMEOUT_MSG);
}
}
+
+ /// Startup pre-connect: opens the connection and pays the TLS handshake,
bounded by `timeoutMs`.
+ void preConnect(long timeoutMs)
+ throws InterruptedException, TimeoutException {
+ if (_channelLock.tryLock(TRY_CONNECT_CHANNEL_LOCK_TIMEOUT_MS,
TimeUnit.MILLISECONDS)) {
+ try {
+ preConnectWithoutLocking(timeoutMs);
+ } finally {
+ _channelLock.unlock();
+ }
+ } else {
+ throw new TimeoutException(CHANNEL_LOCK_TIMEOUT_MSG);
+ }
+ }
}
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/transport/ServerChannelsTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/transport/ServerChannelsTest.java
index e2b7c89db5f..7fec4b33fed 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/transport/ServerChannelsTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/transport/ServerChannelsTest.java
@@ -44,8 +44,11 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
public class ServerChannelsTest {
@@ -197,4 +200,96 @@ public class ServerChannelsTest {
serverChannels.shutDown();
}
+
+ /// The load-bearing guarantee for clusters that do not use the single-stage
engine: pre-connect must
+ /// not make `hasChannel()` true. `retryUnhealthyServer` reads it to decide
whether the single-stage
+ /// transport has any opinion on a server's health, and answers `UNKNOWN`
when it does not. If opening a
+ /// channel ahead of traffic flipped it, an MSE-only cluster would start
voting servers `UNHEALTHY` on
+ /// Netty reachability -- and that vote short-circuits the retrier loop
before gRPC is ever consulted.
+ @Test
+ public void testSuccessfulPreConnectLeavesHasChannelFalse()
+ throws Exception {
+ HttpServer dummyServer = HttpServer.create();
+ dummyServer.bind(new InetSocketAddress("localhost", 0), 0);
+ dummyServer.start();
+ ServerChannels serverChannels =
+ new ServerChannels(mock(QueryRouter.class), null, null,
ThreadAccountantUtils.getNoOpAccountant());
+ try {
+ ServerRoutingInstance instance =
+ new ServerRoutingInstance("localhost",
dummyServer.getAddress().getPort(), TableType.OFFLINE);
+ serverChannels.preConnect(instance, 5_000L);
+ assertFalse(serverChannels.hasChannel(instance),
+ "Pre-connect must not make the server look query-carrying to the
failure detector");
+ } finally {
+ serverChannels.shutDown();
+ dummyServer.stop(0);
+ }
+ }
+
+ /// ... and a failed pre-connect must not either, so an unreachable server
is not voted on.
+ @Test
+ public void testFailedPreConnectLeavesHasChannelFalse() {
+ ServerChannels serverChannels =
+ new ServerChannels(mock(QueryRouter.class), null, null,
ThreadAccountantUtils.getNoOpAccountant());
+ // Port 1 on localhost: nothing listens, so the connect is refused rather
than timing out.
+ ServerRoutingInstance unreachable = new ServerRoutingInstance("localhost",
1, TableType.OFFLINE);
+ try {
+ assertThrows(Exception.class, () ->
serverChannels.preConnect(unreachable, 5_000L));
+ assertFalse(serverChannels.hasChannel(unreachable));
+ } finally {
+ serverChannels.shutDown();
+ }
+ }
+
+ /// Sending a query is what makes a server query-carrying -- unchanged from
before pre-connect existed,
+ /// including when the connect itself fails.
+ @Test
+ public void testSendRequestMakesHasChannelTrue() {
+ ServerChannels serverChannels =
+ new ServerChannels(mock(QueryRouter.class), null, null,
ThreadAccountantUtils.getNoOpAccountant());
+ ServerRoutingInstance unreachable = new ServerRoutingInstance("localhost",
1, TableType.OFFLINE);
+ try {
+ InstanceRequest instanceRequest = new InstanceRequest();
+ instanceRequest.setRequestId(1L);
+ instanceRequest.setQuery(new BrokerRequest());
+ assertFalse(serverChannels.hasChannel(unreachable));
+ assertThrows(Exception.class, () -> serverChannels.sendRequest("t",
mock(AsyncQueryResponse.class), unreachable,
+ instanceRequest, 5_000L));
+ assertTrue(serverChannels.hasChannel(unreachable));
+ } finally {
+ serverChannels.shutDown();
+ }
+ }
+
+ /// The point of pre-connect: the first query reuses the warm channel rather
than opening a second one,
+ /// and only then does the server become query-carrying.
+ @Test
+ public void testFirstQueryReusesThePreConnectedChannel()
+ throws Exception {
+ HttpServer dummyServer = HttpServer.create();
+ dummyServer.bind(new InetSocketAddress("localhost", 0), 0);
+ dummyServer.start();
+ ServerChannels serverChannels =
+ new ServerChannels(mock(QueryRouter.class), null, null,
ThreadAccountantUtils.getNoOpAccountant());
+ try {
+ ServerRoutingInstance instance =
+ new ServerRoutingInstance("localhost",
dummyServer.getAddress().getPort(), TableType.OFFLINE);
+ serverChannels.preConnect(instance, 5_000L);
+ ServerChannels.ServerChannel preConnected =
serverChannels.getOrCreateServerChannel(instance);
+ Channel warmChannel = preConnected._channel;
+ assertNotNull(warmChannel);
+
+ InstanceRequest instanceRequest = new InstanceRequest();
+ instanceRequest.setRequestId(1L);
+ instanceRequest.setQuery(new BrokerRequest());
+ serverChannels.sendRequest("t", mock(AsyncQueryResponse.class),
instance, instanceRequest, 5_000L);
+
+ assertSame(serverChannels.getOrCreateServerChannel(instance),
preConnected);
+ assertSame(preConnected._channel, warmChannel, "The query must reuse the
pre-connected channel");
+ assertTrue(serverChannels.hasChannel(instance));
+ } finally {
+ serverChannels.shutDown();
+ dummyServer.stop(0);
+ }
+ }
}
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BrokerServerPreConnectIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BrokerServerPreConnectIntegrationTest.java
new file mode 100644
index 00000000000..54ac65a0147
--- /dev/null
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BrokerServerPreConnectIntegrationTest.java
@@ -0,0 +1,142 @@
+/**
+ * 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.integration.tests;
+
+import java.io.File;
+import java.util.List;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.broker.requesthandler.ServerPreConnector;
+import org.apache.pinot.common.utils.ServiceStatus;
+import org.apache.pinot.core.transport.ServerInstance;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.util.TestUtils;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.mock;
+
+
+/// Integration test for the broker startup server pre-connect feature
+/// (`pinot.broker.startup.preconnect.enabled`). Brings up a real ZK +
controller + server + broker with
+/// an offline table, then verifies end-to-end that:
+///
+/// 1. The production pre-connect path (`RoutingManager` -> `QueryRouter` ->
`ServerChannels` -> a live
+/// server) opens exactly the channels routing derives -- for this
offline-only table, one OFFLINE
+/// channel per serving server and no REALTIME channels.
+/// 2. The readiness gate genuinely holds and then releases -- exercised
directly against
+/// [ServerPreConnector] with a slow connector, because the broker's own
gate is fast enough that
+/// asserting on its terminal status could not distinguish a working gate
from one stuck open.
+///
+/// The server is started **before** the broker so that routing, and therefore
the set of channels to
+/// pre-connect, is non-empty by the time the broker's pre-connect thread
runs. With the broker first
+/// there is nothing to connect and the feature would appear to pass while
doing nothing.
+public class BrokerServerPreConnectIntegrationTest extends
BaseClusterIntegrationTest {
+ private static final long PRECONNECT_TIMEOUT_MS = 30_000L;
+
+ @Override
+ protected void overrideBrokerConf(PinotConfiguration brokerConf) {
+
brokerConf.setProperty(CommonConstants.Broker.CONFIG_OF_BROKER_STARTUP_PRECONNECT_ENABLED,
true);
+
brokerConf.setProperty(CommonConstants.Broker.CONFIG_OF_BROKER_STARTUP_PRECONNECT_TIMEOUT_MS,
+ PRECONNECT_TIMEOUT_MS);
+ }
+
+ @BeforeClass
+ public void setUp()
+ throws Exception {
+ TestUtils.ensureDirectoriesExistAndEmpty(_tempDir, _segmentDir, _tarDir);
+ startZk();
+ startController();
+ startServer();
+ startBroker();
+
+ Schema schema = createSchema();
+ addSchema(schema);
+ TableConfig tableConfig = createOfflineTableConfig();
+ addTableConfig(tableConfig);
+
+ // Build and upload segments so the broker has a live server to route to
-- and therefore a real
+ // channel to pre-connect against.
+ List<File> avroFiles = unpackAvroData(_tempDir);
+ ClusterIntegrationTestUtils.buildSegmentsFromAvro(avroFiles, tableConfig,
schema, 0, _segmentDir, _tarDir);
+ uploadSegments(getTableName(), _tarDir);
+
+ waitForAllDocsLoaded(600_000L);
+ }
+
+ @AfterClass
+ public void tearDown()
+ throws Exception {
+ dropOfflineTable(getTableName());
+ stopBroker();
+ stopServer();
+ stopController();
+ stopZk();
+ FileUtils.deleteDirectory(_tempDir);
+ }
+
+ @Test
+ public void preConnectEnabledBrokerReachesGoodServiceStatus() {
+ String instanceId = _brokerStarters.get(0).getInstanceId();
+ TestUtils.waitForCondition(aVoid ->
ServiceStatus.getServiceStatus(instanceId) == ServiceStatus.Status.GOOD,
+ PRECONNECT_TIMEOUT_MS, "Broker with pre-connect enabled never reported
GOOD service status");
+ }
+
+ @Test
+ public void preConnectOpensChannelsToEveryLiveServer() {
+ // Routing-derived targets: an offline-only table yields one OFFLINE
channel per serving server and no
+ // REALTIME channels. Connecting an already-open channel is a no-op that
still counts as connected, so
+ // the expected count holds regardless of any earlier lazy connects from
setUp's queries.
+ int expectedChannels = _serverStarters.size();
+ int connected = _brokerStarters.get(0).getBrokerRequestHandler()
+ .preConnectServers(System.currentTimeMillis() + PRECONNECT_TIMEOUT_MS);
+ Assert.assertEquals(connected, expectedChannels,
+ "Pre-connect should open one OFFLINE channel per serving server and no
REALTIME channels");
+ }
+
+ /// The readiness gate is the highest-risk part of the feature, and the
broker's own pre-connect
+ /// finishes in milliseconds here, so a terminal-status assertion cannot
tell a working gate from one
+ /// that never engaged. Drive [ServerPreConnector] directly with a connector
slower than the budget and
+ /// assert both halves of the contract: the wait is bounded, and it ends.
+ @Test
+ public void preConnectBoundsTheGateWhenServersAreSlow() {
+ List<ServerPreConnector.ChannelTarget> targets =
+ List.of(new
ServerPreConnector.ChannelTarget(mock(ServerInstance.class),
TableType.OFFLINE));
+ long budgetMs = 500L;
+ long startMs = System.currentTimeMillis();
+ int connected = new ServerPreConnector(() -> targets, (server, tableType,
timeoutMs) -> {
+ try {
+ Thread.sleep(30_000L);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ return false;
+ }).preConnect(startMs + budgetMs);
+ long elapsedMs = System.currentTimeMillis() - startMs;
+
+ Assert.assertEquals(connected, 0);
+ Assert.assertTrue(elapsedMs < 10_000L,
+ "Pre-connect held the readiness gate for " + elapsedMs + " ms, well
past its " + budgetMs + " ms budget");
+ }
+}
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TlsIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TlsIntegrationTest.java
index 7da5dce82e2..9c8db73105e 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TlsIntegrationTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TlsIntegrationTest.java
@@ -589,6 +589,35 @@ public class TlsIntegrationTest extends
BaseClusterIntegrationTest {
}
}
+ /// Startup pre-connect over a **real** broker-to-server TLS channel. This
cluster runs the server with
+ /// `netty.enabled=false` and `nettytls.enabled=true`, so every single-stage
channel the broker opens
+ /// carries an `SslHandler` -- which is the case pre-connect exists for, and
the one no plaintext test
+ /// can reach.
+ ///
+ /// The assertion is the channel count rather than a log line because
`preConnectServers` awaits the
+ /// handshake and reports a channel as connected only once it has completed:
a handshake that failed,
+ /// timed out, or was never awaited would show up here as a short count.
Pre-connect swallows its own
+ /// failures by design, so without this the TLS path could break silently
and every other assertion in
+ /// this class would still pass.
+ ///
+ /// Connecting an already-open channel is a no-op that still counts, so the
expected count holds
+ /// regardless of channels the preceding tests' queries already opened
lazily.
+ ///
+ /// Pre-connect opens only the (server, table type) pairs routing derives.
This cluster's offline table
+ /// has no segments uploaded -- only the realtime table is fed, via Kafka --
so only the REALTIME channel
+ /// is routed: one per serving server, and no OFFLINE channel. The old cross
product would have opened an
+ /// OFFLINE channel here too, to a server holding no offline segment:
exactly the wasted TLS handshake and
+ /// idle socket this change removes.
+ @Test
+ public void testPreConnectOpensTlsChannelsToEveryServer() {
+ int expectedChannels = _serverStarters.size();
+ int connected = _brokerStarters.get(0).getBrokerRequestHandler()
+ .preConnectServers(System.currentTimeMillis() + 30_000L);
+ Assert.assertEquals(connected, expectedChannels,
+ "Pre-connect should complete the TLS handshake for the realtime
channel each server serves, and open "
+ + "no offline channel");
+ }
+
@Test
public void testLogicalTableTlsRouting()
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 50cb94b4a15..02c91c1afe1 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
@@ -434,6 +434,21 @@ public class CommonConstants {
public static final String CONFIG_OF_BROKER_MIN_RESOURCE_PERCENT_FOR_START
=
"pinot.broker.startup.minResourcePercent";
public static final double DEFAULT_BROKER_MIN_RESOURCE_PERCENT_FOR_START =
100.0;
+
+ // When enabled, once Helix converges at startup the broker opens a Netty
channel to every (server,
+ // table type) it routes to -- including the TLS handshake when
broker->server TLS is on -- so the
+ // first real query does not pay the blocking connect on its critical
path. Runs on a background
+ // thread and is bounded by
CONFIG_OF_BROKER_STARTUP_PRECONNECT_TIMEOUT_MS; channels that do not make
+ // it fall back to the lazy path. On by default; set to false to restore
the pure lazy-connect path,
+ // whose behaviour is then unchanged.
+ public static final String CONFIG_OF_BROKER_STARTUP_PRECONNECT_ENABLED =
+ "pinot.broker.startup.preconnect.enabled";
+ public static final boolean DEFAULT_BROKER_STARTUP_PRECONNECT_ENABLED =
true;
+ // Upper bound on the whole pre-connect step so a slow or unreachable
server cannot delay it
+ // indefinitely; channels not connected within the budget fall back to the
lazy path.
+ public static final String CONFIG_OF_BROKER_STARTUP_PRECONNECT_TIMEOUT_MS =
+ "pinot.broker.startup.preconnect.timeoutMs";
+ public static final long DEFAULT_BROKER_STARTUP_PRECONNECT_TIMEOUT_MS =
30_000L;
public static final String CONFIG_OF_ENABLE_QUERY_LIMIT_OVERRIDE =
"pinot.broker.enable.query.limit.override";
// Config for number of threads to use for Broker reduce-phase.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
