gortiz commented on code in PR #19407: URL: https://github.com/apache/pinot/pull/19407#discussion_r3925839622
########## 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), + new ThreadFactoryBuilder().setNameFormat("broker-preconnect-%d").setDaemon(true).build()); + // A completion service hands channels back in the order they finish, not the order submitted, so a + // slow or unreachable server never blocks the counting of faster ones ahead of the shared deadline + // -- no head-of-line blocking, and no under-count of channels that already connected in parallel. + CompletionService<Boolean> completionService = new ExecutorCompletionService<>(executor); + int connected = 0; + try { + for (ServerInstance server : servers) { + for (TableType tableType : TableType.values()) { Review Comment: Yes — I'd switch. And having dug into the keying, I think the case is stronger than either of us put it, because the table type is irrelevant to the transport. `ServerInstance#toServerRoutingInstance` (lines 162-176) derives the port from the *routing* type only — `_port` for `NETTY`, `_grpcPort` for `GRPC`, `_nettyTlsPort` for `NETTY_TLS`. `tableType` is passed straight through and never touches hostname or port. But `ServerRoutingInstance#equals`/`hashCode` include `_tableType`, and `_serverToChannelMap` is keyed by that object. So the two "channels" for a server are two TCP connections to the identical `host:port`: ``` server0:8098 ├── (server0, 8098, OFFLINE) ──► ServerChannel A ──► TCP+TLS to server0:8098 └── (server0, 8098, REALTIME) ──► ServerChannel B ──► TCP+TLS to server0:8098 ``` The `ServerRoutingInstance` javadoc says it outright: *"Different table types on same host and port are counted as different instances."* That undercuts the table-added-later case specifically. The second channel shares nothing with the first — same endpoint, same TCP handshake, same TLS handshake, same certificate verification, no session reuse. So pre-warming the REALTIME leg isn't warming a path or amortising anything; it opens a duplicate socket at full cost against the chance that a realtime table later lands on that same server, in this same broker process, before the next restart. On an offline-only cluster it is 50% of all pre-connect work, and 50% of the server-side handshake load, spent on sockets that stay idle for the process lifetime — and nothing evicts them (`DirectOOMHandler` is the only thing that ever removes a map entry). To be clear about *why* the cross product exists, since it is not gratuitous: the channel is the response-demultiplexing key. `DataTableHandler` is constructed per channel with a fixed `_serverRoutingInstance` and passes it to `QueryRouter#receiveDataTable`, and `AsyncQueryResponse._responseMap` is keyed by `ServerRoutingInstance` — so a response's identity comes from which socket it arrived on, not from the payload. Two sockets is how a hybrid query's OFFLINE and REALTIME legs to one physical server are kept apart. Which means "just pre-connect per server" is not available: with the map keyed by table type, opening only OFFLINE leaves no REALTIME entry, so a realtime query still pays a lazy connect. The narrowing that works is the legs routing says you will actually use — iterate this broker's routed tables, take the type from the table name, and `getServingInstances()` for the servers. On the API surface: `getServingInstances(tableNameWithType)` is already on the interface, so the only thing missing is enumerating this broker's routed tables, and `_routingEntryMap` is already keyed by `tableNameWithType` inside `BaseBrokerRoutingManager`. One accessor on a class the PR already edits, and it fixes both multipliers at once, since the table name carries its own type. Worth confirming your own finding too, because it is the larger of the two and I had underweighted it: `getRoutableServerInstanceMap()` really is every enabled server in the cluster. `_routableServerInstanceMap = buildRoutableServerInstanceMap(enabledServers)` at `BaseBrokerRoutingManager:465`, where `enabledServers` is built by iterating every ZNRecord under `/INSTANCES` and filtering through `isEnabledServer` (`:508-518`), which checks only "is a server instance", "is Helix-enabled", "is not shutting down". No tenant, no table, no routing. So on a 1000-server cluster whose broker serves a 50-server tenant with offline tables only, pre-connect opens 2000 channels where 50 would do — and puts a TLS handshake on ~950 servers belonging to other tenants that this broker will never query, at every restart. Routing-derived pairs fix that in the same change. Separately, and explicitly **not** for this PR: keying `_serverToChannelMap` by the transport endpoint `(host, port)` and demultiplexing on `DataTable.MetadataKey.TABLE` (which already exists) would permanently halve broker-to-server connections on every hybrid cluster. That is a transport refactor with its own risk and its own review; I mention it only so the duplicate-socket observation does not get lost. Happy to file it as an issue if you would rather not carry it. -- 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]
