gortiz commented on code in PR #19407: URL: https://github.com/apache/pinot/pull/19407#discussion_r3949957158
########## 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: Your `connected == 0` variant is better than what I drafted, and the regression test names exactly why — an instantly-refused connect consuming iteration zero would have started the grace clock and abandoned the healthy-but-slower channels. Good catch. One follow-up, **non-blocking** — I'm approving regardless. Your tests cover many channels and slow channels separately, but never together, and that combination is where the grace window misreads the signal. I compiled `ServerPreConnector` from this head and ran both at once, with **every server healthy and nothing stuck**: | targets | per-connect | connected | elapsed | |---|---|---|---| | 16 | 3000 ms | 16/16 | 3024 ms | | **48** | **3000 ms** | **16/48** | **5011 ms** | | 48 | 100 ms | 48/48 | 316 ms | With more channels than workers the surplus completes in waves one connect-latency apart, so the quiet window the heuristic reads as *stuck* is really just the gap between waves. Two thirds of a healthy cluster drops out of the warm-up, and the WARN attributes it to a straggler grace window when nothing was straggling. The effect is benign — `shutdown()` lets the queued tasks finish and publish, so those channels still warm up in the background, and the broker is serving by then. What's lost is that the feature quietly under-delivers on any cluster whose connects are slower than the window, and the log misexplains it. Given `MAX_CONNECT_THREADS = 16`, a tenant of more than 16 (server, table type) pairs is ordinary, so this isn't only a pathological shape. Scaling the window off the first observed connect latency covers it. All tasks start together, so the first success approximates one connect's latency: ```diff int connected = 0; boolean releasedEarly = false; + long graceMs = STRAGGLER_GRACE_MS; try { ... - long waitMs = connected == 0 ? remainingMs : Math.min(remainingMs, STRAGGLER_GRACE_MS); + long waitMs = connected == 0 ? remainingMs : Math.min(remainingMs, graceMs); try { Future<Boolean> future = completionService.poll(waitMs, TimeUnit.MILLISECONDS); ... if (Boolean.TRUE.equals(future.get())) { + if (connected == 0) { + // All tasks started together, so the first success approximates one connect's latency. With + // more channels than workers the surplus completes in waves one latency apart, so a window + // narrower than that latency would abandon healthy channels still queued behind the pool. + graceMs = Math.max(STRAGGLER_GRACE_MS, 2 * (System.currentTimeMillis() - startMs)); + } connected++; } ``` Verified against both shapes, so the straggler behaviour this thread was about is untouched: | scenario | as pushed | with the scaled window | |---|---|---| | 48 healthy @ 3000 ms | 16/48 in 5011 ms | **48/48 in 9013 ms** | | 48 healthy @ 100 ms | 48/48 in 316 ms | 48/48 in 315 ms | | 8 channels, one black-holed | 7/8 in 2022 ms | 7/8 in **2022 ms** | | 48 channels, one black-holed | 47/48 in 2008 ms | 47/48 in **2006 ms** | Entirely your call whether to take it now or leave it — the current behaviour degrades safely, so it isn't a merge blocker. If you do take it, a test with `MAX_CONNECT_THREADS * 3` targets each slower than `STRAGGLER_GRACE_MS` would pin it; that's the shape that produced the 16/48 above. -- 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]
