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 26f47121863 Broker startup pre-connect: remove the straggler-grace
window (#19519)
26f47121863 is described below
commit 26f471218633374f1eea3f6b5c6d81461b9b832b
Author: Jinesh Parakh <[email protected]>
AuthorDate: Thu Sep 10 21:41:12 2026 +0530
Broker startup pre-connect: remove the straggler-grace window (#19519)
---
.../broker/requesthandler/ServerPreConnector.java | 72 ++++++--------
.../requesthandler/ServerPreConnectorTest.java | 106 +++++++++++++++------
2 files changed, 107 insertions(+), 71 deletions(-)
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/ServerPreConnector.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/ServerPreConnector.java
index c90b65e9f0b..3e01cdaa2fa 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/ServerPreConnector.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/ServerPreConnector.java
@@ -52,11 +52,13 @@ import org.slf4j.LoggerFactory;
/// no-op, so this is safe to call more than once.
///
/// Bounded so it can never stall startup: a capped thread pool, a per-channel
connect bound derived from
-/// the remaining budget, a per-channel wait clamped to the caller's deadline,
and a straggler grace window
-/// ([#STRAGGLER_GRACE_MS]) that, once at least one channel is up, releases
the caller when the rest stop
-/// arriving rather than waiting out the whole budget on one stuck server. A
server that is unreachable or
-/// itself restarting is logged and skipped -- the existing lazy-connect path
still serves it. This class
-/// is stateless and thread-safe.
+/// the remaining budget, and the caller's `deadlineMs` as the single release
bound. There is no early
+/// release: [#preConnect] waits for each channel to resolve (connect or fail)
up to the deadline, then
+/// returns. That keeps readiness honest -- the broker is released once its
channels are actually warm, or
+/// the budget is spent -- rather than guessing from a quiet period that the
rest are stuck, which cannot
+/// tell a slow-but-healthy connect from a dead one and so can release with
healthy channels still cold. A
+/// failed or unreachable connect is logged and skipped and never stops the
others from being waited for;
+/// the deadline is the only thing that ends the wait early. This class is
stateless and thread-safe.
///
/// It takes its dependencies as functions rather than concrete
`RoutingManager`/`QueryRouter` types so
/// the parallelism, budget and failure handling can be unit-tested without a
live broker.
@@ -70,19 +72,11 @@ public class ServerPreConnector {
/// query load, so the threads are almost entirely parked rather than
contending for CPU.
///
/// It is a throughput cap, not a safety bound: with more channels than
threads the surplus queues
- /// behind the workers, so the budget alone must not be what stops a stuck
connect. That is why
+ /// behind the workers, so the deadline alone must not be what stops a stuck
connect. That is why
/// [ChannelConnector] takes a per-channel timeout.
@VisibleForTesting
static final int MAX_CONNECT_THREADS = 16;
- /// How long to keep waiting, once at least one channel is up, for the next
one before concluding the
- /// rest are stuck. A quiet window this long means what is left is stuck
rather than merely slow, so the
- /// caller is released and the stragglers finish -- or time out -- on their
own daemon threads. Until the
- /// first *successful* connect the whole budget is available: with nothing
up yet there is no way to tell
- /// "every server is slow" from "a few are stuck", and a fast failure must
not start the clock.
- @VisibleForTesting
- static final long STRAGGLER_GRACE_MS = 2_000L;
-
/// Opens one broker-to-server channel. Implementations must bound their own
wait by `timeoutMs` and
/// must not throw; the return value reports whether the channel is
connected.
@FunctionalInterface
@@ -110,10 +104,10 @@ public class ServerPreConnector {
}
/// Opens the supplied (server, table type) channels in parallel, bounded by
`deadlineMs` (an absolute
- /// [System#currentTimeMillis] value). Returns the number of channels
connected **before the caller was
- /// released** -- so a straggler that connects after the grace window (see
[#STRAGGLER_GRACE_MS]) is not
- /// counted, even though its channel is still published for the first query
to reuse. Never throws: a
- /// channel that fails or times out is logged and skipped.
+ /// [System#currentTimeMillis] value). Returns the number of channels
connected by the deadline. A channel
+ /// still connecting when the deadline passes keeps warming on its daemon
thread and is published for the
+ /// first query to reuse; it is simply not counted. Never throws: a channel
that fails or times out is
+ /// logged and skipped, and never stops the others from being waited for.
public int preConnect(long deadlineMs) {
// Snapshot the target view once. The supplier may derive from a live
routing view that another thread
// updates during startup; snapshotting keeps the channel count consistent
with the tasks actually
@@ -132,7 +126,6 @@ public class ServerPreConnector {
// still queues for a worker, which is what the per-channel timeout bounds.
CompletionService<Boolean> completionService = new
ExecutorCompletionService<>(executor);
int connected = 0;
- boolean releasedEarly = false;
try {
for (ChannelTarget target : targets) {
completionService.submit(() ->
_connector.connect(target.serverInstance(), target.tableType(),
@@ -143,24 +136,16 @@ public class ServerPreConnector {
if (remainingMs <= 0) {
break;
}
- // Keep the whole budget available until the first *successful*
connect: a quiet window only means
- // "the rest are stuck" once at least one channel has actually come
up. Keying the exemption on the
- // first completion instead would let a single fast event -- an
instantly refused connect, or one
- // nearby server -- start the grace clock before the
healthy-but-slower channels return, abandoning
- // them. Once one channel is up, a quiet grace window is the signal
that what is left is stuck rather
- // than slow, and the caller is released -- one unreachable server
otherwise holds the gate for the
- // entire budget. (A cluster where no server ever connects still exits
promptly when connects fail
- // fast, and waits the budget only when every connect black-holes,
which is the correct thing to do
- // for a broker that can reach nothing.)
- long waitMs = connected == 0 ? remainingMs : Math.min(remainingMs,
STRAGGLER_GRACE_MS);
try {
- Future<Boolean> future = completionService.poll(waitMs,
TimeUnit.MILLISECONDS);
+ // Wait up to the whole remaining budget for the next channel. A
completed channel returns
+ // immediately, so a fully healthy cluster is released as soon as
its last channel is up -- not at
+ // the deadline. The only thing this actually blocks on is a channel
that never completes (an
+ // unreachable server that black-holes to its own connect timeout);
the deadline is the single
+ // bound on that, and one such server costs at most the budget,
never the others' warm-up.
+ Future<Boolean> future = completionService.poll(remainingMs,
TimeUnit.MILLISECONDS);
if (future == null) {
- // The grace cap was binding (we could have waited longer but
chose not to) only when waitMs was
- // clamped below the remaining budget; otherwise this is plain
budget exhaustion.
- releasedEarly = waitMs < remainingMs;
- LOGGER.info("No pre-connect channel completed in {} ms with {}/{}
still outstanding; releasing startup "
- + "and leaving them to the lazy connect path", waitMs,
channelCount - i, channelCount);
+ LOGGER.info("Pre-connect budget elapsed with {}/{} channel(s)
still outstanding; releasing startup "
+ + "and leaving them to the lazy connect path", channelCount -
i, channelCount);
break;
}
if (Boolean.TRUE.equals(future.get())) {
@@ -171,23 +156,22 @@ public class ServerPreConnector {
Thread.currentThread().interrupt();
break;
} catch (ExecutionException e) {
- // A server that is unreachable or itself restarting must not block
startup.
+ // A server that is unreachable or itself restarting must not stop
us waiting for the others.
LOGGER.debug("Pre-connect did not complete for one channel", e);
}
}
} finally {
- // shutdown(), not shutdownNow(): a channel we stopped waiting on is
still connecting on a daemon
- // thread. Interrupting a worker parked in connect().sync() abandons a
ChannelFuture that can still
- // complete, leaking a socket nobody references or closes. Letting the
workers run means a late
- // channel is still published for the first query to reuse, and each
task is already bounded by its
- // own deadline-derived timeout, so none outlives the budget.
+ // shutdown(), not shutdownNow(): a channel still connecting past the
deadline is on a daemon thread.
+ // Interrupting a worker parked in connect().sync() abandons a
ChannelFuture that can still complete,
+ // leaking a socket nobody references or closes. Letting the workers run
means a late channel still
+ // warms in the background and is published for the first query to
reuse, and each task is already
+ // bounded by its own deadline-derived timeout, so none outlives the
budget.
executor.shutdown();
}
long elapsedMs = System.currentTimeMillis() - startMs;
if (connected < channelCount) {
- LOGGER.warn("Broker pre-connected {}/{} channel(s) in {} ms ({}); the
rest fall back to the lazy connect "
- + "path", connected, channelCount, elapsedMs, releasedEarly ?
"released early on a straggler grace window"
- : "budget elapsed");
+ LOGGER.warn("Broker pre-connected {}/{} channel(s) in {} ms; the rest
fall back to the lazy connect path",
+ connected, channelCount, elapsedMs);
} else {
LOGGER.info("Broker pre-connected {}/{} channel(s) in {} ms", connected,
channelCount, elapsedMs);
}
diff --git
a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/ServerPreConnectorTest.java
b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/ServerPreConnectorTest.java
index 00db6cad8fc..8c2bd3dbe47 100644
---
a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/ServerPreConnectorTest.java
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/ServerPreConnectorTest.java
@@ -169,14 +169,14 @@ public class ServerPreConnectorTest {
assertTrue(elapsedMs < 3_000L, "preConnect took " + elapsedMs + " ms,
expected it to honor the budget");
}
- /// One unreachable server must not hold startup for the whole budget: once
the healthy channels are back
- /// and nothing more arrives for the grace window, preConnect returns and
leaves the straggler to finish
- /// (or time out) on its own daemon thread. Without the grace window the
final poll would block for the
- /// rest of the budget.
+ /// An unreachable server that black-holes (never completes) holds
pre-connect only until the deadline,
+ /// and the healthy channels are still counted. There is no early release:
the deadline is the single
+ /// bound, so a broker with one dead server pays at most the budget on
startup, and that cost is tuned
+ /// through the budget rather than a heuristic.
@Test
- public void oneStuckChannelDoesNotHoldStartupForTheWholeBudget() {
+ public void
blackHoledChannelIsBoundedByTheDeadlineAndHealthyChannelsAreCounted() {
List<ServerInstance> servers = mockServers(4); // 4 x 2 table types = 8
channels
- long budgetMs = 30_000L;
+ long budgetMs = 800L;
AtomicInteger n = new AtomicInteger();
long startMs = System.currentTimeMillis();
@@ -184,7 +184,7 @@ public class ServerPreConnectorTest {
(server, tableType, timeoutMs) -> {
if (n.getAndIncrement() == 0) {
try {
- Thread.sleep(timeoutMs);
+ Thread.sleep(5_000L); // black-hole well past the budget;
never connects in time
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
@@ -195,18 +195,18 @@ public class ServerPreConnectorTest {
long elapsedMs = System.currentTimeMillis() - startMs;
assertEquals(connected, 7, "the seven healthy channels must still be
counted");
- assertTrue(elapsedMs < 5 * ServerPreConnector.STRAGGLER_GRACE_MS,
- "one stuck channel held startup for " + elapsedMs + " ms of a " +
budgetMs + " ms budget");
+ // Bounded by the deadline: the one black-holed channel holds only until
the budget, never longer.
+ assertTrue(elapsedMs < 3 * budgetMs,
+ "one black-holed channel held startup for " + elapsedMs + " ms of an "
+ budgetMs + " ms budget");
}
- /// The whole budget is available until the first successful connect: a
cluster whose channels are all
- /// slower than the grace window (but faster than the budget) must still
connect every one, not bail at
- /// the grace window having connected nothing. If the exemption were
missing, the first poll would time
- /// out at the grace window before any channel returned, and connected would
be 0.
+ /// Slow-but-healthy channels are all counted: with the deadline as the only
bound, a cluster whose every
+ /// channel is slow (but faster than the budget) still connects every one
and is released once they are
+ /// all up -- not at the deadline, and never with any of them abandoned.
@Test
- public void slowFirstChannelIsStillCountedAndNotAbandonedByGraceWindow() {
+ public void slowHealthyChannelsAreAllCounted() {
List<ServerInstance> servers = mockServers(3); // 3 x 2 = 6 channels
- long slowMs = ServerPreConnector.STRAGGLER_GRACE_MS + 500L; // slower
than grace, faster than budget
+ long slowMs = 2_500L; // slow, but faster
than the budget
long budgetMs = 30_000L;
long startMs = System.currentTimeMillis();
@@ -222,27 +222,26 @@ public class ServerPreConnectorTest {
}).preConnect(startMs + budgetMs);
long elapsedMs = System.currentTimeMillis() - startMs;
- assertEquals(connected, 6, "every channel must be counted even though all
are slower than the grace window");
- assertTrue(elapsedMs >= slowMs, "the first channel must be waited for past
the grace window, not abandoned");
- assertTrue(elapsedMs < budgetMs, "must not wait the whole budget once the
channels are back");
+ assertEquals(connected, 6, "every channel must be counted even though all
are slow");
+ assertTrue(elapsedMs >= slowMs, "the slow channels must be waited for, not
abandoned");
+ assertTrue(elapsedMs < budgetMs, "must be released once the channels are
back, not held to the deadline");
}
- /// A fast failure (an instantly refused connect) that completes before the
healthy channels must NOT
- /// consume the whole-budget exemption and cause the grace window to abandon
the healthy-but-slower
- /// channels. Since the exemption keys on the first *successful* connect,
the fast failure does not start
- /// the grace clock, and the five healthy channels are all waited for and
counted. Regression test for a
- /// grace-window bug where keying on the first *completion* undercounted to
0 here.
+ /// The core invariant: a failed connect must not stop us waiting for the
others. One connect fails
+ /// instantly and completes first; the five healthy-but-slower channels must
still all be waited for and
+ /// counted. (Under the old grace window, keying the release on the first
*completion* undercounted this
+ /// to 0; waiting to the deadline makes it unconditional.)
@Test
- public void healthyButSlowChannelsNotAbandonedAfterFastFailure() {
+ public void aFastFailureDoesNotStopWaitingForTheOtherChannels() {
List<ServerInstance> servers = mockServers(3); // 3 x 2 = 6 channels
- long slowMs = ServerPreConnector.STRAGGLER_GRACE_MS + 500L; // slower
than grace, faster than budget
+ long slowMs = 2_500L; // slow, but faster
than the budget
long budgetMs = 30_000L;
AtomicInteger n = new AtomicInteger();
int connected = new ServerPreConnector(() -> targets(servers,
TableType.OFFLINE, TableType.REALTIME),
(server, tableType, timeoutMs) -> {
if (n.getAndIncrement() == 0) {
- return false; // one instant failure, completes first, must not
start the grace clock
+ return false; // one instant failure, completes first, must not
end the wait
}
try {
Thread.sleep(slowMs);
@@ -254,7 +253,7 @@ public class ServerPreConnectorTest {
}).preConnect(System.currentTimeMillis() + budgetMs);
assertEquals(connected, 5,
- "the five healthy channels must be counted; a fast failure must not
trigger the grace window");
+ "the five healthy channels must be counted; a fast failure must not
stop us waiting for them");
}
/// More channels than worker threads: the surplus queues behind the pool
and still all connect. Exercises
@@ -276,6 +275,59 @@ public class ServerPreConnectorTest {
assertEquals(calls.get(), count * 2, "every channel must be attempted");
}
+ /// More channels than worker threads, every one HEALTHY but slow. The
surplus completes in waves one
+ /// connect-latency apart; because pre-connect waits to the deadline rather
than releasing on a quiet
+ /// window, every wave is waited for and all connect. (The grace window this
replaces under-counted this
+ /// to ~one wave when the inter-wave gap exceeded the window.)
+ @Test
+ public void manyHealthyChannelsInWavesAllConnect() {
+ int count = ServerPreConnector.MAX_CONNECT_THREADS * 3; // 48 channels,
pool caps at 16 -> 3 waves
+ List<ServerInstance> servers = mockServers(count);
+ long slowMs = 3_000L;
+
+ int connected = new ServerPreConnector(() -> targets(servers,
TableType.OFFLINE),
+ (server, tableType, timeoutMs) -> {
+ try {
+ Thread.sleep(slowMs);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ return true;
+ }).preConnect(System.currentTimeMillis() + 60_000L);
+
+ assertEquals(connected, count, "all healthy channels must connect even
when they complete in waves");
+ }
+
+ /// Mixed connect latencies -- one fast server, the rest slower -- now
connect **all** channels. The old
+ /// grace window sized itself off the first (fastest) connect and released
before the slower healthy
+ /// channels returned, counting only 1; waiting to the deadline waits for
every one. Regression test for
+ /// that mixed-latency under-count.
+ @Test
+ public void mixedLatencyAllChannelsConnect() {
+ List<ServerInstance> servers = mockServers(8); // 8 channels, all start
at once on the 16-worker pool
+ long slowMs = 3_000L;
+ long budgetMs = 30_000L;
+ AtomicInteger n = new AtomicInteger();
+
+ int connected = new ServerPreConnector(() -> targets(servers,
TableType.OFFLINE),
+ (server, tableType, timeoutMs) -> {
+ if (n.getAndIncrement() == 0) {
+ return true; // one fast connect; must not curtail waiting for
the slower healthy ones
+ }
+ try {
+ Thread.sleep(slowMs);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ return true;
+ }).preConnect(System.currentTimeMillis() + budgetMs);
+
+ assertEquals(connected, 8,
+ "with the deadline as the only bound, a fast connect no longer
abandons the slower healthy channels");
+ }
+
/// The thread pool is a throughput cap, not a safety bound, so each connect
has to carry its own
/// deadline-derived timeout. Without it a channel queued behind a stuck
worker could outlive the
/// budget entirely.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
