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 568df83b4f4 Add broker startup warmup for the query serve path (SSE)
(#19500)
568df83b4f4 is described below
commit 568df83b4f49c95328a57e378718993bfb166dc5
Author: Jinesh Parakh <[email protected]>
AuthorDate: Wed Sep 9 22:52:09 2026 +0530
Add broker startup warmup for the query serve path (SSE) (#19500)
---
.../broker/broker/helix/BaseBrokerStarter.java | 148 +++++-
.../requesthandler/BrokerRequestHandler.java | 27 ++
.../BrokerRequestHandlerDelegate.java | 8 +
.../broker/requesthandler/BrokerWarmupConfig.java | 60 +++
.../SingleConnectionBrokerRequestHandler.java | 536 +++++++++++++++++++++
.../broker/broker/helix/BrokerWarmupGateTest.java | 50 ++
.../broker/requesthandler/BrokerWarmupTest.java | 454 +++++++++++++++++
.../apache/pinot/common/metrics/BrokerGauge.java | 11 +
.../apache/pinot/common/metrics/BrokerMetrics.java | 7 +
.../apache/pinot/common/metrics/BrokerTimer.java | 3 +
.../tests/BrokerStartupWarmupIntegrationTest.java | 142 ++++++
.../apache/pinot/spi/utils/CommonConstants.java | 26 +
12 files changed, 1467 insertions(+), 5 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 e02549da2e9..826c732d4d5 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
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.broker.broker.helix;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import java.io.IOException;
import java.net.InetAddress;
@@ -29,6 +30,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
import javax.annotation.Nullable;
import javax.net.ssl.SSLContext;
import nl.altindag.ssl.SSLFactory;
@@ -55,6 +57,7 @@ import
org.apache.pinot.broker.requesthandler.BaseSingleStageBrokerRequestHandle
import org.apache.pinot.broker.requesthandler.BrokerRequestHandler;
import org.apache.pinot.broker.requesthandler.BrokerRequestHandlerDelegate;
import org.apache.pinot.broker.requesthandler.BrokerRequestIdGenerator;
+import org.apache.pinot.broker.requesthandler.BrokerWarmupConfig;
import org.apache.pinot.broker.requesthandler.GrpcBrokerRequestHandler;
import org.apache.pinot.broker.requesthandler.MultiStageBrokerRequestHandler;
import org.apache.pinot.broker.requesthandler.MultiStageQueryThrottler;
@@ -145,6 +148,13 @@ public abstract class BaseBrokerStarter implements
ServiceStartable {
/// 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;
+ /// How long shutdown waits for the interrupted warmup thread to unwind
before
+ /// giving up on it. Bounds the teardown race without letting a stuck thread
hold up shutdown.
+ private static final long STARTUP_THREAD_JOIN_TIMEOUT_MS = 1_000L;
+ /// Readiness description reported while the warmup gate holds the broker at
STARTING. Shared with tests so
+ /// the exact gate transition can be asserted.
+ @VisibleForTesting
+ static final String WARMUP_GATE_STARTING_DESCRIPTION = "Warming up broker
data plane";
protected PinotConfiguration _brokerConf;
protected List<ListenerConfig> _listenerConfigs;
@@ -193,6 +203,14 @@ public abstract class BaseBrokerStarter implements
ServiceStartable {
protected BrokerGrpcServer _brokerGrpcServer;
protected FailureDetector _failureDetector;
protected ThreadAccountant _threadAccountant;
+ /// Startup data-plane warmup config. Read once in `start()`.
+ protected BrokerWarmupConfig _warmupConfig =
+ new BrokerWarmupConfig(false, 0L, 0, 1);
+ /// Whether the data plane has been warmed. Gates readiness when warmup is
enabled; always true
+ /// otherwise, so the status callback behaves exactly as before for existing
deployments.
+ private volatile boolean _isWarm;
+ @Nullable
+ private volatile Thread _warmupThread;
/// 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.
@@ -691,6 +709,11 @@ public abstract class BaseBrokerStarter implements
ServiceStartable {
Broker.DEFAULT_BROKER_STARTUP_PRECONNECT_ENABLED);
_preConnectTimeoutMs =
_brokerConf.getProperty(Broker.CONFIG_OF_BROKER_STARTUP_PRECONNECT_TIMEOUT_MS,
Broker.DEFAULT_BROKER_STARTUP_PRECONNECT_TIMEOUT_MS);
+ // Read the warmup config before registering the status handler: the
handler adds the warmup readiness
+ // gate only when warmup is enabled, and the gate must be in place before
the handler is registered so
+ // there is no window where readiness is granted un-gated.
+ _warmupConfig = BrokerWarmupConfig.from(_brokerConf);
+ // Register the service status handler
registerServiceStatusHandler();
if (_preConnectEnabled) {
// Startup is not finished until the broker-to-server channels are open,
so `_isStarting` stays set
@@ -703,6 +726,9 @@ public abstract class BaseBrokerStarter implements
ServiceStartable {
} else {
_isStarting = false;
}
+ // Warmup gates readiness on its own _isWarm flag (added to the status
handler above when enabled), so
+ // it composes with pre-connect's _isStarting gate: readiness is granted
only once both are satisfied.
+ startWarmup();
_brokerMetrics.addTimedValue(BrokerTimer.STARTUP_SUCCESS_DURATION_MS,
System.currentTimeMillis() - startTimeMs, TimeUnit.MILLISECONDS);
@@ -894,10 +920,96 @@ public abstract class BaseBrokerStarter implements
ServiceStartable {
new
ServiceStatus.IdealStateAndExternalViewMatchServiceStatusCallback(_participantHelixManager,
_clusterName, _instanceId, resourcesToMonitor,
minResourcePercentForStartup)));
+ List<ServiceStatus.ServiceStatusCallback> callbacks = new ArrayList<>(3);
+ callbacks.add(new
ServiceStatus.LifecycleServiceStatusCallback(this::isStarting,
this::isShuttingDown));
+ callbacks.add(_helixConvergenceCallback);
+ if (_warmupConfig.enabled()) {
+ // The warmup readiness gate. Reports STARTING (not a new status value)
until warmup completes:
+ // callers throughout the codebase test for GOOD, and a new enum
constant would be visible to older
+ // mixed-version peers. MultipleCallbackServiceStatusCallback surfaces
the first non-GOOD callback, so
+ // this composes without touching the Helix or lifecycle callbacks.
+ //
+ // Scope: this gates HTTP readiness only -- the /health endpoint
(getBrokerHealth -> ServiceStatus)
+ // that a load balancer or Kubernetes readiness probe polls, so a
warming broker is not put into
+ // rotation there. It does NOT change Helix discovery: the broker is
already ONLINE in the broker
+ // resource's external view by this point, so a client that resolves
brokers straight from Helix could
+ // still route to it while it warms. That is acceptable -- warmup only
makes the first queries faster,
+ // never wrong -- and keeping the broker in Helix is deliberate, since
removing it would be a
+ // routing/discovery change far beyond a startup latency optimization.
+ callbacks.add(warmupGateCallback(() -> _isWarm));
+ }
ServiceStatus.setServiceStatusCallback(_instanceId,
- new ServiceStatus.MultipleCallbackServiceStatusCallback(List.of(
- new ServiceStatus.LifecycleServiceStatusCallback(this::isStarting,
this::isShuttingDown),
- _helixConvergenceCallback)));
+ new ServiceStatus.MultipleCallbackServiceStatusCallback(callbacks));
+ }
+
+ /// The startup-warmup readiness gate as a standalone callback: reports
STARTING with the "warming up"
+ /// description until `isWarm` turns true, then GOOD with no description.
Extracted and package-private so
+ /// the STARTING -> GOOD transition can be unit-tested deterministically,
without standing up a broker or
+ /// racing an actual warmup to observe it mid-flight.
+ @VisibleForTesting
+ static ServiceStatus.ServiceStatusCallback
warmupGateCallback(BooleanSupplier isWarm) {
+ return new ServiceStatus.ServiceStatusCallback() {
+ @Override
+ public ServiceStatus.Status getServiceStatus() {
+ return isWarm.getAsBoolean() ? ServiceStatus.Status.GOOD :
ServiceStatus.Status.STARTING;
+ }
+
+ @Override
+ public String getStatusDescription() {
+ return isWarm.getAsBoolean() ? ServiceStatus.STATUS_DESCRIPTION_NONE :
WARMUP_GATE_STARTING_DESCRIPTION;
+ }
+ };
+ }
+
+ /// Runs the data-plane warmup on a background thread and flips [#_isWarm]
when it finishes.
+ ///
+ /// Asynchronous so `start()` still returns promptly -- the gate is enforced
through `ServiceStatus`, not
+ /// by blocking startup. The flag is set in a `finally` so the gate opens
even if warmup throws:
+ /// readiness held open indefinitely would stall a rolling restart, a worse
failure than serving a cold
+ /// broker. When disabled this is a no-op and readiness behaves exactly as
before.
+ private void startWarmup() {
+ // The gauge is published in both branches so dashboards can rely on it
always existing; with warmup
+ // disabled it simply reads 1 from the start, matching pre-change
behaviour.
+ _brokerMetrics.setOrUpdateGlobalGauge(BrokerGauge.STARTUP_WARMUP_COMPLETE,
() -> _isWarm ? 1L : 0L);
+ if (!_warmupConfig.enabled()) {
+ _isWarm = true;
+ return;
+ }
+ _warmupThread = 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: readiness
+ // is already withheld until convergence by the Helix callbacks, so it
costs nothing.
+ long warmStartMs = 0L;
+ try {
+ long threadStartMs = System.currentTimeMillis();
+ awaitHelixConvergence();
+ warmStartMs = System.currentTimeMillis();
+ LOGGER.info("Helix converged after {} ms; starting data-plane warmup",
warmStartMs - threadStartMs);
+ // Saturating add: a pathologically large budgetMs must not overflow
the deadline negative (which
+ // would make warmup a silent no-op). A negative/zero budget still
yields a past deadline (no-op),
+ // which is the intended fail-open behaviour.
+ long budgetMs = _warmupConfig.budgetMs();
+ long deadlineMs = budgetMs > Long.MAX_VALUE - warmStartMs ?
Long.MAX_VALUE : warmStartMs + budgetMs;
+ boolean reachedFloor = _brokerRequestHandler.warmUp(_warmupConfig,
deadlineMs);
+ LOGGER.info("Broker warmup finished in {} ms (reachedFloor={})",
System.currentTimeMillis() - warmStartMs,
+ reachedFloor);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOGGER.info("Broker warmup interrupted before completion");
+ } catch (Throwable t) {
+ LOGGER.warn("Broker warmup threw; opening readiness anyway", t);
+ } finally {
+ _isWarm = true;
+ // Record the duration only if convergence was reached, so the metric
measures the warmup work
+ // itself and never the (unbounded) convergence wait -- e.g. when
shutdown interrupts the wait.
+ if (warmStartMs > 0L) {
+ _brokerMetrics.addTimedValue(BrokerTimer.STARTUP_WARMUP_DURATION_MS,
+ System.currentTimeMillis() - warmStartMs, TimeUnit.MILLISECONDS);
+ }
+ }
+ }, "broker-startup-warmup");
+ _warmupThread.setDaemon(true);
+ _warmupThread.start();
}
/// Runs startup server pre-connect on a background thread and ends startup
([#_isStarting]) when it
@@ -958,8 +1070,18 @@ public abstract class BaseBrokerStarter implements
ServiceStartable {
}
}
- /// 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.
+ /// Interrupts an in-flight warmup so a broker stopped mid-warmup does not
keep issuing probe queries
+ /// against a request handler being torn down, then joins briefly so an
in-flight probe unwinds before the
+ /// handler is shut down (avoiding a probe racing a closing [QueryRouter]).
Bounded: the thread is a daemon
+ /// whose `finally` opens the gate regardless, so shutdown never waits on it
beyond the short join.
+ private void stopWarmup() {
+ interruptAndJoin(_warmupThread, "broker warmup");
+ }
+
+ /// Interrupts an in-flight pre-connect so shutdown never waits on it.
Interrupt-only (no join) so this
+ /// PR leaves the already-merged pre-connect feature's shutdown behaviour
unchanged; the join added for
+ /// warmup applies to the warmup thread only. Best effort: the thread is a
daemon that records its metric
+ /// in a `finally` regardless.
private void stopPreConnect() {
Thread thread = _preConnectThread;
if (thread != null && thread.isAlive()) {
@@ -968,6 +1090,21 @@ public abstract class BaseBrokerStarter implements
ServiceStartable {
}
}
+ /// Interrupts `thread` (if alive) and waits up to
[#STARTUP_THREAD_JOIN_TIMEOUT_MS] for it to unwind.
+ /// Best effort: a thread that does not stop in time is left to its daemon
`finally` and shutdown proceeds.
+ private static void interruptAndJoin(@Nullable Thread thread, String what) {
+ if (thread == null || !thread.isAlive()) {
+ return;
+ }
+ LOGGER.info("Interrupting in-flight {} for shutdown", what);
+ thread.interrupt();
+ try {
+ thread.join(STARTUP_THREAD_JOIN_TIMEOUT_MS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
private String getDefaultBrokerId() {
try {
return InetAddress.getLocalHost().getHostName();
@@ -1000,6 +1137,7 @@ public abstract class BaseBrokerStarter implements
ServiceStartable {
public void stop() {
LOGGER.info("Shutting down Pinot broker");
_isShuttingDown = true;
+ stopWarmup();
stopPreConnect();
LOGGER.info("Disconnecting participant Helix manager");
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 c9dcc497277..57bb219d87d 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,33 @@ public interface BrokerRequestHandler {
void shutDown();
+ /// Warms this handler's data plane so the first real query does not pay for
it, and reports whether the
+ /// handler reached its warmth floor. Called during startup after Helix
convergence, before readiness is
+ /// granted.
+ ///
+ /// Implementations must honour three contracts, because the caller depends
on each:
+ /// - **Bounded — must return by `deadlineMs`.** This is the load-bearing
clause of the feature: a
+ /// warmup that runs past the deadline stalls a rolling restart.
`deadlineMs` is an absolute
+ /// [System#currentTimeMillis] value, and every wait an implementation
performs (each probe, each
+ /// blocking get) must be bounded by the remaining budget, not a fixed
constant. Time already spent
+ /// waiting for the cluster view to converge counts against the same
budget, so the rolling-restart
+ /// cost stays bounded by one number.
+ /// - **Must never throw.** The caller opens the readiness gate off this
call; an exception must be
+ /// swallowed and treated as "did not reach the floor", never propagated.
+ /// - **Safe to run against a handler that is already serving.** Nothing
stops this being invoked on a
+ /// started, traffic-serving handler --
[org.apache.pinot.integration.tests] does exactly that, and an
+ /// admin endpoint wiring it later would too -- so an implementation
must not mutate shared serving
+ /// state or assume it is the only in-flight work.
+ ///
+ /// This gate composes with any other startup gate the broker registers
(e.g. [#preConnectServers(long)]):
+ /// each gate is independent with its own budget, and readiness is granted
only once **all** are satisfied.
+ ///
+ /// @return `true` if the handler reached its warmth floor, `false` if the
deadline passed first. Either
+ /// way the caller proceeds; the result is for logging and metrics.
+ default boolean warmUp(BrokerWarmupConfig config, long deadlineMs) {
+ return true;
+ }
+
/// 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.
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 bca292de883..ee0586f2815 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
@@ -94,6 +94,14 @@ public class BrokerRequestHandlerDelegate implements
BrokerRequestHandler {
}
}
+ /// Warms only the single-stage handler. It owns the broker-to-server netty
channels, which is the data
+ /// plane that starts empty on a fresh broker; the multi-stage handler
already warms its own compile path
+ /// in `start()`, and the time-series handler shares the single-stage
transport.
+ @Override
+ public boolean warmUp(BrokerWarmupConfig config, long deadlineMs) {
+ return _singleStageBrokerRequestHandler.warmUp(config, deadlineMs);
+ }
+
@Override
public BrokerResponse handleRequest(JsonNode request, @Nullable
SqlNodeAndOptions sqlNodeAndOptions,
@Nullable RequesterIdentity requesterIdentity, RequestContext
requestContext, @Nullable HttpHeaders httpHeaders)
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BrokerWarmupConfig.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BrokerWarmupConfig.java
new file mode 100644
index 00000000000..b3225aac83f
--- /dev/null
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BrokerWarmupConfig.java
@@ -0,0 +1,60 @@
+/**
+ * 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 org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+
+
+/// Settings for the broker startup warmup performed before readiness is
granted.
+///
+/// Warmup always runs the same static probe -- `SELECT * FROM "<t>" LIMIT 1`
over a set-cover of tables
+/// spanning every routable server -- with no query or table overrides to
configure.
+///
+/// The exit condition is a **depth floor OR a budget ceiling**, never a
latency guess:
+/// - [#budgetMs] is a hard ceiling. Warmup releases the readiness gate when
it expires whatever the
+/// probe progress, so a slow or unreachable server cannot stall a rolling
restart.
+/// - [#minIterations] is the depth floor: it declares the broker warm once
this many probe queries have
+/// completed successfully -- enough invocations to drive the query path's
JIT to its top tier. (A
+/// latency target was deliberately avoided: a trivial probe reaches low
latency after a handful of
+/// iterations while the code is still only partially compiled, so latency
is a false early-exit signal.)
+///
+/// [#concurrency] is how many probes are fired at once per round: serial (1,
the default) warms the
+/// serve path; a higher value additionally warms the concurrency step the
first real burst hits.
+///
+/// Immutable; safe to share across threads.
+public record BrokerWarmupConfig(boolean enabled, long budgetMs, int
minIterations, int concurrency) {
+
+ /// Upper bound on the probe concurrency. Warmup runs before any real
traffic, so a modest cap is plenty;
+ /// this only exists so a fat-fingered config value cannot ask for a
pathological thread pool (an
+ /// OutOfMemoryError creating native threads), which would defeat the whole
point of warming up.
+ private static final int MAX_CONCURRENCY = 64;
+
+ public static BrokerWarmupConfig from(PinotConfiguration config) {
+ return new BrokerWarmupConfig(
+ config.getProperty(Broker.CONFIG_OF_BROKER_STARTUP_WARMUP_ENABLED,
+ Broker.DEFAULT_BROKER_STARTUP_WARMUP_ENABLED),
+ config.getProperty(Broker.CONFIG_OF_BROKER_STARTUP_WARMUP_BUDGET_MS,
+ Broker.DEFAULT_BROKER_STARTUP_WARMUP_BUDGET_MS),
+ Math.max(1,
config.getProperty(Broker.CONFIG_OF_BROKER_STARTUP_WARMUP_MIN_ITERATIONS,
+ Broker.DEFAULT_BROKER_STARTUP_WARMUP_MIN_ITERATIONS)),
+ Math.min(MAX_CONCURRENCY, Math.max(1,
config.getProperty(Broker.CONFIG_OF_BROKER_STARTUP_WARMUP_CONCURRENCY,
+ Broker.DEFAULT_BROKER_STARTUP_WARMUP_CONCURRENCY))));
+ }
+}
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 4fd4ef47bac..ea9364d8af4 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
@@ -20,13 +20,22 @@ package org.apache.pinot.broker.requesthandler;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Maps;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.IdentityHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;
@@ -37,13 +46,19 @@ import org.apache.pinot.common.config.TlsConfig;
import org.apache.pinot.common.config.provider.TableCache;
import org.apache.pinot.common.datatable.DataTable;
import org.apache.pinot.common.failuredetector.FailureDetector;
+import org.apache.pinot.common.metrics.BrokerGauge;
import org.apache.pinot.common.metrics.BrokerMeter;
+import org.apache.pinot.common.metrics.BrokerMetrics;
import org.apache.pinot.common.metrics.BrokerQueryPhase;
import org.apache.pinot.common.request.BrokerRequest;
+import org.apache.pinot.common.request.PinotQuery;
import org.apache.pinot.common.response.broker.BrokerResponseNative;
import org.apache.pinot.common.response.broker.QueryProcessingException;
+import org.apache.pinot.common.response.broker.ResultTable;
+import org.apache.pinot.common.utils.DataSchema;
import org.apache.pinot.common.utils.config.QueryOptionsUtils;
import org.apache.pinot.core.query.reduce.BrokerReduceService;
+import org.apache.pinot.core.routing.ImplicitHybridTableRouteInfo;
import org.apache.pinot.core.routing.MultiClusterRoutingContext;
import org.apache.pinot.core.routing.RoutingManager;
import org.apache.pinot.core.routing.TableRouteInfo;
@@ -60,9 +75,12 @@ 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;
+import org.apache.pinot.spi.query.QueryExecutionContext;
+import org.apache.pinot.spi.query.QueryThreadContext;
import org.apache.pinot.spi.trace.RequestContext;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
+import org.apache.pinot.sql.parsers.CalciteSqlCompiler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -72,6 +90,25 @@ import org.slf4j.LoggerFactory;
@ThreadSafe
public class SingleConnectionBrokerRequestHandler extends
BaseSingleStageBrokerRequestHandler {
private static final Logger LOGGER =
LoggerFactory.getLogger(SingleConnectionBrokerRequestHandler.class);
+ /// Per-probe cap, also bounded by whatever remains of the warmup budget.
+ private static final long WARMUP_PROBE_TIMEOUT_MS = 5_000L;
+ /// Upper bound on the local (stage 1) warmup iterations. Enough to drive
the trivial compile + serialize
+ /// path to JIT, but decoupled from `minIterations` (which governs the far
more expensive NETWORK probe
+ /// depth): a large `minIterations` must not turn stage 1 into a local spin
that eats the whole budget
+ /// before the network probe runs.
+ private static final int LOCAL_WARMUP_MAX_ITERATIONS = 2_000;
+ /// How long `warmUp` waits after `shutdownNow` for interrupted probe
workers to unwind before returning,
+ /// so a probe cannot outlive the call and race the request handler being
torn down on shutdown. Bounded
+ /// so shutdown never hangs on it; interrupted probes unwind well within
this.
+ private static final long PROBE_POOL_SHUTDOWN_WAIT_MS = 1_000L;
+ /// Backoff after an unproductive probe round (nothing routable / every
probe failed), so a not-yet-ready
+ /// cluster cannot busy-spin the warmup thread for the whole budget during
startup.
+ private static final long WARMUP_FAILURE_BACKOFF_MS = 100L;
+ /// Warmup reduces are best-effort JIT warming whose result is discarded, so
their metrics go to the shared
+ /// no-op instance instead of the broker's real counters -- otherwise
synthetic startup probes would
+ /// pollute latency timers, documentsScanned, and the per-table meters
before any real traffic. Safe to
+ /// share across the concurrent probe threads: every increment on a no-op
registry is a no-op.
+ private static final BrokerMetrics WARMUP_NOOP_METRICS =
BrokerMetrics.noop();
protected final BrokerReduceService _brokerReduceService;
protected final QueryRouter _queryRouter;
@@ -117,6 +154,505 @@ public class SingleConnectionBrokerRequestHandler extends
BaseSingleStageBrokerR
_brokerReduceService.shutDown();
}
+ /// Warms this broker's serve path before readiness is granted, in two
stages:
+ /// 1. a local, no-network pass ([#warmUpLocal]) that JIT-warms the
compile and response-serialization
+ /// paths, so they are warm even if no server is reachable; then
+ /// 2. a network probe -- repeatedly running real probe queries until they
have run enough times (the
+ /// `minIterations` depth floor) or the budget expires -- which warms
the scatter/gather/deserialize
+ /// and reduce paths and opens the broker-to-server channels lazily as
it goes.
+ ///
+ /// Probes go through [QueryRouter] rather than [#handleRequest],
deliberately bypassing access control,
+ /// query quota and the query log -- warmup then needs no synthetic identity
and pollutes no
+ /// customer-facing surface (probe reduces also use a throwaway metrics
registry, see
+ /// [#WARMUP_NOOP_METRICS]).
+ ///
+ /// Contract for the readiness caller: this **never throws** and **always
returns by `deadlineMs`** (an
+ /// absolute [System#currentTimeMillis] value). Every wait -- each probe and
each `Future#get` -- is
+ /// bounded by the remaining budget, and once it is spent all outstanding
probes are cancelled, so a slow
+ /// or unreachable server can never stall the rolling restart this gates.
The return value (reached the
+ /// floor vs. hit the budget) is for logging and metrics only; the caller
proceeds either way.
+ ///
+ /// Runs on the single `broker-startup-warmup` thread. The probe pool it
creates is local to the call and
+ /// shut down before returning, so there is no cross-call shared mutable
state.
+ @Override
+ public boolean warmUp(BrokerWarmupConfig config, long deadlineMs) {
+ // Stage 1: local, no-network warmup of the compile +
response-serialization paths, run before the
+ // network probe so they are warm even if no server is reachable. Looped
enough to JIT the serialization
+ // path (a single pass leaves it interpreted): minIterations iterations,
capped at 2000
+ // (LOCAL_WARMUP_MAX_ITERATIONS -- so the cap only bites if minIterations
is raised above 2000) and
+ // deadline-guarded, so it stays a quick prelude and never eats the budget
the network probe needs.
+ warmUpLocal(Math.min(config.minIterations(), LOCAL_WARMUP_MAX_ITERATIONS),
deadlineMs);
+ int concurrency = Math.max(1, config.concurrency());
+ ExecutorService probePool = Executors.newFixedThreadPool(concurrency,
+ new
ThreadFactoryBuilder().setNameFormat("broker-warmup-probe-%d").setDaemon(true).build());
+ // Tables at least one probe actually reached the servers for (probe()
adds to it from the pool threads,
+ // so it must be thread-safe). Read below to report per-server coverage --
only AFTER the pool drains, so
+ // no in-flight probe is still writing it.
+ Set<String> probedTables = ConcurrentHashMap.newKeySet();
+ try {
+ return warmUpNetwork(config, deadlineMs, probePool, concurrency,
probedTables);
+ } catch (Exception e) {
+ LOGGER.warn("Broker warmup failed; proceeding without it", e);
+ return false;
+ } finally {
+ probePool.shutdownNow();
+ awaitPoolDrain(probePool, PROBE_POOL_SHUTDOWN_WAIT_MS);
+ // Pool drained: probedTables is now stable. Report which routable
servers this run warmed (0 on a
+ // clean floor exit; non-zero if the budget expired before the
round-robin reached every server).
+ // Guarded so this finally can never make warmUp throw (the interface
contract is "never throws").
+ try {
+ reportServerCoverage(_routingManager, _brokerMetrics, probedTables);
+ } catch (Exception e) {
+ LOGGER.debug("Warmup coverage report failed (continuing)", e);
+ }
+ }
+ }
+
+ /// Waits up to `waitMs` for `pool` to terminate after a `shutdownNow`, so
an interrupted probe worker
+ /// unwinds before `warmUp` returns and cannot race the request handler
being torn down on shutdown.
+ ///
+ /// Crucially this **saves and clears** the interrupt flag around the wait.
On the shutdown path the warmup
+ /// thread is interrupted (`stopWarmup` -> `interrupt`), and
`awaitTermination` is interruptible -- with the
+ /// flag set it throws `InterruptedException` on entry and waits 0ms,
skipping the drain on exactly the path
+ /// it exists for. Clearing the flag lets the bounded wait actually happen;
the flag is restored afterwards.
+ /// Static and package-private so the interrupted-path behavior is
unit-testable.
+ @VisibleForTesting
+ static void awaitPoolDrain(ExecutorService pool, long waitMs) {
+ boolean interrupted = Thread.interrupted();
+ try {
+ if (!pool.awaitTermination(waitMs, TimeUnit.MILLISECONDS)) {
+ LOGGER.debug("Warmup probe pool did not fully terminate within {} ms;
proceeding", waitMs);
+ }
+ } catch (InterruptedException e) {
+ interrupted = true;
+ } finally {
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
+ /// Stage 1: local, no-network warmup. The network probe goes through
[QueryRouter] and stops at the
+ /// gathered DataTables, so it never exercises the response build + JSON
serialization the first real query
+ /// pays (the compile path it does re-warm, but only via a real table). This
compiles a throwaway query and
+ /// serializes a small synthetic [BrokerResponseNative] `iterations` times
so the serialization path
+ /// reaches JIT rather than staying interpreted after a single pass.
`iterations` is capped by the caller
+ /// (see [#LOCAL_WARMUP_MAX_ITERATIONS]); the loop is guarded by both
`deadlineMs` and the interrupt flag,
+ /// so stage 1 stays a quick prelude and stops promptly when shutdown
interrupts the warmup thread. Never
+ /// throws; never runs the network.
+ private void warmUpLocal(int iterations, long deadlineMs) {
+ try {
+ for (int i = 0; i < iterations && System.currentTimeMillis() < deadlineMs
+ && !Thread.currentThread().isInterrupted(); i++) {
+ CalciteSqlCompiler.compileToBrokerRequest("SELECT 1");
+ BrokerResponseNative response = new BrokerResponseNative();
+ response.setResultTable(new ResultTable(
+ new DataSchema(new String[]{"warmup"}, new
DataSchema.ColumnDataType[]{DataSchema.ColumnDataType.LONG}),
+ Collections.singletonList(new Object[]{1L})));
+ response.setNumDocsScanned(1);
+ response.toJsonString();
+ }
+ } catch (Exception e) {
+ LOGGER.debug("Local (stage 1) warmup failed; continuing", e);
+ }
+ }
+
+ /// Fires the given probe tasks concurrently on the shared pool and returns
the latencies (ms) of the
+ /// probes that completed successfully. Empty means the whole round was
unproductive (nothing routable
+ /// yet / all failed); a failed probe simply does not contribute.
+ ///
+ /// The total wait is bounded by `deadlineMs`, never by a fixed per-probe
constant: each `get()` waits
+ /// only the remaining budget, and once the deadline has passed every
still-outstanding future is
+ /// cancelled instead of waited on. This is what makes the budget a hard
ceiling even when more tasks were
+ /// submitted than the pool has threads (so tasks queue) -- otherwise a
queued straggler could hold the
+ /// readiness gate well past the budget.
+ ///
+ /// Static and package-private so the budget / cancellation / interrupt
semantics can be unit-tested with
+ /// injected probe callables, without a live cluster.
`firstProbeThrowLogged` is a 1-element per-run latch
+ /// so a probe that *throws* is surfaced once for the whole warmup run, not
once per round.
+ @VisibleForTesting
+ static List<Long> runConcurrentRound(List<Callable<Long>> tasks,
ExecutorService pool, long deadlineMs,
+ boolean[] firstProbeThrowLogged) {
+ List<Future<Long>> futures = new ArrayList<>(tasks.size());
+ for (Callable<Long> task : tasks) {
+ futures.add(pool.submit(task));
+ }
+ List<Long> latencies = new ArrayList<>(futures.size());
+ for (int i = 0; i < futures.size(); i++) {
+ long remainingMs = deadlineMs - System.currentTimeMillis();
+ if (remainingMs <= 0) {
+ // Budget spent: stop waiting and cancel everything still outstanding
so no probe outlives it.
+ cancelAll(futures, i);
+ break;
+ }
+ Future<Long> future = futures.get(i);
+ try {
+ Long elapsedMs = future.get(remainingMs, TimeUnit.MILLISECONDS);
+ if (elapsedMs != null && elapsedMs >= 0) {
+ latencies.add(elapsedMs);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ cancelAll(futures, i);
+ return latencies;
+ } catch (ExecutionException e) {
+ // The probe task itself threw (not a timeout). It does not count;
surface the first one of the whole
+ // run so a consistently broken probe is diagnosable without a debug
rebuild.
+ if (!firstProbeThrowLogged[0]) {
+ firstProbeThrowLogged[0] = true;
+ LOGGER.warn("Broker warmup probe threw (further occurrences
suppressed)", e.getCause());
+ }
+ future.cancel(true);
+ } catch (Exception e) {
+ // TimeoutException (probe overran the remaining budget) or
CancellationException: expected, does not
+ // count, and is cancelled so it does not keep running behind the next
round (no-op if already done).
+ future.cancel(true);
+ }
+ }
+ return latencies;
+ }
+
+ private static void cancelAll(List<Future<Long>> futures, int from) {
+ for (int j = from; j < futures.size(); j++) {
+ futures.get(j).cancel(true);
+ }
+ }
+
+ /// Sleeps [#WARMUP_FAILURE_BACKOFF_MS] after an unproductive probe round.
Returns `false` if interrupted
+ /// (shutdown) so the caller stops promptly and readiness opens.
+ private boolean warmupBackoff() {
+ try {
+ Thread.sleep(WARMUP_FAILURE_BACKOFF_MS);
+ return true;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ }
+
+ /// The network warmup: probe the static `SELECT * FROM "<t>" LIMIT 1` over
a greedy set-cover of tables
+ /// covering every routable server. Exits once at least `minIterations`
probes have completed successfully
+ /// -- a depth floor that guarantees enough invocations to drive the query
path's JIT to its top tier --
+ /// OR the budget expires. A latency target is deliberately not used: a
trivial probe reaches low latency
+ /// after a handful of iterations while the code is still only partially
compiled, so latency is a false
+ /// early-exit signal.
+ private boolean warmUpNetwork(BrokerWarmupConfig config, long deadlineMs,
ExecutorService pool,
+ int concurrency, Set<String> probedTables) {
+ long successfulProbes = 0;
+ int rounds = 0;
+ // Select the probe tables once and reuse them across rounds. Only
re-select while the result is empty
+ // (routing not populated yet); recomputing the greedy set-cover every
round would repeat O(tables) work
+ // up to minIterations times on a large-table tenant, for a result that
does not change once non-empty.
+ List<String> tables = List.of();
+ // Monotonic across rounds so the round-robin actually advances through
every covered table even at
+ // concurrency 1; resetting per round would probe only the first
`concurrency` tables forever.
+ int probeSeq = 0;
+ boolean reachedFloor = false;
+ // Per-run latch so a probe that throws is logged once for the whole run,
not once per round.
+ boolean[] firstProbeThrowLogged = new boolean[1];
+ while (System.currentTimeMillis() < deadlineMs &&
!Thread.currentThread().isInterrupted()) {
+ if (tables.isEmpty()) {
+ tables = selectProbeTables(_routingManager);
+ }
+ if (tables.isEmpty()) {
+ // Converged but nothing routable yet: back off and retry rather than
declaring the broker warm --
+ // an empty routing table here would otherwise make the gate a no-op
precisely on a cold broker.
+ if (!warmupBackoff()) {
+ return false;
+ }
+ continue;
+ }
+ if (System.currentTimeMillis() >= deadlineMs) {
+ break;
+ }
+ rounds++;
+ // A batch of `concurrency` probes, round-robin over the covered tables
(advancing across rounds via
+ // probeSeq): exercises every server (coverage) AND real concurrency
(contention) at once. Each task
+ // derives its own timeout from the deadline when it actually starts
(see probeWithinDeadline), so a
+ // task that queued behind the pool cannot overrun the budget.
+ List<Callable<Long>> tasks = new ArrayList<>(concurrency);
+ for (String tableNameWithType : roundRobinBatch(tables, probeSeq,
concurrency)) {
+ tasks.add(() -> probeWithinDeadline(compileProbe(tableNameWithType),
deadlineMs, probedTables));
+ }
+ probeSeq += concurrency;
+ List<Long> latencies = runConcurrentRound(tasks, pool, deadlineMs,
firstProbeThrowLogged);
+ if (latencies.isEmpty()) {
+ if (!warmupBackoff()) {
+ return false;
+ }
+ continue;
+ }
+ successfulProbes += latencies.size();
+ if (successfulProbes >= config.minIterations()) {
+ reachedFloor = true;
+ break;
+ }
+ }
+ // Coverage is reported by the caller after the pool drains (probedTables
must be stable). Here we only
+ // signal floor vs. budget.
+ if (reachedFloor) {
+ LOGGER.info("Broker warmup completed after {} round(s) at concurrency
{}; {} probes (floor {})", rounds,
+ concurrency, successfulProbes, config.minIterations());
+ return true;
+ }
+ logBudgetExpiry(rounds, concurrency, successfulProbes,
config.minIterations());
+ return false;
+ }
+
+ /// Logs warmup budget expiry: WARN only when nothing warmed
(`successfulProbes == 0`), INFO otherwise.
+ /// Expiring after warming some probes is the expected steady-state exit for
a probe too expensive to reach
+ /// the floor within the budget -- it warmed as much as the budget allowed
-- so it is not warning-worthy.
+ private static void logBudgetExpiry(int rounds, int concurrency, long
successfulProbes, int floor) {
+ if (successfulProbes == 0) {
+ LOGGER.warn("Broker warmup budget expired after {} round(s) at
concurrency {}; {} probes (floor {}). "
+ + "Proceeding to serve traffic (nothing warmed).", rounds,
concurrency, successfulProbes, floor);
+ } else {
+ LOGGER.info("Broker warmup budget expired after {} round(s) at
concurrency {}; {} probes (floor {}). "
+ + "Proceeding to serve traffic.", rounds, concurrency,
successfulProbes, floor);
+ }
+ }
+
+ /// Compiles the default probe query for a single physical table.
+ private BrokerRequest compileProbe(String tableNameWithType) {
+ return CalciteSqlCompiler.compileToBrokerRequest("SELECT * FROM \"" +
tableNameWithType + "\" LIMIT 1");
+ }
+
+ /// Runs one probe with a timeout derived from the remaining budget **at the
moment the task starts**,
+ /// never a fixed per-probe constant. When more tasks are submitted than the
pool has threads they queue,
+ /// and a queued task may not start until much of the budget is already
gone; deriving the timeout here
+ /// (rather than when the round was built) is what keeps the budget a hard
ceiling. Returns -1 (uncounted)
+ /// if the budget is already spent when the task starts, so no probe is
fired past the deadline.
+ private long probeWithinDeadline(BrokerRequest brokerRequest, long
deadlineMs, Set<String> probedTables) {
+ long timeoutMs = Math.min(deadlineMs - System.currentTimeMillis(),
WARMUP_PROBE_TIMEOUT_MS);
+ return timeoutMs <= 0 ? -1 : probe(brokerRequest, timeoutMs, probedTables);
+ }
+
+ /// Issues one probe query through [QueryRouter] and returns its wall-clock
duration in ms, or -1 on
+ /// failure. Builds the route the same way the normal path does, minus
auth/quota/logging. When at least
+ /// one server responds, records the (type-suffixed) table name in
`probedTables` so the caller can tell,
+ /// at exit, which routable servers were actually warmed.
+ private long probe(BrokerRequest brokerRequest, long timeoutMs, Set<String>
probedTables) {
+ String tableName = brokerRequest.getQuerySource().getTableName();
+ try {
+ String rawTableName = TableNameBuilder.extractRawTableName(tableName);
+ TableType tableType =
TableNameBuilder.getTableTypeFromTableName(tableName);
+ long requestId = _requestIdGenerator.get();
+ TableRouteInfo routeInfo =
+ _implicitHybridTableRouteProvider.getTableRouteInfo(rawTableName,
_tableCache, _routingManager);
+ if (!(routeInfo instanceof ImplicitHybridTableRouteInfo) ||
!routeInfo.isExists()) {
+ return -1;
+ }
+ // The probe table is always type-suffixed (the set-cover picks names
straight from getRoutableTables),
+ // so it targets exactly one type. Each leg's request must carry the
type-suffixed table name -- that
+ // is what routing resolves against -- so build a per-leg request from
the base query (as the normal
+ // path does). shouldRouteOffline/Realtime still handle the general case
for robustness.
+ BrokerRequest offlineBrokerRequest = shouldRouteOffline(tableType,
routeInfo)
+ ? typedRequest(brokerRequest, routeInfo.getOfflineTableName()) :
null;
+ BrokerRequest realtimeBrokerRequest = shouldRouteRealtime(tableType,
routeInfo)
+ ? typedRequest(brokerRequest, routeInfo.getRealtimeTableName()) :
null;
+ if (offlineBrokerRequest == null && realtimeBrokerRequest == null) {
+ // Neither leg is routable right now (e.g. a hybrid table probed via
its offline name before the
+ // time boundary exists). submitQuery asserts at least one leg is
non-null, so skip cleanly rather
+ // than build an empty route -- the table is retried on the next round
once its leg appears.
+ return -1;
+ }
+ // calculateRoutes sets the per-leg broker requests on the routeInfo
itself (nulling a leg whose
+ // routing table turns out empty), so we pass them as arguments and do
not pre-set them here.
+ _implicitHybridTableRouteProvider.calculateRoutes(routeInfo,
_routingManager, offlineBrokerRequest,
+ realtimeBrokerRequest, requestId);
+ if (!routeInfo.isRouteExists()) {
+ return -1;
+ }
+ long startMs = System.currentTimeMillis();
+ // Scatter/gather (+ DataTable deserialize on receive) is the dominant
cost and does not need a
+ // QueryThreadContext; run it directly so its warmth always counts
toward the probe latency.
+ // Side effect worth naming: submitQuery records
adaptive-server-selector stats (AsyncQueryResponse ->
+ // ServerRoutingStatsManager), so probes seed each server's latency EMA
before real traffic. A probe
+ // that gets a real response seeds the EMA with that server's (cold)
response latency; a probe that
+ // times out or errors seeds it with the full timeout (see
AsyncQueryResponse#getFinalResponses). On a
+ // healthy cluster every probe responds, so the seeds are comparable
across servers and the EMA decays
+ // to true warm latencies within a few real requests -- a small,
self-correcting bias. The one case
+ // that biases RELATIVE ordering is a server slow enough to time out
probes while its peers respond:
+ // it is seeded high and the selector routes less to it at first. That
is acceptable (it steers early
+ // traffic away from a genuinely slow server and self-corrects), and
still strictly better than the
+ // selector starting with no per-server history at all.
+ AsyncQueryResponse response = _queryRouter.submitQuery(requestId,
rawTableName, routeInfo, timeoutMs);
+ Map<ServerRoutingInstance, ServerResponse> finalResponses =
response.getFinalResponses();
+ // Coverage: if at least one server returned data, this table's servers
were reached this run. A table
+ // whose servers all timed out is deliberately NOT recorded, so it
counts as uncovered at exit.
+ for (ServerResponse serverResponse : finalResponses.values()) {
+ if (serverResponse.getDataTable() != null) {
+ probedTables.add(tableName);
+ break;
+ }
+ }
+ // Best-effort: also warm the broker reduce path (result discarded) so
the first real query of this
+ // shape does not pay it. Isolated in its own try -- reduceOnDataTable
requires a QueryThreadContext
+ // and could fail for edge cases; a reduce failure must never fail the
probe, since the
+ // scatter/gather/deserialize warmth has already happened.
+ try (QueryThreadContext ignore = QueryThreadContext.open(
+ new QueryExecutionContext(QueryExecutionContext.QueryType.SSE,
requestId, Long.toString(requestId),
+ "warmup", startMs, Long.MAX_VALUE, Long.MAX_VALUE, _brokerId,
_brokerId, ""), _threadAccountant)) {
+ Map<ServerRoutingInstance, DataTable> dataTableMap = new
HashMap<>(finalResponses.size());
+ for (Map.Entry<ServerRoutingInstance, ServerResponse> entry :
finalResponses.entrySet()) {
+ DataTable dataTable = entry.getValue().getDataTable();
+ if (dataTable != null) {
+ dataTableMap.put(entry.getKey(), dataTable);
+ }
+ }
+ if (!dataTableMap.isEmpty()) {
+ // Discard-only warmth: route metrics to the throwaway registry so
the probe does not pollute the
+ // broker's real counters (see WARMUP_NOOP_METRICS).
+ _brokerReduceService.reduceOnDataTable(brokerRequest, brokerRequest,
dataTableMap, timeoutMs,
+ WARMUP_NOOP_METRICS);
+ }
+ } catch (Exception reduceEx) {
+ LOGGER.debug("Warmup reduce step failed (continuing); scatter/gather
already warmed", reduceEx);
+ }
+ return System.currentTimeMillis() - startMs;
+ } catch (Exception e) {
+ if (e instanceof InterruptedException) {
+ // Shutdown interrupted this probe (via shutdownNow); restore the flag
so the pool worker unwinds.
+ Thread.currentThread().interrupt();
+ }
+ LOGGER.debug("Warmup probe failed for query on table: {}", tableName, e);
+ return -1;
+ }
+ }
+
+ /// Builds a per-leg probe request carrying the type-suffixed table name
that routing resolves against,
+ /// from the base (possibly raw-named) request. Deep-copies so the base
request is not mutated.
+ private static BrokerRequest typedRequest(BrokerRequest base, String
tableNameWithType) {
+ PinotQuery pinotQuery = base.getPinotQuery().deepCopy();
+ pinotQuery.getDataSource().setTableName(tableNameWithType);
+ return CalciteSqlCompiler.convertToBrokerRequest(pinotQuery);
+ }
+
+ /// Whether a probe routes the offline leg: the query targets offline (a
type-suffixed offline name) or is
+ /// untyped (a raw name, `tableType` null), AND an offline table exists.
Static so the offline / realtime /
+ /// hybrid decision is unit-testable against a mocked route without a live
cluster.
+ @VisibleForTesting
+ static boolean shouldRouteOffline(@Nullable TableType tableType,
TableRouteInfo routeInfo) {
+ return tableType != TableType.REALTIME && routeInfo.hasOffline();
+ }
+
+ /// Whether a probe routes the realtime leg: the query targets realtime or
is untyped, AND a realtime
+ /// table exists.
+ @VisibleForTesting
+ static boolean shouldRouteRealtime(@Nullable TableType tableType,
TableRouteInfo routeInfo) {
+ return tableType != TableType.OFFLINE && routeInfo.hasRealtime();
+ }
+
+ /// Picks the fewest tables that between them cover **every** routable
server -- the one, only behavior:
+ /// warmup always covers the whole fleet, with no cap to tune.
+ ///
+ /// Greedy set cover rather than "all tables" or a configured list: warmup
needs every broker-to-server
+ /// channel and the whole query path exercised, and coverage delivers
exactly that. The loop stops as soon
+ /// as every server is covered, so it self-limits to at most one table per
server (never the full table
+ /// list) even on a tenant with thousands of tables. Because the coverage
universe is exactly the servers
+ /// some routable table serves, this always achieves full coverage. Static
so it can be unit-tested against
+ /// a mocked [RoutingManager] without constructing a broker.
+ @VisibleForTesting
+ static List<String> selectProbeTables(RoutingManager routingManager) {
+ Set<String> uncoveredServers = routableServers(routingManager);
+ // Sorted so selection is deterministic across brokers and across restarts.
+ List<String> candidates = new
ArrayList<>(routingManager.getRoutableTables());
+ Collections.sort(candidates);
+ // Capacity bounded by the server count -- the true ceiling on how many
tables get selected.
+ List<String> selected = new ArrayList<>(Math.min(uncoveredServers.size(),
candidates.size()));
+ for (String tableNameWithType : candidates) {
+ if (uncoveredServers.isEmpty()) {
+ break;
+ }
+ Set<String> serving =
routingManager.getServingInstances(tableNameWithType);
+ if (serving == null || serving.isEmpty()) {
+ continue;
+ }
+ if (uncoveredServers.removeAll(serving)) {
+ selected.add(tableNameWithType);
+ }
+ }
+ return selected;
+ }
+
+ /// Records [BrokerGauge#STARTUP_WARMUP_UNCOVERED_SERVERS] on warmup exit:
the routable servers that
+ /// `probedTables` (the tables at least one probe actually reached this run)
do NOT, between them, route
+ /// to. The set-cover guarantees the *selected* tables span every server, so
this is `0` when warmup reaches
+ /// its floor (every table probed); it goes non-zero only when warmup exits
early -- the budget expired, or
+ /// servers were too slow, before the round-robin reached every table.
+ ///
+ /// It is a warmup **completeness** signal, not "these servers are stone
cold": the broker's serve-path JIT
+ /// is warmed per-JVM (not per-server) and channels are opened by
pre-connect, so an unreached server is
+ /// only marginally colder. Read a non-zero value as "warmup ran out of
budget before its intended
+ /// coverage", most meaningful alongside whether the floor was reached.
+ ///
+ /// Uncovered is derived from a single [#routableServers] read (via
[#uncoveredRoutableServers]) so the
+ /// count is self-consistent. Static (collaborators as parameters) so the
emission is unit-testable against
+ /// a mocked [RoutingManager] and [BrokerMetrics].
+ @VisibleForTesting
+ static void reportServerCoverage(RoutingManager routingManager,
BrokerMetrics brokerMetrics,
+ Collection<String> probedTables) {
+ Set<String> uncovered = uncoveredRoutableServers(routingManager,
probedTables);
+
brokerMetrics.setValueOfGlobalGauge(BrokerGauge.STARTUP_WARMUP_UNCOVERED_SERVERS,
uncovered.size());
+ if (!uncovered.isEmpty()) {
+ LOGGER.warn("Broker warmup did not reach {} routable server(s) before it
exited (budget expired or "
+ + "servers too slow): {}. Serve-path JIT is warmed regardless; those
servers are only marginally "
+ + "colder.", uncovered.size(), uncovered);
+ }
+ }
+
+ /// Returns the routable servers that none of the given probe tables route
to. Empty means the tables
+ /// span every server this broker can reach. Static so it can be unit-tested
against a mocked
+ /// [RoutingManager].
+ @VisibleForTesting
+ static Set<String> uncoveredRoutableServers(RoutingManager routingManager,
Collection<String> tables) {
+ Set<String> uncovered = routableServers(routingManager);
+ for (String tableNameWithType : tables) {
+ Set<String> serving =
routingManager.getServingInstances(tableNameWithType);
+ if (serving != null) {
+ uncovered.removeAll(serving);
+ }
+ }
+ return uncovered;
+ }
+
+ /// The servers this broker actually routes to: the union of the serving
instances of its routable tables.
+ ///
+ /// Deliberately NOT `getRoutableServerInstanceMap()`, which is every
enabled server in the whole cluster
+ /// (no tenant filter) -- on a multi-tenant cluster that would count other
tenants' servers this broker
+ /// never queries, so the coverage metric would be permanently non-zero and
its warning unactionable. This
+ /// mirrors how startup pre-connect derives its channels from routing.
+ @VisibleForTesting
+ static Set<String> routableServers(RoutingManager routingManager) {
+ Set<String> servers = new HashSet<>();
+ for (String tableNameWithType : routingManager.getRoutableTables()) {
+ Set<String> serving =
routingManager.getServingInstances(tableNameWithType);
+ if (serving != null) {
+ servers.addAll(serving);
+ }
+ }
+ return servers;
+ }
+
+ /// Returns the next `concurrency` items to probe, round-robin starting at
`startSeq`. Kept generic and
+ /// static so the rotation -- which must advance across rounds, not reset
each round, or only the first
+ /// `concurrency` items would ever be probed at low concurrency -- is
unit-testable without a live probe.
+ /// [Math#floorMod(int,int)] keeps the index valid even if `startSeq`
overflows to a negative value.
+ @VisibleForTesting
+ static <T> List<T> roundRobinBatch(List<T> items, int startSeq, int
concurrency) {
+ int size = items.size();
+ if (size == 0) {
+ // Defensive: an empty list would make the floorMod below divide by
zero. Callers already guard, but a
+ // future caller must get an empty batch, not an ArithmeticException.
+ return List.of();
+ }
+ List<T> batch = new ArrayList<>(concurrency);
+ for (int i = 0; i < concurrency; i++) {
+ batch.add(items.get(Math.floorMod(startSeq + i, size)));
+ }
+ return batch;
+ }
+
/// 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
diff --git
a/pinot-broker/src/test/java/org/apache/pinot/broker/broker/helix/BrokerWarmupGateTest.java
b/pinot-broker/src/test/java/org/apache/pinot/broker/broker/helix/BrokerWarmupGateTest.java
new file mode 100644
index 00000000000..c1a11d7cdfc
--- /dev/null
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/broker/helix/BrokerWarmupGateTest.java
@@ -0,0 +1,50 @@
+/**
+ * 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.broker.helix;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.pinot.common.utils.ServiceStatus;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+
+
+/// Deterministic coverage for the startup-warmup readiness gate's STARTING ->
GOOD transition. Observing a
+/// live broker mid-warmup is inherently racy (warmup against a healthy
cluster completes in well under a
+/// poll interval), so the gate callback is tested directly against a flag
that stands in for `_isWarm` --
+/// the same flag the runtime callback closes over.
+public class BrokerWarmupGateTest {
+
+ @Test
+ public void gateReportsStartingUntilWarmThenGood() {
+ AtomicBoolean warm = new AtomicBoolean(false);
+ ServiceStatus.ServiceStatusCallback gate =
BaseBrokerStarter.warmupGateCallback(warm::get);
+
+ // While warming: STARTING (an existing status value, not a new enum
constant) with the warming-up
+ // description, so the readiness probe holds the broker out of rotation
and the reason is visible.
+ assertEquals(gate.getServiceStatus(), ServiceStatus.Status.STARTING);
+ assertEquals(gate.getStatusDescription(),
BaseBrokerStarter.WARMUP_GATE_STARTING_DESCRIPTION);
+
+ // The same callback flips the instant the flag does, exactly as _isWarm
does at runtime: GOOD, no
+ // description. No new callback is installed -- the gate transitions in
place.
+ warm.set(true);
+ assertEquals(gate.getServiceStatus(), ServiceStatus.Status.GOOD);
+ assertEquals(gate.getStatusDescription(),
ServiceStatus.STATUS_DESCRIPTION_NONE);
+ }
+}
diff --git
a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/BrokerWarmupTest.java
b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/BrokerWarmupTest.java
new file mode 100644
index 00000000000..457033f9066
--- /dev/null
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/BrokerWarmupTest.java
@@ -0,0 +1,454 @@
+/**
+ * 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.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.pinot.common.metrics.BrokerGauge;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.core.routing.RoutingManager;
+import org.apache.pinot.core.routing.TableRouteInfo;
+import org.apache.pinot.core.transport.ServerInstance;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.testng.annotations.Test;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+
+
+/// Unit coverage for broker startup warmup: config parsing (including the
concurrency and minIterations
+/// knobs and their floors), the no-op interface default, the greedy table
selection that always covers every
+/// routable server, the static-probe routing decision, the round-robin
rotation, the server-coverage gauge,
+/// and the concurrent-round budget semantics.
+public class BrokerWarmupTest {
+
+ @Test
+ public void warmupIsDisabledByDefault() {
+ BrokerWarmupConfig config = BrokerWarmupConfig.from(new
PinotConfiguration());
+ assertFalse(config.enabled());
+ // Defaults still parse, so flipping the flag alone yields sane behavior.
+ assertEquals(config.budgetMs(),
Broker.DEFAULT_BROKER_STARTUP_WARMUP_BUDGET_MS);
+ assertEquals(config.minIterations(),
Broker.DEFAULT_BROKER_STARTUP_WARMUP_MIN_ITERATIONS);
+ // Concurrency defaults to serial.
+ assertEquals(config.concurrency(),
Broker.DEFAULT_BROKER_STARTUP_WARMUP_CONCURRENCY);
+ assertEquals(config.concurrency(), 1);
+ }
+
+ @Test
+ public void configIsReadFromProperties() {
+ PinotConfiguration properties = new PinotConfiguration();
+ properties.setProperty(Broker.CONFIG_OF_BROKER_STARTUP_WARMUP_ENABLED,
true);
+ properties.setProperty(Broker.CONFIG_OF_BROKER_STARTUP_WARMUP_BUDGET_MS,
4321L);
+
properties.setProperty(Broker.CONFIG_OF_BROKER_STARTUP_WARMUP_MIN_ITERATIONS,
250);
+ properties.setProperty(Broker.CONFIG_OF_BROKER_STARTUP_WARMUP_CONCURRENCY,
12);
+
+ BrokerWarmupConfig config = BrokerWarmupConfig.from(properties);
+ assertTrue(config.enabled());
+ assertEquals(config.budgetMs(), 4321L);
+ assertEquals(config.minIterations(), 250);
+ assertEquals(config.concurrency(), 12);
+ }
+
+ /// concurrency and minIterations are floored at 1 even if misconfigured to
0/negative, so the probe pool
+ /// and the exit floor are always valid.
+ @Test
+ public void concurrencyAndMinIterationsFlooredAtOne() {
+ PinotConfiguration properties = new PinotConfiguration();
+ properties.setProperty(Broker.CONFIG_OF_BROKER_STARTUP_WARMUP_CONCURRENCY,
0);
+
properties.setProperty(Broker.CONFIG_OF_BROKER_STARTUP_WARMUP_MIN_ITERATIONS,
-5);
+ BrokerWarmupConfig config = BrokerWarmupConfig.from(properties);
+ assertEquals(config.concurrency(), 1);
+ assertEquals(config.minIterations(), 1);
+ }
+
+ /// concurrency is clamped to an upper bound so a fat-fingered value cannot
ask for a pathological thread
+ /// pool (an OutOfMemoryError creating native threads) at startup.
+ @Test
+ public void concurrencyIsClampedToAnUpperBound() {
+ PinotConfiguration properties = new PinotConfiguration();
+ properties.setProperty(Broker.CONFIG_OF_BROKER_STARTUP_WARMUP_CONCURRENCY,
1_000_000);
+ BrokerWarmupConfig config = BrokerWarmupConfig.from(properties);
+ assertTrue(config.concurrency() >= 1 && config.concurrency() <= 64,
+ "concurrency must be clamped into [1, 64] but was " +
config.concurrency());
+ }
+
+ /// A handler that does not override warmUp must report warm immediately. If
the default blocked or
+ /// returned false, every non-single-stage handler would hold readiness shut
forever.
+ @Test
+ public void defaultWarmUpIsANoOp() {
+ BrokerRequestHandler handler = mock(BrokerRequestHandler.class);
+ when(handler.warmUp(any(), anyLong())).thenCallRealMethod();
+ assertTrue(handler.warmUp(new BrokerWarmupConfig(true, 1L, 1, 1),
+ System.currentTimeMillis() + 1_000L));
+ }
+
+ /// Three tables, three servers, one table per server: all three must be
picked, since dropping any one
+ /// leaves a broker-to-server channel unwarmed.
+ @Test
+ public void selectionCoversEveryServer() {
+ RoutingManager routing = routing(
+ Map.of("a_OFFLINE", Set.of("s1"), "b_OFFLINE", Set.of("s2"),
"c_OFFLINE", Set.of("s3")),
+ Set.of("s1", "s2", "s3"));
+
+ assertEquals(new
java.util.HashSet<>(SingleConnectionBrokerRequestHandler.selectProbeTables(routing)),
+ Set.of("a_OFFLINE", "b_OFFLINE", "c_OFFLINE"));
+ }
+
+ /// A table adding no new server is skipped: probing it costs a query and
warms nothing extra.
+ @Test
+ public void selectionSkipsRedundantTables() {
+ RoutingManager routing = routing(
+ Map.of("a_OFFLINE", Set.of("s1", "s2"), "b_OFFLINE", Set.of("s1"),
"c_OFFLINE", Set.of("s2")),
+ Set.of("s1", "s2"));
+
+ // "a_OFFLINE" sorts first and already covers both servers, so nothing
else is needed.
+
assertEquals(SingleConnectionBrokerRequestHandler.selectProbeTables(routing),
List.of("a_OFFLINE"));
+ }
+
+ /// The one, only behavior: the set-cover always spans EVERY routable
server, however many tables that
+ /// needs (there is no cap). The loop still stops at full coverage, so it
picks exactly one table per
+ /// server here and no more.
+ @Test
+ public void selectionCoversWideFleet() {
+ // Seven servers, each served by exactly one distinct table: covering all
of them needs all seven tables.
+ RoutingManager routing = routing(
+ Map.of("a_OFFLINE", Set.of("s1"), "b_OFFLINE", Set.of("s2"),
"c_OFFLINE", Set.of("s3"),
+ "d_OFFLINE", Set.of("s4"), "e_OFFLINE", Set.of("s5"), "f_OFFLINE",
Set.of("s6"),
+ "g_OFFLINE", Set.of("s7")),
+ Set.of("s1", "s2", "s3", "s4", "s5", "s6", "s7"));
+
+ List<String> selected =
SingleConnectionBrokerRequestHandler.selectProbeTables(routing);
+ assertEquals(selected.size(), 7);
+
assertTrue(SingleConnectionBrokerRequestHandler.uncoveredRoutableServers(routing,
selected).isEmpty());
+ }
+
+ /// Warmup runs before the first query and may legitimately find nothing
routable yet.
+ @Test
+ public void selectionHandlesEmptyCluster() {
+
assertTrue(SingleConnectionBrokerRequestHandler.selectProbeTables(routing(Map.of(),
Set.of())).isEmpty());
+ }
+
+ /// Must not blow up on a table whose serving-instance set is empty or
unknown.
+ @Test
+ public void selectionIgnoresTablesWithoutServingInstances() {
+ RoutingManager routing = routing(Map.of("a_OFFLINE", Set.of(),
"b_OFFLINE", Set.of("s1")), Set.of("s1"));
+
assertEquals(SingleConnectionBrokerRequestHandler.selectProbeTables(routing),
List.of("b_OFFLINE"));
+ }
+
+ /// The auto-select set-cover always leaves nothing uncovered -- every
routable server is spanned.
+ @Test
+ public void coverageIsCompleteForAutoSelect() {
+ RoutingManager routing = routing(
+ Map.of("a_OFFLINE", Set.of("s1"), "b_OFFLINE", Set.of("s2"),
"c_OFFLINE", Set.of("s3"),
+ "d_OFFLINE", Set.of("s4")),
+ Set.of("s1", "s2", "s3", "s4"));
+
+ List<String> selected =
SingleConnectionBrokerRequestHandler.selectProbeTables(routing);
+
assertTrue(SingleConnectionBrokerRequestHandler.uncoveredRoutableServers(routing,
selected).isEmpty());
+ }
+
+ /// uncoveredRoutableServers surfaces the servers a given table set misses:
probing only a_OFFLINE (which
+ /// serves s1) leaves s2 uncovered.
+ @Test
+ public void uncoveredRoutableServersIdentifiesMissedServers() {
+ RoutingManager routing = routing(
+ Map.of("a_OFFLINE", Set.of("s1"), "b_OFFLINE", Set.of("s2")),
Set.of("s1", "s2"));
+
+
assertEquals(SingleConnectionBrokerRequestHandler.uncoveredRoutableServers(routing,
List.of("a_OFFLINE")),
+ Set.of("s2"));
+ }
+
+ /// At concurrency 1 the round-robin must advance across rounds (startSeq
monotonic) so successive batches
+ /// walk every item and wrap. A per-round reset would probe only
items.get(0) forever.
+ @Test
+ public void roundRobinBatchWalksAllItemsAtConcurrencyOne() {
+ List<String> items = List.of("a", "b", "c");
+ assertEquals(SingleConnectionBrokerRequestHandler.roundRobinBatch(items,
0, 1), List.of("a"));
+ assertEquals(SingleConnectionBrokerRequestHandler.roundRobinBatch(items,
1, 1), List.of("b"));
+ assertEquals(SingleConnectionBrokerRequestHandler.roundRobinBatch(items,
2, 1), List.of("c"));
+ assertEquals(SingleConnectionBrokerRequestHandler.roundRobinBatch(items,
3, 1), List.of("a"));
+ }
+
+ /// A concurrency batch wraps around the item list, and an overflowed
(negative) startSeq still yields a
+ /// valid in-range index rather than throwing (Math.floorMod).
+ @Test
+ public void roundRobinBatchWrapsAndSurvivesOverflow() {
+ List<String> items = List.of("a", "b", "c");
+ assertEquals(SingleConnectionBrokerRequestHandler.roundRobinBatch(items,
0, 2), List.of("a", "b"));
+ assertEquals(SingleConnectionBrokerRequestHandler.roundRobinBatch(items,
2, 2), List.of("c", "a"));
+ // Batch larger than the list repeats items.
+ assertEquals(SingleConnectionBrokerRequestHandler.roundRobinBatch(items,
0, 4), List.of("a", "b", "c", "a"));
+ // startSeq + i overflows int on the second element; must not throw and
must stay in range.
+ assertEquals(SingleConnectionBrokerRequestHandler.roundRobinBatch(items,
Integer.MAX_VALUE, 2).size(), 2);
+ }
+
+ /// An empty item list must yield an empty batch, not an ArithmeticException
from `floorMod(x, 0)`. Callers
+ /// guard today, but the helper must be safe for a future caller.
+ @Test
+ public void roundRobinBatchOnEmptyListIsEmpty() {
+ assertTrue(SingleConnectionBrokerRequestHandler.roundRobinBatch(List.of(),
0, 4).isEmpty());
+
assertTrue(SingleConnectionBrokerRequestHandler.roundRobinBatch(List.<String>of(),
7, 1).isEmpty());
+ }
+
+ /// reportServerCoverage records, at exit, the servers left cold given the
tables actually probed: 1 when
+ /// only one of two servers' tables was probed (the other missed, e.g.
budget expired first), 0 when every
+ /// server's table was probed. A cluster server no routable table routes to
is excluded from the universe.
+ @Test
+ public void reportServerCoverageRecordsUncoveredGauge() {
+ // s3 is in the cluster-wide server map but NO table routes to it --
another tenant's server. The
+ // coverage universe is the servers THIS broker routes to (union of
serving instances over its routable
+ // tables), so s3 must NOT be counted as uncovered.
+ RoutingManager routing = routing(
+ Map.of("a_OFFLINE", Set.of("s1"), "b_OFFLINE", Set.of("s2")),
Set.of("s1", "s2", "s3"));
+ BrokerMetrics metrics = mock(BrokerMetrics.class);
+
+ // Only a_OFFLINE was probed (b_OFFLINE never reached -- budget expired)
-> s2 left cold; s3 excluded.
+ SingleConnectionBrokerRequestHandler.reportServerCoverage(routing,
metrics, List.of("a_OFFLINE"));
+
verify(metrics).setValueOfGlobalGauge(BrokerGauge.STARTUP_WARMUP_UNCOVERED_SERVERS,
1L);
+
+ // Both tables probed -> every server warmed.
+ SingleConnectionBrokerRequestHandler.reportServerCoverage(routing, metrics,
+ List.of("a_OFFLINE", "b_OFFLINE"));
+
verify(metrics).setValueOfGlobalGauge(BrokerGauge.STARTUP_WARMUP_UNCOVERED_SERVERS,
0L);
+
+ RoutingManager full = routing(Map.of("a_OFFLINE", Set.of("s1", "s2")),
Set.of("s1", "s2"));
+ BrokerMetrics fullMetrics = mock(BrokerMetrics.class);
+ SingleConnectionBrokerRequestHandler.reportServerCoverage(full,
fullMetrics, List.of("a_OFFLINE"));
+
verify(fullMetrics).setValueOfGlobalGauge(BrokerGauge.STARTUP_WARMUP_UNCOVERED_SERVERS,
0L);
+ }
+
+ /// The offline/realtime/hybrid routing decision the probe uses: a
type-suffixed name routes only that
+ /// type; a type with no matching table routes nothing. (The untyped/`null`
cases are covered too, as a
+ /// general contract of the helper, though the static probe always supplies
a type-suffixed name.)
+ @Test
+ public void probeRoutingDecisionCoversOfflineRealtimeHybrid() {
+ TableRouteInfo offlineOnly = route(true, false);
+ TableRouteInfo realtimeOnly = route(false, true);
+ TableRouteInfo hybrid = route(true, true);
+
+ assertTrue(SingleConnectionBrokerRequestHandler.shouldRouteOffline(null,
offlineOnly));
+ assertFalse(SingleConnectionBrokerRequestHandler.shouldRouteRealtime(null,
offlineOnly));
+ assertFalse(SingleConnectionBrokerRequestHandler.shouldRouteOffline(null,
realtimeOnly));
+ assertTrue(SingleConnectionBrokerRequestHandler.shouldRouteRealtime(null,
realtimeOnly));
+ assertTrue(SingleConnectionBrokerRequestHandler.shouldRouteOffline(null,
hybrid));
+ assertTrue(SingleConnectionBrokerRequestHandler.shouldRouteRealtime(null,
hybrid));
+
+
assertTrue(SingleConnectionBrokerRequestHandler.shouldRouteOffline(TableType.OFFLINE,
hybrid));
+
assertFalse(SingleConnectionBrokerRequestHandler.shouldRouteRealtime(TableType.OFFLINE,
hybrid));
+
assertFalse(SingleConnectionBrokerRequestHandler.shouldRouteOffline(TableType.REALTIME,
hybrid));
+
assertTrue(SingleConnectionBrokerRequestHandler.shouldRouteRealtime(TableType.REALTIME,
hybrid));
+ // A realtime-typed query against an offline-only table routes nothing.
+
assertFalse(SingleConnectionBrokerRequestHandler.shouldRouteOffline(TableType.REALTIME,
offlineOnly));
+
assertFalse(SingleConnectionBrokerRequestHandler.shouldRouteRealtime(TableType.REALTIME,
offlineOnly));
+ }
+
+ /// A round collects the latencies of probes that completed with a
non-negative value, and silently drops
+ /// a probe that failed (returned -1) or threw. This is the "everything
finishes within budget" happy path.
+ @Test
+ public void runConcurrentRoundCollectsSuccessfulLatenciesOnly() {
+ ExecutorService pool = Executors.newFixedThreadPool(4);
+ try {
+ long deadline = System.currentTimeMillis() + 5_000L;
+ List<Callable<Long>> tasks = List.of(
+ () -> 10L,
+ () -> -1L, // failed probe:
excluded
+ () -> 20L,
+ () -> {
+ throw new RuntimeException("probe blew up"); // threw: excluded,
must not fail the round
+ });
+ List<Long> latencies =
+ SingleConnectionBrokerRequestHandler.runConcurrentRound(tasks, pool,
deadline, new boolean[1]);
+ assertEquals(new HashSet<>(latencies), Set.of(10L, 20L));
+ assertEquals(latencies.size(), 2);
+ } finally {
+ pool.shutdownNow();
+ }
+ }
+
+ /// The budget is a hard ceiling even when more probes are submitted than
the pool has threads: probes 2
+ /// and 3 queue behind probe 1, and every probe sleeps far past the budget.
The round must return at ~the
+ /// budget (never sum-of-sleeps), collect nothing, and not leave the queued
probes to run -- the exact
+ /// overshoot that #1/#2 fixed.
+ @Test
+ public void runConcurrentRoundReturnsByBudgetWhenTasksQueue() {
+ ExecutorService pool = Executors.newFixedThreadPool(1);
+ try {
+ AtomicInteger started = new AtomicInteger(0);
+ List<Callable<Long>> tasks = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ tasks.add(() -> {
+ started.incrementAndGet();
+ Thread.sleep(5_000L);
+ return 1L;
+ });
+ }
+ long deadline = System.currentTimeMillis() + 300L;
+ long start = System.currentTimeMillis();
+ List<Long> latencies =
+ SingleConnectionBrokerRequestHandler.runConcurrentRound(tasks, pool,
deadline, new boolean[1]);
+ long elapsed = System.currentTimeMillis() - start;
+ assertTrue(elapsed < 3_000L, "round must return near the 300ms budget,
took " + elapsed + "ms");
+ assertTrue(latencies.isEmpty(), "no 5s probe can finish within a 300ms
budget");
+ // Only the head-of-queue probe ever ran; the two queued behind it were
cancelled, not fired. (<=2
+ // rather than ==1 only to tolerate the microsecond window where probe
1's interrupt frees the single
+ // pool thread before cancelAll marks probe 2 cancelled.)
+ assertTrue(started.get() <= 2, "queued probes must not run past the
budget, started=" + started.get());
+ } finally {
+ pool.shutdownNow();
+ }
+ }
+
+ /// A deadline already in the past collects nothing and returns at once --
it never waits on a submitted
+ /// probe. Guards the loop's leading remaining-budget check.
+ @Test
+ public void runConcurrentRoundWithExpiredDeadlineCollectsNothing() {
+ ExecutorService pool = Executors.newFixedThreadPool(2);
+ try {
+ List<Callable<Long>> tasks = List.of(() -> 1L, () -> 2L);
+ List<Long> latencies =
SingleConnectionBrokerRequestHandler.runConcurrentRound(tasks, pool,
+ System.currentTimeMillis() - 1L, new boolean[1]);
+ assertTrue(latencies.isEmpty());
+ } finally {
+ pool.shutdownNow();
+ }
+ }
+
+ /// Interrupting the thread running the round (shutdown mid-warmup) must end
it promptly -- not wait out the
+ /// in-flight probes -- return what it had, and preserve the interrupt
status so the warmup loop above it
+ /// sees the interrupt and stops.
+ @Test
+ public void runConcurrentRoundReturnsPromptlyWhenInterrupted() throws
Exception {
+ ExecutorService pool = Executors.newFixedThreadPool(2);
+ try {
+ List<Callable<Long>> tasks = List.of(
+ () -> {
+ Thread.sleep(10_000L);
+ return 1L;
+ },
+ () -> {
+ Thread.sleep(10_000L);
+ return 2L;
+ });
+ // Deadline far off, so only the interrupt -- not the budget -- can end
the round.
+ long deadline = System.currentTimeMillis() + 30_000L;
+ AtomicReference<List<Long>> result = new AtomicReference<>();
+ AtomicBoolean interruptPreserved = new AtomicBoolean(false);
+ Thread runner = new Thread(() -> {
+ List<Long> r =
SingleConnectionBrokerRequestHandler.runConcurrentRound(tasks, pool, deadline,
new boolean[1]);
+ result.set(r);
+ interruptPreserved.set(Thread.currentThread().isInterrupted());
+ });
+ runner.start();
+ Thread.sleep(200L); // let it enter Future.get()
+ runner.interrupt();
+ runner.join(3_000L);
+ assertFalse(runner.isAlive(), "interrupt must end the round promptly,
not wait 10s for the probes");
+ assertTrue(result.get().isEmpty());
+ assertTrue(interruptPreserved.get(), "interrupt status must be re-set
for the caller loop to observe");
+ } finally {
+ pool.shutdownNow();
+ }
+ }
+
+ /// The probe-pool drain must actually wait even when the caller's interrupt
flag is set -- the state
+ /// shutdown leaves the warmup thread in. A busy task that ignores
interrupts stands in for a probe
+ /// mid-flight: a naive `awaitTermination` would throw immediately on the
interrupt flag and skip the
+ /// drain, leaving the pool un-terminated; the save-and-clear must let the
wait complete and restore the
+ /// flag. This is the shutdown path #17 was about.
+ @Test
+ public void awaitPoolDrainWaitsForTasksEvenWhenCallerInterrupted() throws
Exception {
+ ExecutorService pool = Executors.newSingleThreadExecutor();
+ try {
+ CountDownLatch started = new CountDownLatch(1);
+ pool.submit(() -> {
+ started.countDown();
+ long end = System.currentTimeMillis() + 300L;
+ while (System.currentTimeMillis() < end) {
+ // Busy-wait that deliberately ignores interrupts, standing in for a
probe mid-flight.
+ }
+ });
+ assertTrue(started.await(2, TimeUnit.SECONDS));
+ pool.shutdownNow(); // interrupts the
(interrupt-ignoring) task
+ Thread.currentThread().interrupt(); // as stopWarmup() leaves the
warmup thread on shutdown
+
+ SingleConnectionBrokerRequestHandler.awaitPoolDrain(pool, 5_000L);
+
+ // Flag restored (and cleared here so it does not leak to other tests);
and the drain actually waited
+ // for the task rather than returning 0ms on the interrupt flag.
+ assertTrue(Thread.interrupted(), "the interrupt flag must be restored");
+ assertTrue(pool.isTerminated(), "drain must wait for the task, not skip
on the interrupt flag");
+ } finally {
+ pool.shutdownNow();
+ }
+ }
+
+ private static TableRouteInfo route(boolean hasOffline, boolean hasRealtime)
{
+ TableRouteInfo routeInfo = mock(TableRouteInfo.class);
+ when(routeInfo.hasOffline()).thenReturn(hasOffline);
+ when(routeInfo.hasRealtime()).thenReturn(hasRealtime);
+ return routeInfo;
+ }
+
+ /// The coverage universe is tenant-scoped: the union of serving instances
over the broker's routable
+ /// tables, NOT the cluster-wide getRoutableServerInstanceMap(). A server no
routable table routes to
+ /// (another tenant's server) is excluded, so it is never picked as a cover
target nor flagged uncovered.
+ @Test
+ public void routableServersIsTenantScopedNotClusterWide() {
+ // s4 is in the cluster-wide server map but no table routes to it.
+ RoutingManager routing = routing(
+ Map.of("a_OFFLINE", Set.of("s1", "s2"), "b_OFFLINE", Set.of("s3")),
Set.of("s1", "s2", "s3", "s4"));
+
assertEquals(SingleConnectionBrokerRequestHandler.routableServers(routing),
Set.of("s1", "s2", "s3"));
+ // With every routable server covered by the two tables, nothing is
uncovered even though s4 is in the
+ // cluster-wide map.
+ assertTrue(SingleConnectionBrokerRequestHandler.uncoveredRoutableServers(
+ routing, List.of("a_OFFLINE", "b_OFFLINE")).isEmpty());
+ }
+
+ private static RoutingManager routing(Map<String, Set<String>>
tableToServers, Set<String> routableServers) {
+ RoutingManager routingManager = mock(RoutingManager.class);
+
when(routingManager.getRoutableTables()).thenReturn(tableToServers.keySet());
+ tableToServers.forEach((table, servers) ->
when(routingManager.getServingInstances(table)).thenReturn(servers));
+ Map<String, ServerInstance> serverMap = new HashMap<>();
+ for (String server : routableServers) {
+ serverMap.put(server, mock(ServerInstance.class));
+ }
+ when(routingManager.getRoutableServerInstanceMap()).thenReturn(serverMap);
+ return routingManager;
+ }
+}
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerGauge.java
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerGauge.java
index 9387ac13f8d..83289ca40c8 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerGauge.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerGauge.java
@@ -28,6 +28,17 @@ public enum BrokerGauge implements AbstractMetrics.Gauge {
MAX_BURST_QPS("tables", false),
QUERY_RATE_LIMIT_DISABLED("queryQuota", true),
NETTY_CONNECTION_CONNECT_TIME_MS("nettyConnection", true),
+ // 1 once startup warmup has completed and readiness may be granted, 0 while
still warming. Always 1
+ // when warmup is disabled, so dashboards read identically on untouched
deployments.
+ STARTUP_WARMUP_COMPLETE("status", true),
+ // Number of routable servers (the servers this broker's routable tables
route to) that startup warmup did
+ // NOT reach before it exited -- no probe hit them. Set once at warmup exit.
0 on a normal floor exit (the
+ // round-robin probed every table). Non-zero means warmup stopped short of
its intended per-server
+ // coverage: the budget expired, or servers were too slow to respond, before
every server was probed. It
+ // is a completeness signal, not "these servers are cold" -- serve-path JIT
is warmed per-JVM and channels
+ // are opened by pre-connect, so an unreached server is only marginally
colder. Not recorded when warmup
+ // is disabled.
+ STARTUP_WARMUP_UNCOVERED_SERVERS("servers", true),
REQUEST_SIZE("requestSize", false),
RESIZE_TIME_MS("milliseconds", false),
UNHEALTHY_SERVERS("servers", true),
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMetrics.java
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMetrics.java
index 4f027e1702d..df0715e930c 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMetrics.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMetrics.java
@@ -47,6 +47,13 @@ public class BrokerMetrics extends
AbstractMetrics<BrokerQueryPhase, BrokerMeter
return BROKER_METRICS_INSTANCE.get();
}
+ /// A shared no-op instance backed by a
[org.apache.pinot.spi.metrics.NoopPinotMetricsRegistry]: every
+ /// record is discarded. For paths that must supply a [BrokerMetrics] but
must not pollute the real
+ /// counters -- e.g. startup warmup priming the reduce path with synthetic
queries.
+ public static BrokerMetrics noop() {
+ return NOOP;
+ }
+
/// Constructs the broker metrics.
///
/// @param metricsRegistry The metric registry used to register timers and
meters.
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 b049e51f66c..7b64714c452 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,9 @@ 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),
+ // Wall-clock duration of startup warmup, from the moment it begins (Helix
convergence) to the moment
+ // readiness is released. Distinguishes "warmed to the floor" from "hit the
budget ceiling".
+ STARTUP_WARMUP_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),
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BrokerStartupWarmupIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BrokerStartupWarmupIntegrationTest.java
new file mode 100644
index 00000000000..3567ae46d1d
--- /dev/null
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BrokerStartupWarmupIntegrationTest.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.BrokerWarmupConfig;
+import org.apache.pinot.common.utils.ServiceStatus;
+import org.apache.pinot.spi.config.table.TableConfig;
+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;
+
+
+/// Integration test for the broker startup data-plane warmup feature
+/// (`pinot.broker.startup.warmup.*`). Brings up a real ZK + controller +
server + broker with an offline
+/// table, then verifies two things end-to-end:
+///
+/// 1. With warmup enabled the readiness gate opens: the broker's
[ServiceStatus] reaches `GOOD`.
+/// Readiness is held at `STARTING` until warmup completes, and the
broker's query path serves
+/// regardless of [ServiceStatus], so a gate that never released would not
fail `setUp` -- asserting
+/// `GOOD` is what actually proves the gate released after warmup.
+/// 2. The production warmup path (`RoutingManager` set-cover ->
[QueryRouter] probe -> reduce) reaches
+/// its depth floor and returns `true` within the budget against a live
server.
+public class BrokerStartupWarmupIntegrationTest extends
BaseClusterIntegrationTest {
+ private static final long WARMUP_BUDGET_MS = 30_000L;
+ // Small floor so the integration test warms and opens readiness quickly
(the depth floor is only large
+ // in production, to drive JIT to its top tier).
+ private static final int WARMUP_MIN_ITERATIONS = 10;
+
+ @Override
+ protected void overrideBrokerConf(PinotConfiguration brokerConf) {
+
brokerConf.setProperty(CommonConstants.Broker.CONFIG_OF_BROKER_STARTUP_WARMUP_ENABLED,
true);
+
brokerConf.setProperty(CommonConstants.Broker.CONFIG_OF_BROKER_STARTUP_WARMUP_BUDGET_MS,
WARMUP_BUDGET_MS);
+
brokerConf.setProperty(CommonConstants.Broker.CONFIG_OF_BROKER_STARTUP_WARMUP_MIN_ITERATIONS,
+ WARMUP_MIN_ITERATIONS);
+ }
+
+ @BeforeClass
+ public void setUp()
+ throws Exception {
+ TestUtils.ensureDirectoriesExistAndEmpty(_tempDir, _segmentDir, _tarDir);
+ startZk();
+ startController();
+ startBroker();
+ startServer();
+
+ Schema schema = createSchema();
+ addSchema(schema);
+ TableConfig tableConfig = createOfflineTableConfig();
+ addTableConfig(tableConfig);
+
+ // Build and upload segments so the broker has a live server to route
probe queries to.
+ 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 warmupEnabledBrokerReachesGoodServiceStatus() {
+ String instanceId = _brokerStarters.get(0).getInstanceId();
+ TestUtils.waitForCondition(aVoid ->
ServiceStatus.getServiceStatus(instanceId) == ServiceStatus.Status.GOOD,
+ WARMUP_BUDGET_MS, "Broker with warmup enabled never reported GOOD
service status");
+ // Once GOOD, the warmup gate has released: its description is no longer
the "warming up" text. (The
+ // composite ServiceStatus description concatenates every callback's
"<name>:<desc>;", so it is never
+ // literally "None"; the meaningful assertion is that the warming
description is gone.) The full
+ // STARTING -> "Warming up broker data plane" -> GOOD transition is
covered deterministically in
+ // BrokerWarmupGateTest, since a healthy cluster warms faster than this
poll can observe.
+
Assert.assertFalse(ServiceStatus.getStatusDescription(instanceId).contains("Warming
up broker data plane"),
+ "Once GOOD, the warmup gate must no longer report the warming-up
description");
+ }
+
+ @Test
+ public void warmUpReachesFloorAgainstLiveServer() {
+ BrokerWarmupConfig config = new BrokerWarmupConfig(true, WARMUP_BUDGET_MS,
WARMUP_MIN_ITERATIONS, 1);
+ boolean reachedFloor = _brokerStarters.get(0).getBrokerRequestHandler()
+ .warmUp(config, System.currentTimeMillis() + WARMUP_BUDGET_MS);
+ Assert.assertTrue(reachedFloor,
+ "Warmup should reach its probe-count floor against a live server
within the budget");
+ }
+
+ @Test
+ public void warmUpReachesFloorWithConcurrentProbes() {
+ // Concurrency 3: probes fire in parallel on a 3-thread pool, exercising
the concurrent scatter/gather
+ // path the serial arms do not.
+ BrokerWarmupConfig config = new BrokerWarmupConfig(true, WARMUP_BUDGET_MS,
WARMUP_MIN_ITERATIONS, 3);
+ boolean reachedFloor = _brokerStarters.get(0).getBrokerRequestHandler()
+ .warmUp(config, System.currentTimeMillis() + WARMUP_BUDGET_MS);
+ Assert.assertTrue(reachedFloor, "Concurrent warmup should reach its
probe-count floor within the budget");
+ }
+
+ @Test
+ public void warmUpReturnsFalseWhenBudgetExpiresBeforeFloor() {
+ // An unreachable floor with a short budget must exit on the budget
(returning false) and must not hang.
+ long shortBudgetMs = 2_000L;
+ BrokerWarmupConfig config = new BrokerWarmupConfig(true, shortBudgetMs,
100_000_000, 1);
+ long start = System.currentTimeMillis();
+ boolean reachedFloor = _brokerStarters.get(0).getBrokerRequestHandler()
+ .warmUp(config, start + shortBudgetMs);
+ long elapsed = System.currentTimeMillis() - start;
+ Assert.assertFalse(reachedFloor, "Warmup must return false when the budget
expires before the floor");
+ // Tight bound: with the budget a hard ceiling (each Future.get is bounded
by the remaining budget and
+ // stragglers are cancelled), warmUp returns within a small epsilon of the
budget, not merely "eventually".
+ Assert.assertTrue(elapsed < shortBudgetMs + 2_000L,
+ "Warmup must return within ~budget, not hang (elapsed " + elapsed + "
ms, budget " + shortBudgetMs + ")");
+ }
+}
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 02c91c1afe1..2c76cf1e39f 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
@@ -435,6 +435,32 @@ public class CommonConstants {
"pinot.broker.startup.minResourcePercent";
public static final double DEFAULT_BROKER_MIN_RESOURCE_PERCENT_FOR_START =
100.0;
+ // Startup data-plane warmup: before readiness is granted, run probe
queries so the JIT-compiled query
+ // path and per-query caches are warm before the first real traffic. Off
by default; opt-in per
+ // deployment.
+ public static final String CONFIG_OF_BROKER_STARTUP_WARMUP_ENABLED =
"pinot.broker.startup.warmup.enabled";
+ public static final boolean DEFAULT_BROKER_STARTUP_WARMUP_ENABLED = false;
+ // Hard ceiling on the whole warmup (measured from Helix convergence):
readiness opens when it expires
+ // whatever the probe progress, so a slow or unreachable server cannot
stall a rolling restart. Sized so
+ // the minIterations floor is actually reachable on a constrained/TLS
broker (~1000 serial probes take
+ // ~20-25s there) -- the budget is the safety cap, not the normal exit. A
healthy broker reaches the
+ // floor and serves well before this; only a genuinely slow one runs to
the cap.
+ public static final String CONFIG_OF_BROKER_STARTUP_WARMUP_BUDGET_MS =
"pinot.broker.startup.warmup.budgetMs";
+ public static final long DEFAULT_BROKER_STARTUP_WARMUP_BUDGET_MS = 30_000L;
+ // Minimum number of successful probe queries before warmup declares the
broker warm. This is a depth
+ // floor, not a latency guess: enough probe invocations to drive the query
path's JIT to its top tier.
+ // Warmup exits when this many probes have run OR the budget expires --
whichever comes first. The probe
+ // is always the static `SELECT * FROM "<t>" LIMIT 1` over a set-cover of
tables spanning every server.
+ public static final String CONFIG_OF_BROKER_STARTUP_WARMUP_MIN_ITERATIONS =
+ "pinot.broker.startup.warmup.minIterations";
+ public static final int DEFAULT_BROKER_STARTUP_WARMUP_MIN_ITERATIONS =
1000;
+ // Number of probe queries fired concurrently per round. Serial (1) warms
the serve path; a higher value
+ // additionally warms the concurrency step (channel-lock contention,
concurrent scatter/gather/reduce)
+ // that the first real traffic burst hits. Default 1 (serial); raise it to
warm closer to the expected
+ // burst.
+ public static final String CONFIG_OF_BROKER_STARTUP_WARMUP_CONCURRENCY =
+ "pinot.broker.startup.warmup.concurrency";
+ public static final int DEFAULT_BROKER_STARTUP_WARMUP_CONCURRENCY = 1;
// 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
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
