gortiz commented on code in PR #19500:
URL: https://github.com/apache/pinot/pull/19500#discussion_r3965771513
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/SingleConnectionBrokerRequestHandler.java:
##########
@@ -117,6 +138,438 @@ public void shutDown() {
_brokerReduceService.shutDown();
}
+ /// Warms the scatter-gather path before readiness is granted: repeatedly
runs real probe queries until
+ /// they have run enough times (the `minIterations` depth floor) or the
budget expires. Broker-to-server
+ /// channels are opened lazily by the probes themselves.
+ ///
+ /// 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. Never throws, and always returns within the
configured budget.
+ @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.
+ warmUpLocal();
+ int concurrency = Math.max(1, config.concurrency());
+ ExecutorService probePool = Executors.newFixedThreadPool(concurrency,
+ new
ThreadFactoryBuilder().setNameFormat("broker-warmup-probe-%d").setDaemon(true).build());
+ try {
+ return config.hasCustomQuery() ? warmUpWithCustomQuery(config,
deadlineMs, probePool, concurrency)
+ : warmUpWithAutoSelect(config, deadlineMs, probePool, concurrency);
+ } catch (Exception e) {
+ LOGGER.warn("Broker warmup failed; proceeding without it", e);
+ return false;
+ } finally {
+ probePool.shutdownNow();
+ }
+ }
+
+ /// Stage 1: local, no-network warmup. The network probe goes through
[QueryRouter] and stops at the
+ /// gathered DataTables, so it never exercises the SQL compile path's cold
start nor the response build +
+ /// JSON serialization the first real query pays. This compiles a throwaway
query and serializes a small
+ /// synthetic [BrokerResponseNative], warming both. Never throws.
+ private void warmUpLocal() {
+ try {
+ 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.
+ private List<Long> runConcurrentRound(List<Callable<Long>> tasks,
ExecutorService pool) {
+ 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 (Future<Long> future : futures) {
+ try {
+ Long elapsedMs = future.get(WARMUP_PROBE_TIMEOUT_MS + 1000L,
TimeUnit.MILLISECONDS);
Review Comment:
Confirmed fixed, and the shape of the fix is better than what I suggested —
bounding each `get()` by the remaining budget *and* cancelling stragglers
covers both the "entered a round with 1ms left" case and the queued-task case
in one mechanism. `runConcurrentRoundReturnsByBudgetWhenTasksQueue` pins
exactly the overshoot that was unverifiable before.
One small follow-up while you are in here, not blocking: the `catch
(Exception e)` at the bottom of the loop treats a `TimeoutException` and a
genuine `ExecutionException` identically. That is correct for the budget case,
but a probe that consistently *throws* (say a future compile bug in
`compileProbe`) is invisible outside DEBUG logging inside `probe()`. Counting
the two separately, or logging the first `ExecutionException` per warmup run at
INFO, would make a broken probe diagnosable without a debug rebuild.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/SingleConnectionBrokerRequestHandler.java:
##########
@@ -117,6 +138,438 @@ public void shutDown() {
_brokerReduceService.shutDown();
}
+ /// Warms the scatter-gather path before readiness is granted: repeatedly
runs real probe queries until
+ /// they have run enough times (the `minIterations` depth floor) or the
budget expires. Broker-to-server
+ /// channels are opened lazily by the probes themselves.
+ ///
+ /// 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. Never throws, and always returns within the
configured budget.
+ @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.
+ warmUpLocal();
+ int concurrency = Math.max(1, config.concurrency());
+ ExecutorService probePool = Executors.newFixedThreadPool(concurrency,
+ new
ThreadFactoryBuilder().setNameFormat("broker-warmup-probe-%d").setDaemon(true).build());
+ try {
+ return config.hasCustomQuery() ? warmUpWithCustomQuery(config,
deadlineMs, probePool, concurrency)
+ : warmUpWithAutoSelect(config, deadlineMs, probePool, concurrency);
+ } catch (Exception e) {
+ LOGGER.warn("Broker warmup failed; proceeding without it", e);
+ return false;
+ } finally {
+ probePool.shutdownNow();
+ }
+ }
+
+ /// Stage 1: local, no-network warmup. The network probe goes through
[QueryRouter] and stops at the
+ /// gathered DataTables, so it never exercises the SQL compile path's cold
start nor the response build +
+ /// JSON serialization the first real query pays. This compiles a throwaway
query and serializes a small
+ /// synthetic [BrokerResponseNative], warming both. Never throws.
+ private void warmUpLocal() {
+ try {
+ 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.
+ private List<Long> runConcurrentRound(List<Callable<Long>> tasks,
ExecutorService pool) {
+ 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 (Future<Long> future : futures) {
+ try {
+ Long elapsedMs = future.get(WARMUP_PROBE_TIMEOUT_MS + 1000L,
TimeUnit.MILLISECONDS);
+ if (elapsedMs != null && elapsedMs >= 0) {
+ latencies.add(elapsedMs);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return latencies;
+ } catch (Exception e) {
+ // Individual probe failed/timed out; it just does not count toward
the round. Cancel it so a probe
+ // that outlived the round's get()-timeout does not keep running
behind the next round (no-op if it
+ // already completed).
+ future.cancel(true);
+ }
+ }
+ return latencies;
+ }
+
+ /// 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;
+ }
+ }
+
+ /// Default probe path: probe `SELECT * FROM "<t>" LIMIT 1` over the
configured tables or 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 warmUpWithAutoSelect(BrokerWarmupConfig config, long
deadlineMs, ExecutorService pool,
+ int concurrency) {
+ 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();
+ boolean coverageReported = false;
+ // 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;
+ while (System.currentTimeMillis() < deadlineMs &&
!Thread.currentThread().isInterrupted()) {
+ if (tables.isEmpty()) {
+ tables = config.hasCustomTables() ?
filterRoutableTables(_routingManager, config.tables())
+ : selectProbeTables(_routingManager, config.maxTables());
+ }
+ 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 (!coverageReported) {
+ // Report once, now that the probe tables are settled, whether they
route to every server.
+ coverageReported = true;
+ reportServerCoverage(_routingManager, _brokerMetrics,
config.hasCustomTables(), tables);
+ }
+ long remainingMs = deadlineMs - System.currentTimeMillis();
+ if (remainingMs <= 0) {
+ break;
+ }
+ rounds++;
+ long probeTimeoutMs = Math.min(remainingMs, WARMUP_PROBE_TIMEOUT_MS);
+ // 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.
+ List<Callable<Long>> tasks = new ArrayList<>(concurrency);
+ for (String tableNameWithType : roundRobinBatch(tables, probeSeq,
concurrency)) {
+ tasks.add(() -> probe(compileProbe(tableNameWithType),
probeTimeoutMs));
+ }
+ probeSeq += concurrency;
+ List<Long> latencies = runConcurrentRound(tasks, pool);
+ if (latencies.isEmpty()) {
+ if (!warmupBackoff()) {
+ return false;
+ }
+ continue;
+ }
+ successfulProbes += latencies.size();
+ if (successfulProbes >= config.minIterations()) {
+ LOGGER.info("Broker warmup completed after {} round(s) at concurrency
{}; {} probes (floor {})", rounds,
+ concurrency, successfulProbes, config.minIterations());
+ return true;
+ }
+ }
+ LOGGER.warn("Broker warmup budget expired after {} round(s) at concurrency
{}; {} probes (floor {}). Proceeding "
+ + "to serve traffic.", rounds, concurrency, successfulProbes,
config.minIterations());
+ return false;
+ }
+
+ /// Custom-query probe path: run the operator-supplied queries repeatedly,
all of them every round, and
+ /// exit on the same `minIterations` probe-count floor as the default probe,
or the budget. A probe-count
+ /// floor is a cost-independent warm signal; latency plateauing is not (a
trivial query reaches a stable
+ /// latency while still cold), so it is deliberately not used here. An
expensive query that cannot reach
+ /// the floor within the budget exits on the budget, having warmed as much
as the budget allowed.
+ private boolean warmUpWithCustomQuery(BrokerWarmupConfig config, long
deadlineMs, ExecutorService pool,
+ int concurrency) {
+ // Validate-compile each configured query once so a bad one fails fast
(and never busy-spins), keeping
+ // only those that compile. Each probe below compiles its own fresh
request so nothing mutable is shared
+ // across the concurrent tasks.
+ List<String> queries = new ArrayList<>(config.queries().size());
+ for (String query : config.queries()) {
+ try {
+ CalciteSqlCompiler.compileToBrokerRequest(query);
+ queries.add(query);
+ } catch (Exception e) {
+ LOGGER.warn("Broker warmup query failed to compile; skipping it: {}",
query, e);
+ }
+ }
+ if (queries.isEmpty()) {
+ LOGGER.warn("Broker warmup: no configured query compiled; skipping
warmup");
+ return false;
+ }
+ long successfulProbes = 0;
+ int rounds = 0;
+ while (System.currentTimeMillis() < deadlineMs &&
!Thread.currentThread().isInterrupted()) {
+ long remainingMs = deadlineMs - System.currentTimeMillis();
+ if (remainingMs <= 0) {
+ break;
+ }
+ rounds++;
+ long probeTimeoutMs = Math.min(remainingMs, WARMUP_PROBE_TIMEOUT_MS);
+ // Run every configured query each round (each `concurrency` times, to
warm the concurrency step too).
+ List<Callable<Long>> tasks = new ArrayList<>(queries.size() *
concurrency);
+ for (String query : queries) {
+ for (int i = 0; i < concurrency; i++) {
+ tasks.add(() ->
probe(CalciteSqlCompiler.compileToBrokerRequest(query), probeTimeoutMs));
+ }
+ }
+ List<Long> latencies = runConcurrentRound(tasks, pool);
+ if (latencies.isEmpty()) {
+ if (!warmupBackoff()) {
+ return false;
+ }
+ continue;
+ }
+ successfulProbes += latencies.size();
+ // Same depth floor as the default probe: a probe-count floor is a
cost-independent warm signal (it
+ // drives the query path's JIT to its top tier), whereas latency
plateauing is not -- a trivial query
+ // reaches a stable latency while still cold. An expensive query that
cannot reach the floor within the
+ // budget simply exits on the budget, having warmed as much as the
budget allowed.
+ if (successfulProbes >= config.minIterations()) {
+ LOGGER.info("Broker warmup (custom query) completed after {} round(s)
at concurrency {}; {} probes "
+ + "(floor {})", rounds, concurrency, successfulProbes,
config.minIterations());
+ return true;
+ }
+ }
+ LOGGER.warn("Broker warmup (custom query) budget expired after {} round(s)
at concurrency {}; {} probes "
+ + "(floor {}). Proceeding to serve traffic.", rounds, concurrency,
successfulProbes, config.minIterations());
+ return false;
+ }
+
+ /// Compiles the default probe query for a single physical table.
+ private BrokerRequest compileProbe(String tableNameWithType) {
+ return CalciteSqlCompiler.compileToBrokerRequest("SELECT * FROM \"" +
tableNameWithType + "\" LIMIT 1");
+ }
+
+ /// 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.
+ private long probe(BrokerRequest brokerRequest, long timeoutMs) {
+ 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;
+ }
+ // A type-suffixed table (the default probe) targets exactly one type; a
raw name (possible with a
+ // custom query) targets whichever types actually exist. 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) rather than reusing the
raw-named request, which
+ // would route to no server and yield an empty request map.
+ 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;
+ }
+ ImplicitHybridTableRouteInfo hybridRouteInfo =
(ImplicitHybridTableRouteInfo) routeInfo;
+ hybridRouteInfo.setOfflineBrokerRequest(offlineBrokerRequest);
+ hybridRouteInfo.setRealtimeBrokerRequest(realtimeBrokerRequest);
+ _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.
+ AsyncQueryResponse response = _queryRouter.submitQuery(requestId,
rawTableName, routeInfo, timeoutMs);
+ Map<ServerRoutingInstance, ServerResponse> finalResponses =
response.getFinalResponses();
+ // 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()) {
+ _brokerReduceService.reduceOnDataTable(brokerRequest, brokerRequest,
dataTableMap, timeoutMs,
Review Comment:
Confirmed fixed — and putting `noop()` on `BrokerMetrics` rather than
hand-rolling a throwaway registry at the call site is the better factoring;
other "prime the path without polluting counters" callers can reuse it.
One scoping note so the claim in the description is read accurately: this
covers the reduce leg. `submitQuery` still records into
`ServerRoutingStatsManager`, which you have now documented at the call site
(and I agree with that analysis — see my reply on that thread). So "pollutes no
customer-facing surface" is true for metrics, auth, quota and the query log,
but routing stats are deliberately seeded. Worth one clause in the PR
description so nobody reads it as absolute.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/SingleConnectionBrokerRequestHandler.java:
##########
@@ -117,6 +138,438 @@ public void shutDown() {
_brokerReduceService.shutDown();
}
+ /// Warms the scatter-gather path before readiness is granted: repeatedly
runs real probe queries until
+ /// they have run enough times (the `minIterations` depth floor) or the
budget expires. Broker-to-server
+ /// channels are opened lazily by the probes themselves.
+ ///
+ /// 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. Never throws, and always returns within the
configured budget.
+ @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.
+ warmUpLocal();
+ int concurrency = Math.max(1, config.concurrency());
+ ExecutorService probePool = Executors.newFixedThreadPool(concurrency,
+ new
ThreadFactoryBuilder().setNameFormat("broker-warmup-probe-%d").setDaemon(true).build());
+ try {
+ return config.hasCustomQuery() ? warmUpWithCustomQuery(config,
deadlineMs, probePool, concurrency)
+ : warmUpWithAutoSelect(config, deadlineMs, probePool, concurrency);
+ } catch (Exception e) {
+ LOGGER.warn("Broker warmup failed; proceeding without it", e);
+ return false;
+ } finally {
+ probePool.shutdownNow();
+ }
+ }
+
+ /// Stage 1: local, no-network warmup. The network probe goes through
[QueryRouter] and stops at the
+ /// gathered DataTables, so it never exercises the SQL compile path's cold
start nor the response build +
+ /// JSON serialization the first real query pays. This compiles a throwaway
query and serializes a small
+ /// synthetic [BrokerResponseNative], warming both. Never throws.
+ private void warmUpLocal() {
Review Comment:
Confirmed fixed — looping it is what makes the depth argument
self-consistent, and deadline-guarding the loop keeps it from competing with
the network probe for budget.
Tiny doc nit: the javadoc says the local iteration count is "capped well
below the network-probe depth floor", but with `LOCAL_WARMUP_MAX_ITERATIONS =
2000` and `minIterations` defaulting to 1000, `Math.min(...)` yields 1000 — the
same number as the floor, not below it. The cap only bites if an operator
raises `minIterations` past 2000. Behaviour is right; the sentence just
oversells the relationship.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java:
##########
@@ -894,10 +911,76 @@ private void registerServiceStatusHandler() {
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. It
gates the existing readiness
+ // endpoint (getBrokerHealth -> ServiceStatus): a warming broker stays
out of the Service until it
+ // reports GOOD.
+ callbacks.add(new ServiceStatus.ServiceStatusCallback() {
+ @Override
+ public ServiceStatus.Status getServiceStatus() {
+ return _isWarm ? ServiceStatus.Status.GOOD :
ServiceStatus.Status.STARTING;
+ }
+
+ @Override
+ public String getStatusDescription() {
+ return _isWarm ? ServiceStatus.STATUS_DESCRIPTION_NONE : "Warming up
broker data plane";
+ }
+ });
+ }
ServiceStatus.setServiceStatusCallback(_instanceId,
- new ServiceStatus.MultipleCallbackServiceStatusCallback(List.of(
- new ServiceStatus.LifecycleServiceStatusCallback(this::isStarting,
this::isShuttingDown),
- _helixConvergenceCallback)));
+ new ServiceStatus.MultipleCallbackServiceStatusCallback(callbacks));
+ }
+
+ /// 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.BROKER_WARM, () ->
_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);
+ boolean reachedFloor = _brokerRequestHandler.warmUp(_warmupConfig,
warmStartMs + _warmupConfig.budgetMs());
+ 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;
Review Comment:
The `interruptAndJoin` half is right. The `awaitTermination` half does not
do what it looks like it does, though — I have opened a thread on it at
`SingleConnectionBrokerRequestHandler#warmUp`.
Short version: shutdown gets here via `stopWarmup()` ->
`warmupThread.interrupt()`, so the thread's interrupt status is already set
when the `finally` runs. `awaitTermination` is interruptible and throws
`InterruptedException` on entry when the flag is set, so it returns after 0ms.
Verified on this JDK. The result is that the wait works on the budget/floor
exit and is skipped on the shutdown exit — the only path where a probe racing a
closing `QueryRouter` actually matters.
Fix and suggested test are on the other thread.
##########
pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java:
##########
@@ -435,6 +435,42 @@ public static class Broker {
"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.
+ 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 = 15_000L;
+ // Minimum number of successful probe queries before the default probe
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.
+ 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;
Review Comment:
That answers it properly — thank you for going back to the measurements
rather than just moving a number. My premise was wrong: I assumed ~15ms per
probe from the 1000-in-15s arithmetic, and at ~2ms per warm probe the floor is
reachable with a lot of headroom, so the budget is genuinely the safety cap and
not the normal exit. Bumping to 30s plus demoting the budget-expiry log to INFO
when anything warmed resolves the "every restart logs a warning" concern
completely.
I appreciate the caveat that the exact 1000 was not itself benchmarked and
is a stand-in just above the measured-sufficient ~700-930. That is the right
way to state it, and it is a reasonable margin. No further action from me here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
