jineshparakh commented on code in PR #19500:
URL: https://github.com/apache/pinot/pull/19500#discussion_r3965517764
##########
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,
+ _brokerMetrics);
+ }
+ } 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();
+ }
+
+ /// Keeps only the configured tables that this broker actually routes.
Static for unit testing.
+ @VisibleForTesting
+ static List<String> filterRoutableTables(RoutingManager routingManager,
List<String> tables) {
Review Comment:
No longer applicable — `warmup.tables` (and `warmup.queries`) have been
removed. Warmup now has a single fixed behaviour: the static `SELECT * FROM
"<t>" LIMIT 1` over a set-cover of tables spanning every routable server, with
no table/query config.
##########
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,
+ _brokerMetrics);
+ }
+ } 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();
+ }
+
+ /// Keeps only the configured tables that this broker actually routes.
Static for unit testing.
+ @VisibleForTesting
+ static List<String> filterRoutableTables(RoutingManager routingManager,
List<String> tables) {
+ Set<String> routable = routingManager.getRoutableTables();
+ List<String> result = new ArrayList<>(tables.size());
+ for (String table : tables) {
+ if (routable.contains(table)) {
+ result.add(table);
+ } else {
+ LOGGER.warn("Broker warmup: configured table {} is not routable on
this broker; skipping", table);
+ }
+ }
+ return result;
+ }
+
+ /// Picks up to `maxTables` tables that between them cover as many routable
servers as possible.
+ ///
+ /// 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 while degrading
+ /// gracefully on a tenant with thousands of tables. Static so it can be
unit-tested against a mocked
+ /// [RoutingManager] without constructing a broker. When `maxTables` is too
small to span every server,
+ /// the leftover servers stay unprobed; [#reportServerCoverage] surfaces
that as a warning and a metric.
+ @VisibleForTesting
+ static List<String> selectProbeTables(RoutingManager routingManager, int
maxTables) {
+ if (maxTables <= 0) {
+ return List.of();
+ }
+ 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);
+ List<String> selected = new ArrayList<>(Math.min(maxTables,
candidates.size()));
+ for (String tableNameWithType : candidates) {
+ if (selected.size() >= maxTables || 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#BROKER_WARMUP_UNCOVERED_SERVERS] and, when
non-zero, warns with the routable
+ /// servers the chosen probe tables do not route to. For the default
set-cover this means `maxTables` was
+ /// too small to span the server set (raise it); for a custom
`warmup.tables` list it means the list omits
+ /// some servers. Warmup still proceeds: those servers' broker-to-server
channels are opened
+ /// deterministically by startup pre-connect, and the serve-path JIT warmup
drives is server-agnostic, so
+ /// this is an observability signal, not a failure.
+ /// Static (taking its collaborators as parameters) so the gauge emission
and warning are unit-testable
+ /// against a mocked [RoutingManager] and [BrokerMetrics].
+ @VisibleForTesting
+ static void reportServerCoverage(RoutingManager routingManager,
BrokerMetrics brokerMetrics,
+ boolean hasCustomTables, List<String> tables) {
+ int totalServers = routableServers(routingManager).size();
+ Set<String> uncovered = uncoveredRoutableServers(routingManager, tables);
+
brokerMetrics.setValueOfGlobalGauge(BrokerGauge.BROKER_WARMUP_UNCOVERED_SERVERS,
uncovered.size());
+ if (!uncovered.isEmpty()) {
+ LOGGER.warn("Broker warmup probes {} table(s) covering {}/{} routable
server(s); {} left unprobed: {}. {}",
+ tables.size(), totalServers - uncovered.size(), totalServers,
uncovered.size(), uncovered,
+ hasCustomTables
+ ? "Add tables routing to them to
pinot.broker.startup.warmup.tables."
+ : "Raise pinot.broker.startup.warmup.maxTables to span every
server.");
+ }
+ }
+
+ /// 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
the "raise maxTables" warning
+ /// would be 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();
+ List<T> batch = new ArrayList<>(concurrency);
+ for (int i = 0; i < concurrency; i++) {
+ batch.add(items.get(Math.floorMod(startSeq + i, size)));
Review Comment:
Fixed — `roundRobinBatch` returns an empty list for empty input instead of
hitting `floorMod(x, 0)`. Added a unit test.
##########
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() {
Review Comment:
Done — documented that the gate is HTTP-readiness-only (the `/health`
endpoint a load balancer / k8s probe polls) and explicitly does NOT change
Helix discovery: the broker is already ONLINE in the broker-resource external
view by this point, so a client resolving brokers straight from Helix may still
route to it while it warms. That's deliberate and safe.
##########
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:
Fixed. `stopWarmup`/`stopPreConnect` now interrupt and `join(1s)` the
startup thread, and `warmUp`'s `finally` does `shutdownNow()` + a bounded
`awaitTermination(1s)` on the probe pool — so an interrupted probe unwinds
before the request handler is torn down, keeping shutdown logs clean. Bounded,
so shutdown never hangs on it.
##########
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:
Answering directly from the latest experiment (CS-408 CP29: single broker,
288 q/s, 2-vCPU, TLS). The deep-warm arm converged at **~700–930 probes in
~1.5–1.8 s**, exiting well before the budget — not the 30 s cap. (That build
used a latency-target exit we've since replaced with the `minIterations` count
floor.) At the measured **~2 ms** per warm probe, ~1000 probes ≈ ~2 s, so the
shipped 1000-count floor is the exit here, **not the budget**.
So the `1000-in-15s` premise doesn't actually hold: the warm probe is ~2 ms
(not ~15 ms), so the floor was reachable even at the old 15 s. I still bumped
`budgetMs` 15s→30s as a pure safety cap for degraded conditions
(slow/unreachable servers), and demoted the budget-expiry log from WARN to INFO
whenever any probes completed, so a normal budget-exit no longer reads as a
failure. One honest caveat: the exact 1000-count exit wasn't itself benchmarked
— it's a count-based stand-in set just above the measured-sufficient depth
(~700–930).
--
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]
