jineshparakh commented on code in PR #19500:
URL: https://github.com/apache/pinot/pull/19500#discussion_r3965511902
##########
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:
Fixed. `runConcurrentRound` now takes the deadline and bounds each
`Future.get(...)` by the *remaining* budget; once the budget is spent, every
still-outstanding probe is cancelled instead of awaited. So the ceiling is
enforced per-wait, not just checked between rounds.
--
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]
