gortiz commented on code in PR #19407: URL: https://github.com/apache/pinot/pull/19407#discussion_r3932692104
########## pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/ServerPreConnector.java: ########## @@ -0,0 +1,135 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.broker.requesthandler; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.util.concurrent.ThreadFactoryBuilder; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.concurrent.CompletionService; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorCompletionService; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.function.BiPredicate; +import java.util.function.Supplier; +import javax.annotation.concurrent.ThreadSafe; +import org.apache.pinot.core.transport.ServerInstance; +import org.apache.pinot.spi.config.table.TableType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/// Opens broker-to-server Netty channels ahead of query traffic, so the first real query does not pay +/// the blocking `connect()` -- and, when broker-to-server TLS is on, the handshake -- on its critical +/// path. +/// +/// `ServerRoutingInstance` identity includes the table type, so OFFLINE and REALTIME are **separate** +/// channels to the same physical server; both are connected here. Connecting an already-active channel +/// is a no-op, so this is safe to call more than once. +/// +/// Bounded on two axes so it can never stall startup: a capped thread pool, and a per-channel wait +/// clamped to the caller's deadline. 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. +/// +/// 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. +@ThreadSafe +public class ServerPreConnector { + private static final Logger LOGGER = LoggerFactory.getLogger(ServerPreConnector.class); + + /// Cap on the connect thread pool: a large tenant must not spawn a thread per server. Safe to exceed + /// the core count even on a 2- or 4-vCPU broker: each task is blocking connect + TLS handshake (mostly + /// network wait, with the actual I/O on Netty's event loop), and this runs during startup before any + /// query load, so the threads are almost entirely parked rather than contending for CPU. + @VisibleForTesting + static final int MAX_CONNECT_THREADS = 16; + + private final Supplier<Collection<ServerInstance>> _routableServersSupplier; + private final BiPredicate<ServerInstance, TableType> _connectFn; + + /// @param routableServersSupplier supplies the servers to connect, evaluated once per [#preConnect] + /// call after the caller has ensured routing is built + /// @param connectFn opens the channel for one (server, table type) and returns whether it succeeded + public ServerPreConnector(Supplier<Collection<ServerInstance>> routableServersSupplier, + BiPredicate<ServerInstance, TableType> connectFn) { + _routableServersSupplier = routableServersSupplier; + _connectFn = connectFn; + } + + /// Opens a channel to every routable server, for both table types, in parallel, bounded by + /// `deadlineMs` (an absolute [System#currentTimeMillis] value). Returns the number of channels + /// successfully connected. Never throws: a channel that fails or times out is logged and skipped. + public int preConnect(long deadlineMs) { + // Snapshot the routable-server view once. The supplier may return a live map view that another thread + // updates during startup; snapshotting keeps the channel count consistent with the tasks actually + // submitted below, so we never poll for phantom channels or under-count real ones. + List<ServerInstance> servers = new ArrayList<>(_routableServersSupplier.get()); + if (servers.isEmpty() || System.currentTimeMillis() >= deadlineMs) { + return 0; + } + long startMs = System.currentTimeMillis(); + int channelCount = servers.size() * TableType.values().length; + ExecutorService executor = Executors.newFixedThreadPool(Math.min(channelCount, MAX_CONNECT_THREADS), Review Comment: Thanks — and I'd say **don't add the cap**. The pool cap can stay exactly as it is. What I'd change is what `preConnect` *waits for*, because I think the cliff is the less important half of this. **Starvation isn't required to lose the whole budget — one straggler is enough.** The counting loop runs to `channelCount` and its only early exit is budget exhaustion, so `preConnect` returns at `min(budget, slowest channel)`. And because the per-connect timeout is derived from the remaining budget, a black-holed connect is configured to fail at *approximately the deadline*. So one unreachable server means the final `poll` blocks for the rest of the budget, and `_isStarting` stays set for all of it. I measured this against your current head — the real `ServerPreConnector`, stubbed `ChannelConnector`, 4 servers x 2 table types, one connect sleeping out its whole `timeoutMs`, 30s budget: ``` current: connected=7/8 elapsed=30004 ms with a grace window: connected=7/8 elapsed= 2027 ms ``` Same channels connected. Only the waiting differs. On a rolling restart of 20 brokers, one unreachable server anywhere in the set currently adds about ten minutes to the rollout — and with the server set being cluster-wide rather than tenant-scoped, "anywhere in the set" is a wide net. That's also why I'd avoid the sub-cap specifically: it does bound the readiness delay, which is the right target, but it can't distinguish black-holed from slow-but-reachable. You'd abort exactly the TLS connects pre-connect exists to warm, to fix a delay that has a cheaper cure. The cheaper cure is to stop waiting on stragglers rather than to abort them: ```diff + /// How long to keep waiting once channels have started coming back and then stop arriving. 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 threads. + @VisibleForTesting + static final long STRAGGLER_GRACE_MS = 2_000L; + for (int i = 0; i < channelCount; i++) { long remainingMs = deadlineMs - System.currentTimeMillis(); if (remainingMs <= 0) { break; } + // Until something has come back there is no way to tell "every server is slow" from "a few + // are stuck", so the first channel is given the whole budget. After that, a quiet grace + // window is the signal that what is left is stuck. + long waitMs = i == 0 ? remainingMs : Math.min(remainingMs, STRAGGLER_GRACE_MS); try { - Future<Boolean> future = completionService.poll(remainingMs, TimeUnit.MILLISECONDS); + Future<Boolean> future = completionService.poll(waitMs, TimeUnit.MILLISECONDS); if (future == null) { - // Budget elapsed before the next channel finished; the rest fall back to the lazy path. + LOGGER.info("No pre-connect channel completed in {} ms with {}/{} still outstanding; " + + "releasing startup and leaving them to the lazy path", waitMs, channelCount - i, + channelCount); break; } } finally { - executor.shutdownNow(); + // Graceful, not shutdownNow(). A channel we stopped waiting on is still in flight on a daemon + // thread; interrupting a worker parked in connect().sync() abandons a ChannelFuture that can + // still complete, leaving a socket nobody references or closes. Letting them run means a late + // channel is still published and the first query to that server reuses it. Each task is + // already bounded by its own deadline-derived timeout, so nothing outlives the budget. + executor.shutdown(); } ``` No `BaseBrokerStarter` change needed — `startPreConnect`'s `finally { _isStarting = false; }` runs as soon as `preConnectServers` returns, so returning early *is* releasing startup early. Three notes on the shape: - **`i == 0` gets the whole budget.** That's what keeps it safe: before any result you can't tell "all slow" from "a few stuck", so the first channel waits as long as the budget allows and the grace window only applies once there's a sign of life. Without it, a cluster whose first connect takes 5s would bail at 2s having connected nothing. - **`shutdown()` rather than `shutdownNow()`** also retires the orphaned-channel leak from my first pass — an interrupted worker abandons a `ChannelFuture` that can still complete. - **The return value's meaning shifts** slightly, from "connected within the budget" to "connected before we stopped waiting", so it under-counts a straggler that lands later. Worth a javadoc line. `BrokerServerPreConnectIntegrationTest` and `TlsIntegrationTest` still hold, since everything there connects in milliseconds and no grace window ever expires. A test, which is the harness the numbers above came from: ```java @Test public void oneStuckChannelDoesNotHoldStartupForTheWholeBudget() { List<ServerInstance> servers = mockServers(4); // 4 x 2 table types = 8 channels long budgetMs = 30_000L; AtomicInteger n = new AtomicInteger(); long startMs = System.currentTimeMillis(); int connected = new ServerPreConnector(() -> servers, (server, tableType, timeoutMs) -> { if (n.getAndIncrement() == 0) { try { Thread.sleep(timeoutMs); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } return false; } return true; }).preConnect(startMs + budgetMs); 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"); } ``` Two qualifiers I don't want to gloss over. `2_000L` is a magic number — it needs to sit above the spread between the fastest and slowest *healthy* connect, and with 16 workers and a ~120ms TLS connect, completions arrive in ~120ms waves, so it's around sixteen waves of headroom. A constant seems right; config if anyone hits it. And up to 16 daemon threads can outlive the return, parked on sockets until their own timeouts fire — bounded, and the broker is serving by then, but real. Lastly: this mitigates the readiness gate rather than settling it. If you ever decide the gate shouldn't hold startup at all, the grace window becomes unnecessary — harmless, but unnecessary. -- 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]
