You are viewing a plain text version of this content. The canonical link for it is here.

Posted to commits@pinot.apache.org by "jineshparakh (via GitHub)" <gi...@apache.org> on 2026/08/31 13:21:50 UTC

[PR] Add broker startup pre-connect for broker-to-server channels (SSE) [pinot]

jineshparakh opened a new pull request, #19407:
URL: https://github.com/apache/pinot/pull/19407

   ## Summary
   
   On a freshly (re)started broker, the broker to server Netty channels start empty. The **first** query to
   each server therefore pays the blocking TCP `connect()`, and, when broker to server TLS is enabled, the
   full TLS handshake, **on its own critical path** while holding the per server channel lock. Under a cold
   burst of concurrent queries this serializes: every request thread targeting that server blocks on the one
   in flight connect, and the TLS handshake (two round trips plus certificate validation) lengthens the
   critical section.
   
   This PR opens all broker to server channels **ahead of query traffic** during startup, behind a readiness
   gate. When it is enabled the broker reports `STARTING` until the channels are connected (or a budget
   expires), so no traffic is routed to it until the connect and handshake cost has already been paid off the
   query path. It is a startup only, best effort warmup: on by default, bounded so it can never stall a
   rolling restart, and a no op reversion to the existing lazy connect path when disabled.
   
   ## What it does
   
   1. **After Helix converges** at startup (the point at which routing, and the servers it references,
      exist), a background thread opens a channel to every routable server, for **both** table types.
      `ServerRoutingInstance` identity includes the table type, so OFFLINE and REALTIME are separate channels
      to the same physical server.
   2. Pre-connect explicitly **awaits the TLS handshake** on the connecting thread
      (`SslHandler.handshakeFuture().sync()`), so the handshake, not just the TCP connect, is off the first
      query's critical path. On a plaintext channel there is no `SslHandler` in the pipeline and this step is
      a no op.
   3. **Readiness is gated** on completion: the broker's `ServiceStatus` reports `STARTING` (an existing
      status value, so no new enum and mixed version peers are unaffected) until pre-connect finishes. This
      reuses the existing readiness endpoint that the Kubernetes startup probe already polls; no health
      endpoint or probe change is required.
   4. **Bounded and best effort.** The whole step is capped by a configurable budget; the readiness gate
      opens even if pre-connect throws, is interrupted, or the budget expires, so a slow or unreachable
      server can never hold a broker not ready forever. A channel that fails to connect simply falls back to
      the existing lazy path. Connecting an already active channel is a no op, so this is idempotent.
   
   This is **single stage (SSE) only**. The multi stage (gRPC) and time series paths use a different
   transport and are unaffected.
   
   ## Configuration
   
   | Property | Default | Description |
   |---|---|---|
   | `pinot.broker.startup.preconnect.enabled` | `true` | Open broker to server channels at startup and gate readiness on it. Set to `false` to restore the pure lazy connect path (behaviour then unchanged). |
   | `pinot.broker.startup.preconnect.timeoutMs` | `30000` | Upper bound on the whole pre-connect step (measured from Helix convergence). Channels not connected within the budget fall back to the lazy path. |
   
   ## Metrics
   
   | Metric | Type | Meaning |
   |---|---|---|
   | `STARTUP_PRECONNECT_DURATION_MS` | timer | Duration of pre-connect, from Helix convergence to the last channel connecting or the budget expiring. |
   | `NETTY_CONNECTION_CONNECT_TIME` | timer | Time to establish a single broker to server channel (TCP connect plus TLS handshake). Complements the existing last value gauge with a full distribution, so a burst of connections on a cold broker is analyzable. |
   
   ## Performance
   
   ### Setup
   
   All numbers below are from a **TLS-enabled** single broker test cluster (a real ZK plus controller plus
   server plus broker, not a synthetic mock):
   
   - One broker and one server, with **broker to server TLS on** (`nettytls`), so the channels pre-connect
     opens pay a real TLS handshake. This is exactly the cost the fix targets.
   - One offline table: about **4.0 million documents** across **41 segments**, **MMAP** load mode, single
     replica. It routes to the one server, so the broker holds **2 channels** to it (OFFLINE and REALTIME).
   - **Open loop** synthetic cold load at a fixed arrival rate of about **288 queries/second** (step
     arrival, no ramp). Because arrivals do not wait for responses, under elevated cold latency the in flight
     count builds to **several hundred concurrent requests**, which is what makes the per server connect and
     handshake serialize.
   - Two measurements per arm (`pinot.broker.startup.preconnect.enabled` OFF vs ON), n=10 cold restarts each:
     a per-query `serverStats` comparison of the first-query legs, and an open-loop run split into a **cold**
     window `[0, 8 s)` and a **warm** window `[150, 180 s)` measured from the first traffic at readiness,
     with wall and lock profiles captured.
   - Engagement verified in the broker log every ON run: `Broker pre-connected 2/2 channel(s) in 70 to
     121 ms`, then readiness.
   
   ### Gain
   
   Pre-connect removes the connect and TLS handshake from the first query's critical path. Measured directly
   from the first query's `serverStats` legs (median [range] across 10 cold restarts per arm):
   
   ```
   first-query SubmitDelayMs (connect + channel lock) : 240 ms [167-518]  ->  36 ms [0-53]   (-85%)
   first-second submit p95                            : 257 ms [43-778]   ->  38 ms [18-249]  (-85%)
   ```
   
   The broker to server **TLS handshake** for the 2 channels is now paid once at startup
   (`pre-connected 2/2 channel(s) in 70 to 121 ms`), off the query path, rather than by the first query
   holding the channel lock. This is corroborated by the wall profile: the broker to server
   connect/handshake park category (`QueryRouter.submitQuery` to `ServerChannels.sendRequest` to park) is the
   **only** category that changes between arms, dropping from **0.23% to 0.09%** (about 2.5x). By readiness
   `_channel.isActive()` is `true`, so the lazy `connectWithoutLocking` path no ops and the first query
   serialization never happens.
   
   Readiness cost is negligible: the gate adds effectively no startup time (both arms reached readiness in
   about 90 s), and it opens even on failure, so it cannot delay a rolling restart beyond the configured
   budget.
   
   ## Testing
   
   - **Unit** (`ServerPreConnectorTest`, 6 tests): connects every server for both table types; empty server
     list is a no op; already passed deadline is a no op; counts only successful connects; a throwing connect
     is swallowed and the others still connect; the budget bounds the wait and does not block on slow
     connects. The connector takes its dependencies (routable server supplier, connect function) as
     functions, so parallelism, budget, and failure handling are covered without a live broker.
   - **Integration** (`BrokerServerPreConnectIntegrationTest`): brings up a real ZK plus controller plus
     server plus broker with an offline table and asserts (1) with pre-connect enabled the readiness gate
     opens, the broker's `ServiceStatus` reaches `GOOD`; and (2) the production path (`RoutingManager`
     routable server supplier to `QueryRouter` to `ServerChannels` to a live server) opens one channel per
     (server, table type).
   
   ## Backward compatibility
   
   - Readiness reports the existing `STARTING` status, with **no new status enum value**, so mixed version
     broker/controller peers are unaffected.
   - With `preconnect.enabled=false` the code path is a strict no op and behaviour is identical to before.
   - No wire protocol or serialization changes.


-- 
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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add broker startup pre-connect for broker-to-server channels (SSE) [pinot]

Posted by "jineshparakh (via GitHub)" <gi...@apache.org>.

jineshparakh commented on PR #19407:
URL: https://github.com/apache/pinot/pull/19407#issuecomment-5571586191

   >One non-blocking follow-up is in the grace-window thread — with more channels than worker threads and connects slower than the window, the heuristic reads the gap between pool waves as a straggler and releases early (measured 16/48 with every server healthy). It degrades safely, since shutdown() lets the queued tasks finish and publish, so it is not a merge blocker; there is a measured one-line fix in that thread if you want it, and it is fine to leave as-is too.
   
   
   @gortiz I'd like to take this in a follow up PR if it's fine.


-- 
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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add broker startup pre-connect for broker-to-server channels (SSE) [pinot]

Posted by "gortiz (via GitHub)" <gi...@apache.org>.

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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add broker startup pre-connect for broker-to-server channels (SSE) [pinot]

Posted by "jineshparakh (via GitHub)" <gi...@apache.org>.

jineshparakh commented on PR #19407:
URL: https://github.com/apache/pinot/pull/19407#issuecomment-5510545020

Thanks for the review — it was unusually useful, and two of the items turned out to be load-bearing rather than cosmetic. All six inline comments are addressed in `37ad9142`; replies are on the individual threads. Here are the three you raised at the top level.

### On the test gap

You were right, and it would have passed with the feature stubbed out. Worse than you spotted: `registerServiceStatusHandler()` computes `resourcesToMonitor` from the `brokerResource` ideal state at registration time, and with no table added yet this broker is in no partition — so the Helix callbacks report `GOOD` immediately and `preConnect()` returns 0 on an empty server map.

`startServer()` now precedes `startBroker()`, and the assertions are stronger: one channel per (server, table type), plus a slow-connector case proving readiness is released within the budget.

Separately, the TLS path had no coverage at all, which is worse given that TLS is the reason the change exists. Added `TlsIntegrationTest#testPreConnectOpensTlsChannelsToEveryServer` — that cluster runs its server with `netty.enabled=false` / `nettytls.enabled=true`, so every single-stage channel the broker opens carries a real `SslHandler`. Pre-connect counts a channel as connected only once the handshake completes, so asserting the channel count there is what proves the handshake is actually awaited. Without it the TLS path could break silently, since pre-connect swallows its own failures by design and falls back to the lazy path — every other assertion in that class would still pass.

### 1. Is the readiness gate worth it?

I've kept it, but the mechanism is now much smaller, which I think changes the trade.

It no longer adds anything: no new status value, no extra `ServiceStatusCallback`, no separate completion flag. Pre-connect is part of startup, so it simply keeps `_isStarting` set until it finishes and the `LifecycleServiceStatusCallback` that is already registered reports `STARTING` as it always would. That also removes the ordering hazard the first version needed a comment to explain away — `_isStarting` is set before the status handler is registered, so there is structurally no un-gated window rather than one avoided by care. `BaseBrokerStarter` is a net deletion as a result.

On the probe risk you are right that the broker has only `/health`, with no liveness/readiness split. One correction though: the bundled chart does **not** define a startupProbe for the broker — `values.yaml` has `startupEnabled: false` with `livenessEnabled: true` pointed at `/health`. So the safety there comes from `failureThreshold: 10 x periodSeconds: 10` = 100s against a 30s budget, not from a startupProbe. That makes your underlying concern stronger, not weaker.

The real fix is to give the broker `/health/liveness` and `/health/readiness`, mirroring what `pinot-server` has had for a long time (`HealthCheckResource`). I prototyped it and then took it back out: it is a pre-existing gap that any future readiness gate hits, and bundling an endpoint addition into this change is the wrong trade. I'd rather do it as its own PR — happy to open one.

### 2. Is one-shot warm-up the right scope?

Agreed on all three limitations, and none of them are addressed here.

One correction on the mechanism for full-cluster cold start: `getRoutableServerInstanceMap()` is built from persistent instance configs, not live instances, so the servers *are* in the map and the connects fail fast — rather than the map being empty. The outcome is the one you called, though: `connected == 0` and the step achieves nothing. Scale-up and later-joining servers are likewise uncovered.

Hooking routing changes, or moving channel establishment out of the `sendRequest` lock into a shared per-channel future, would cover all three. I'd like to keep those separate — the second in particular is a hot-path change, and this PR's measured win is specifically the startup case.

### On the metric name

Fixed, though not as `NETTY_CONNECTION_CONNECT_TIME_MS`. Both enums derive their metric name from the constant via `Utils.toCamelCase(name().toLowerCase())`, so that name renders identically to the existing `BrokerGauge.NETTY_CONNECTION_CONNECT_TIME_MS` and the two collide in the metrics registry — `YammerSettableGauge cannot be cast to Timer`, which failed three tests the moment I made the change you asked for.

Renamed to `NETTY_CONNECTION_CONNECT_LATENCY_MS`: carries the `_MS` suffix, matches `NETTY_CONNECTION_SEND_REQUEST_LATENCY` next door, and is no longer confusable with the gauge. The timer is new in this PR, so nothing published is being renamed. Your nit was more than a nit — thanks.

--
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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add broker startup pre-connect for broker-to-server channels (SSE) [pinot]

Posted by "gortiz (via GitHub)" <gi...@apache.org>.

gortiz commented on code in PR #19407:
URL: https://github.com/apache/pinot/pull/19407#discussion_r3895695617


##########
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:
   `MAX_CONNECT_THREADS = 16` acts as a starvation cliff here, not just a throughput cap. The Bootstrap sets no `ChannelOption.CONNECT_TIMEOUT_MILLIS` (`ServerChannels.java:179-180`), so Netty's 30s default applies — which is exactly `DEFAULT_BROKER_STARTUP_PRECONNECT_TIMEOUT_MS`. Sixteen servers whose SYN is dropped rather than refused (booting, or a security-group/firewall black hole) park the entire pool for the whole budget: zero channels connected, and the readiness gate held the full 30s having achieved nothing.
   
   `ECONNREFUSED` returns in microseconds, so ordinary cold start is fine — but the case the 30s budget exists *for* is precisely the one where the thread count is the binding constraint. Setting `CONNECT_TIMEOUT_MILLIS` on the pre-connect path, so a single connect can't outlive the budget, would bound this independently of the pool size.
   
   Relatedly, the comment on line 95 says "no head-of-line blocking". That's true of the `ExecutorCompletionService`, which removes head-of-line blocking from the *counting*; it doesn't remove it from *execution*, where the fixed workers are the queue. Worth rewording so it doesn't read as a stronger guarantee than it makes.



-- 
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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add broker startup pre-connect for broker-to-server channels (SSE) [pinot]

Posted by "gortiz (via GitHub)" <gi...@apache.org>.

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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add broker startup pre-connect for broker-to-server channels (SSE) [pinot]

Posted by "jineshparakh (via GitHub)" <gi...@apache.org>.

jineshparakh commented on code in PR #19407:
URL: https://github.com/apache/pinot/pull/19407#discussion_r3948975683


##########
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:
   Done



-- 
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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add broker startup pre-connect for broker-to-server channels (SSE) [pinot]

Posted by "jineshparakh (via GitHub)" <gi...@apache.org>.

jineshparakh commented on code in PR #19407:
URL: https://github.com/apache/pinot/pull/19407#discussion_r3914754942


##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -236,11 +237,45 @@ void sendRequest(String rawTableName, AsyncQueryResponse asyncQueryResponse,
 
     void connectWithoutLocking()

Review Comment:
   Good catch, and you're right that it made the backward-compatibility claim inaccurate. Fixed: `recordConnectTime()` sets both the gauge and the new timer, and it is now called from `connectWithoutLocking()` as well — so the lazy path, which is the only connect path when `enabled=false`, records exactly as it did before.
   



##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -236,11 +237,45 @@ void sendRequest(String rawTableName, AsyncQueryResponse asyncQueryResponse,
 
     void connectWithoutLocking()
         throws InterruptedException {
+      // Lazy query path: open the TCP connection only, exactly as before this feature existed. Any TLS
+      // handshake is left to proceed asynchronously so the channel lock is released as soon as the socket
+      // is up, keeping the first query's critical section short.
+      if (_channel == null || !_channel.isActive()) {
+        _channel = _bootstrap.connect().sync().channel();
+      }
+    }
+
+    /// Like [#connectWithoutLocking()] but additionally waits out the TLS handshake and records the
+    /// establish latency. Used only by startup pre-connect ([#connect()]), so paying the handshake -- and
+    /// the longer critical section it implies -- never touches the lazy query path. Runs under the channel
+    /// lock, like its sibling.
+    void connectAndAwaitHandshakeWithoutLocking()
+        throws InterruptedException {
       if (_channel == null || !_channel.isActive()) {
         long startTime = System.currentTimeMillis();
         _channel = _bootstrap.connect().sync().channel();
-        _brokerMetrics.setValueOfGlobalGauge(BrokerGauge.NETTY_CONNECTION_CONNECT_TIME_MS,
-            System.currentTimeMillis() - startTime);
+        awaitTlsHandshake(_channel);
+        long connectTimeMs = System.currentTimeMillis() - startTime;
+        _brokerMetrics.setValueOfGlobalGauge(BrokerGauge.NETTY_CONNECTION_CONNECT_TIME_MS, connectTimeMs);
+        _brokerMetrics.addTimedValue(BrokerTimer.NETTY_CONNECTION_CONNECT_TIME, connectTimeMs,
+            TimeUnit.MILLISECONDS);
+      }
+    }
+
+    /// Blocks until the TLS handshake on a freshly-connected channel completes.
+    ///
+    /// `bootstrap.connect().sync()` returns once the TCP connection is up; the client-mode [SslHandler]
+    /// then drives the handshake asynchronously on the event loop. Awaiting its future here pays the
+    /// handshake -- two round trips plus certificate validation -- on the connecting thread rather than
+    /// on the first query that writes to the channel. On a plaintext channel there is no [SslHandler] in
+    /// the pipeline and this is a no-op. The calling thread is never an event-loop thread, so this
+    /// cannot deadlock. A failed handshake propagates as an unchecked exception, which the caller
+    /// treats exactly like a failed connect.
+    private void awaitTlsHandshake(Channel channel)
+        throws InterruptedException {
+      SslHandler sslHandler = channel.pipeline().get(SslHandler.class);
+      if (sslHandler != null) {
+        sslHandler.handshakeFuture().sync();

Review Comment:
   Fixed. `_channel` is assigned only after `awaitTlsHandshake` returns; on failure the channel is closed and the exception propagates, so a doomed channel is never published.
   
   Worth recording why the race is real rather than theoretical: in `SslHandler#setHandshakeFailure`, `Promise#tryFailure` runs *before* `SslUtils#handleHandshakeFailure` -> `ctx.close()`, and that close is itself asynchronous. So a channel assigned up front genuinely can still report `isActive()` at the moment the waiter throws.
   
   On the 10s bound — rather than document the dependency I removed it. The handshake is now awaited with `await(remainingBudget)` instead of `sync()`, so it is bounded by the caller's deadline rather than by `SslHandler`'s default.
   



##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -271,7 +306,7 @@ void connect()
         throws InterruptedException, TimeoutException {
       if (_channelLock.tryLock(TRY_CONNECT_CHANNEL_LOCK_TIMEOUT_MS, TimeUnit.MILLISECONDS)) {
         try {
-          connectWithoutLocking();
+          connectAndAwaitHandshakeWithoutLocking();

Review Comment:
   You're right, and the javadoc was simply false. Split as you suggested: `connect()` is back on the TCP-only `connectWithoutLocking()`, and startup pre-connect has its own `ServerChannels#preConnect` entry point that awaits the handshake. No handshake is held under `_channelLock` on any path that runs under live traffic.
   
   One refinement on severity, for the record: unhealthy servers are removed from routing (`BaseBrokerStarter` wires `registerUnhealthyServerNotifier` -> `excludeServerFromRouting`), so the concurrent-query pile-up is narrower than described. The residual that did matter is that the single `failure-detector-retry` thread processes servers sequentially, so one hung TLS peer would have delayed every *other* server's recovery by up to the handshake timeout. Fixed either way.
   



-- 
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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add broker startup pre-connect for broker-to-server channels (SSE) [pinot]

Posted by "jineshparakh (via GitHub)" <gi...@apache.org>.

jineshparakh commented on PR #19407:
URL: https://github.com/apache/pinot/pull/19407#issuecomment-5520855865

   @gortiz can you please re-review?


-- 
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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add broker startup pre-connect for broker-to-server channels (SSE) [pinot]

Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.

codecov-commenter commented on PR #19407:
URL: https://github.com/apache/pinot/pull/19407#issuecomment-5479617873

   ## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/19407?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
   :x: Patch coverage is `66.66667%` with `6 lines` in your changes missing coverage. Please review.
   :white_check_mark: Project coverage is 57.66%. Comparing base ([`80dbb7d`](https://app.codecov.io/gh/apache/pinot/commit/80dbb7d02dfd92ccc20ae2606ac712ea4f73d3cf?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`e8caf61`](https://app.codecov.io/gh/apache/pinot/commit/e8caf615a1c4d529e88c7cb983d5921fd642bd1e?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
   
   | [Files with missing lines](https://app.codecov.io/gh/apache/pinot/pull/19407?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
   |---|---|---|
   | [...a/org/apache/pinot/core/transport/QueryRouter.java](https://app.codecov.io/gh/apache/pinot/pull/19407?src=pr&el=tree&filepath=pinot-core%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcore%2Ftransport%2FQueryRouter.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS90cmFuc3BvcnQvUXVlcnlSb3V0ZXIuamF2YQ==) | 0.00% | [4 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/19407?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
   | [...rg/apache/pinot/core/transport/ServerChannels.java](https://app.codecov.io/gh/apache/pinot/pull/19407?src=pr&el=tree&filepath=pinot-core%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcore%2Ftransport%2FServerChannels.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS90cmFuc3BvcnQvU2VydmVyQ2hhbm5lbHMuamF2YQ==) | 83.33% | [1 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/pinot/pull/19407?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
   > :exclamation:  There is a different number of reports uploaded between BASE (80dbb7d) and HEAD (e8caf61). Click for more details.
   > 
   > <details><summary>HEAD has 24 uploads less than BASE</summary>
   >
   >| Flag | BASE (80dbb7d) | HEAD (e8caf61) |
   >|------|------|------|
   >|unittests|2|1|
   >|java-25|6|1|
   >|temurin|6|1|
   >|lane-a|2|0|
   >|integration|4|0|
   >|integration1|2|0|
   >|lane-b|2|0|
   >|integration2|2|0|
   >|unittests2|1|0|
   ></details>
   
   <details><summary>Additional details and impacted files</summary>
   
   
   
   ```diff
   @@             Coverage Diff              @@
   ##             master   #19407      +/-   ##
   ============================================
   - Coverage     67.56%   57.66%   -9.90%     
   + Complexity     1430        1    -1429     
   ============================================
     Files          3486     2683     -803     
     Lines        224173   163997   -60176     
     Branches      35381    26629    -8752     
   ============================================
   - Hits         151462    94570   -56892     
   - Misses        60684    61419     +735     
   + Partials      12027     8008    -4019     
   ```
   
   | [Flag](https://app.codecov.io/gh/apache/pinot/pull/19407/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
   |---|---|---|
   | [integration](https://app.codecov.io/gh/apache/pinot/pull/19407/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [integration1](https://app.codecov.io/gh/apache/pinot/pull/19407/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [integration2](https://app.codecov.io/gh/apache/pinot/pull/19407/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [java-25](https://app.codecov.io/gh/apache/pinot/pull/19407/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `57.66% <66.66%> (-9.90%)` | :arrow_down: |
   | [lane-a](https://app.codecov.io/gh/apache/pinot/pull/19407/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [lane-b](https://app.codecov.io/gh/apache/pinot/pull/19407/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [temurin](https://app.codecov.io/gh/apache/pinot/pull/19407/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `57.66% <66.66%> (-9.90%)` | :arrow_down: |
   | [unittests](https://app.codecov.io/gh/apache/pinot/pull/19407/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `57.66% <66.66%> (-9.90%)` | :arrow_down: |
   | [unittests1](https://app.codecov.io/gh/apache/pinot/pull/19407/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `57.66% <66.66%> (-0.01%)` | :arrow_down: |
   | [unittests2](https://app.codecov.io/gh/apache/pinot/pull/19407/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   
   Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
   </details>
   
   [:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/pinot/pull/19407?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).   
   :loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
   <details><summary> :rocket: New features to boost your workflow: </summary>
   
   - :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
   </details>


-- 
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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add broker startup pre-connect for broker-to-server channels (SSE) [pinot]

Posted by "gortiz (via GitHub)" <gi...@apache.org>.

gortiz commented on code in PR #19407:
URL: https://github.com/apache/pinot/pull/19407#discussion_r3895695572


##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -236,11 +237,45 @@ void sendRequest(String rawTableName, AsyncQueryResponse asyncQueryResponse,
 
     void connectWithoutLocking()

Review Comment:
   The lazy query path no longer records `BrokerGauge.NETTY_CONNECTION_CONNECT_TIME_MS`. Before this PR, every connect set that gauge; now only `connectAndAwaitHandshakeWithoutLocking()` does, and `sendRequest()` -> `connectWithoutLocking()` records nothing.
   
   That makes the backward-compatibility claim inaccurate: with `preconnect.enabled=false` — documented as "a strict no-op, behaviour identical to before" — this is the *only* connect path, so any existing dashboard or alert on `NETTY_CONNECTION_CONNECT_TIME_MS` goes permanently flat. And even with the flag on, the gauge's meaning silently narrows from "time to establish any channel" to "time to establish a pre-connect / failure-detector channel".
   
   Suggest keeping the gauge (and ideally the new timer) in `connectWithoutLocking()` as well. Two `System.currentTimeMillis()` calls are not what makes the critical section long — the `sync()` is.



##########
pinot-core/src/main/java/org/apache/pinot/core/transport/QueryRouter.java:
##########
@@ -156,19 +156,36 @@ public boolean hasChannel(ServerInstance serverInstance) {
     }
   }
 
-  /// Connects to the given server, returns `true` if the server is successfully connected.
+  /// Connects the OFFLINE channel to the given server, returns `true` if it is successfully connected.
+  ///
+  /// Unchanged in behaviour: this is the reachability probe the failure detector already used (it
+  /// opened the OFFLINE channel), now expressed as a one-line delegation. It deliberately opens a
+  /// single channel rather than every channel the server may need. Callers that want the server fully
+  /// connected -- startup pre-connect, for instance -- should use [#connect(ServerInstance, TableType)]
+  /// for each table type instead.
   public boolean connect(ServerInstance serverInstance) {

Review Comment:
   A side effect that isn't called out in the description: pre-connect calls `ServerChannels.connect()` for every routable server, and that does `computeIfAbsent` on `_serverToChannelMap` — the entry is created even when the connect itself fails. Since `hasChannel()` (line 149) tests for the OFFLINE entry, it becomes unconditionally `true` after pre-connect.
   
   That kills the escape hatch at `SingleConnectionBrokerRequestHandler.retryUnhealthyServer():436`, which returns `ServerState.UNKNOWN` when `!hasChannel(serverInstance)` specifically so an MSE-only cluster doesn't drive its servers through the SSE-channel health state machine. With pre-connect on (the default) that branch is dead, and MSE-only clusters will start marking servers UNHEALTHY based on SSE reachability.
   
   Either outcome may be defensible, but it should be a deliberate decision with a test pinning it.



##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -271,7 +306,7 @@ void connect()
         throws InterruptedException, TimeoutException {
       if (_channelLock.tryLock(TRY_CONNECT_CHANNEL_LOCK_TIMEOUT_MS, TimeUnit.MILLISECONDS)) {
         try {
-          connectWithoutLocking();
+          connectAndAwaitHandshakeWithoutLocking();

Review Comment:
   The javadoc on `connectAndAwaitHandshakeWithoutLocking()` says it is "used only by startup pre-connect", but `ServerChannels.connect()` is also the failure detector's reconnect probe: `SingleConnectionBrokerRequestHandler.retryUnhealthyServer()` -> `QueryRouter.connect()` -> here. That path runs at steady state, under live traffic.
   
   So this change means the probe now holds `_channelLock` across the entire TLS handshake. `sendRequest()` acquires the same lock with `tryLock(queryTimeoutMs)`, so concurrent queries to a server that just came back queue behind the handshake — bounded only by Netty's default 10s `handshakeTimeoutMillis`, since `ChannelHandlerFactory.getClientTlsHandler` doesn't set one. That is the same serialization this PR sets out to remove, reintroduced on the runtime path.
   
   Suggest splitting the two callers: keep `connect()` on the TCP-only variant, and give startup pre-connect its own entry point that awaits the handshake.



##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -236,11 +237,45 @@ void sendRequest(String rawTableName, AsyncQueryResponse asyncQueryResponse,
 
     void connectWithoutLocking()
         throws InterruptedException {
+      // Lazy query path: open the TCP connection only, exactly as before this feature existed. Any TLS
+      // handshake is left to proceed asynchronously so the channel lock is released as soon as the socket
+      // is up, keeping the first query's critical section short.
+      if (_channel == null || !_channel.isActive()) {
+        _channel = _bootstrap.connect().sync().channel();
+      }
+    }
+
+    /// Like [#connectWithoutLocking()] but additionally waits out the TLS handshake and records the
+    /// establish latency. Used only by startup pre-connect ([#connect()]), so paying the handshake -- and
+    /// the longer critical section it implies -- never touches the lazy query path. Runs under the channel
+    /// lock, like its sibling.
+    void connectAndAwaitHandshakeWithoutLocking()
+        throws InterruptedException {
       if (_channel == null || !_channel.isActive()) {
         long startTime = System.currentTimeMillis();
         _channel = _bootstrap.connect().sync().channel();
-        _brokerMetrics.setValueOfGlobalGauge(BrokerGauge.NETTY_CONNECTION_CONNECT_TIME_MS,
-            System.currentTimeMillis() - startTime);
+        awaitTlsHandshake(_channel);
+        long connectTimeMs = System.currentTimeMillis() - startTime;
+        _brokerMetrics.setValueOfGlobalGauge(BrokerGauge.NETTY_CONNECTION_CONNECT_TIME_MS, connectTimeMs);
+        _brokerMetrics.addTimedValue(BrokerTimer.NETTY_CONNECTION_CONNECT_TIME, connectTimeMs,
+            TimeUnit.MILLISECONDS);
+      }
+    }
+
+    /// Blocks until the TLS handshake on a freshly-connected channel completes.
+    ///
+    /// `bootstrap.connect().sync()` returns once the TCP connection is up; the client-mode [SslHandler]
+    /// then drives the handshake asynchronously on the event loop. Awaiting its future here pays the
+    /// handshake -- two round trips plus certificate validation -- on the connecting thread rather than
+    /// on the first query that writes to the channel. On a plaintext channel there is no [SslHandler] in
+    /// the pipeline and this is a no-op. The calling thread is never an event-loop thread, so this
+    /// cannot deadlock. A failed handshake propagates as an unchecked exception, which the caller
+    /// treats exactly like a failed connect.
+    private void awaitTlsHandshake(Channel channel)
+        throws InterruptedException {
+      SslHandler sslHandler = channel.pipeline().get(SslHandler.class);
+      if (sslHandler != null) {
+        sslHandler.handshakeFuture().sync();

Review Comment:
   `handshakeFuture().sync()` rethrows a handshake failure *after* `_channel` has already been assigned on line 256. The field is then left pointing at a channel whose handshake failed, and until Netty finishes closing it, `_channel.isActive()` can still return `true` — at which point `connectWithoutLocking()` will write a query into it.
   
   Narrow race, but cheap to close: assign `_channel` only after `awaitTlsHandshake` succeeds, or close/null it in a `catch` before propagating.
   
   Also worth reflecting in the javadoc: `sync()` here is bounded by Netty's default `SslHandler` handshake timeout (10s), not by the caller's deadline, so one hung TLS peer parks a pre-connect worker for 10s regardless of the configured budget.



##########
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:
   Connecting both `TableType` values unconditionally doubles the channel count regardless of what the cluster actually routes. On an offline-only cluster, every broker opens and then holds N idle REALTIME TLS connections that will never carry a query — and pays N extra handshakes, on both ends, at every broker restart. `ServerChannel` entries are never evicted from `_serverToChannelMap`, so they persist for the process lifetime.
   
   `RoutingManager` already knows which table types route to which server. Deriving the (server, tableType) pairs from routing instead of taking the cross product would be exactly right on hybrid clusters and halve the work on single-type ones. It also shrinks the `hasChannel()` side effect noted on `QueryRouter`.



##########
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:
   Two things here — one concrete bug in the bound, one suggestion on the threading model.
   
   **`MAX_CONNECT_THREADS = 16` is a starvation cliff, not just a throughput cap.** The Bootstrap sets no `ChannelOption.CONNECT_TIMEOUT_MILLIS` (`ServerChannels.java:179-180`), so Netty's 30s default applies — which is exactly `DEFAULT_BROKER_STARTUP_PRECONNECT_TIMEOUT_MS`. Sixteen servers whose SYN is dropped rather than refused (booting, or a security-group/firewall black hole) park the entire pool for the whole budget: zero channels connected, and the readiness gate held the full 30s having achieved nothing. `ECONNREFUSED` returns in microseconds, so ordinary cold start is fine — but the case the 30s budget exists *for* is precisely the one where 16 threads is the binding constraint. Worth setting `CONNECT_TIMEOUT_MILLIS` on the pre-connect path so a single connect can't outlive the budget.
   
   Relatedly, the comment on line 95 says "no head-of-line blocking". That's true of the `ExecutorCompletionService`, which removes head-of-line blocking from the *counting*; it doesn't remove it from *execution*, where the 16 fixed workers are the queue. Suggest rewording so it doesn't read as a stronger guarantee than it makes.
   
   **On the threading model**, three options, ranked:
   
   1. *Don't block at all.* `Bootstrap.connect()` already returns a `ChannelFuture` and the real I/O runs on `_eventLoopGroup` — every thread here exists only because `ServerChannels` calls `.sync()`. Add a non-blocking connect entry point that returns the future, fire all N from the single pre-connect thread, then wait once on the aggregate with the remaining budget. Zero extra threads, no cap, no starvation. And because you still hold the futures, the timeout path can `cancel()` + `close()` them, which also fixes the orphaned-channel leak below.
   2. *`Executors.newVirtualThreadPerTaskExecutor()`.* One virtual thread per channel, so the 16-worker cliff disappears and the budget becomes the only bound. Safe on this repo's JDK 25 baseline: Netty 4.1.137's `DefaultPromise.await()` blocks in `synchronized (this) { wait(); }`, which pinned the carrier before JDK 24 and no longer does after JEP 491. Note this would be the repo's first virtual-thread usage — nothing currently calls `ofVirtual()` or `newVirtualThreadPerTaskExecutor()` — so it's a project-level precedent worth a maintainer's opinion, not just a local choice.
   3. *Either way, keep a bound, but on the right axis.* Virtual threads add no I/O capacity: the connects and, more importantly, the TLS handshake crypto (cert-chain verification) all run on `_eventLoopGroup`, which is **shared with the query path**. Unbounded fan-out on a large tenant trades thread starvation for event-loop saturation at exactly the moment the broker is about to take traffic. A `Semaphore` over in-flight handshakes expresses that bound directly; a fixed thread pool only expresses it by accident.



-- 
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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add broker startup pre-connect for broker-to-server channels (SSE) [pinot]

Posted by "jineshparakh (via GitHub)" <gi...@apache.org>.

jineshparakh commented on code in PR #19407:
URL: https://github.com/apache/pinot/pull/19407#discussion_r3914757114


##########
pinot-core/src/main/java/org/apache/pinot/core/transport/QueryRouter.java:
##########
@@ -156,19 +156,36 @@ public boolean hasChannel(ServerInstance serverInstance) {
     }
   }
 
-  /// Connects to the given server, returns `true` if the server is successfully connected.
+  /// Connects the OFFLINE channel to the given server, returns `true` if it is successfully connected.
+  ///
+  /// Unchanged in behaviour: this is the reachability probe the failure detector already used (it
+  /// opened the OFFLINE channel), now expressed as a one-line delegation. It deliberately opens a
+  /// single channel rather than every channel the server may need. Callers that want the server fully
+  /// connected -- startup pre-connect, for instance -- should use [#connect(ServerInstance, TableType)]
+  /// for each table type instead.
   public boolean connect(ServerInstance serverInstance) {

Review Comment:
   Fixed, and it needed a different fix than I first reached for. Withholding the map entry on a *failed* connect is not enough: a reachable server still flips `hasChannel()`, and that is most MSE-only clusters.
   
   `hasChannel()` now means "has a query been sent, or attempted, to this server" — a flag set only in `sendRequest`. That restores the invariant that already held before this PR: `connect()` is unreachable until `hasChannel()` is already true, so an entry could only ever originate from `sendRequest`.
   
   This turned out to matter more than the `UNKNOWN` branch being dead. `_unhealthyServerRetriers` is an ordered list and the loop breaks on the first `UNHEALTHY`, and the single-stage retrier is registered before the multi-stage one — so a server reachable over gRPC but not over Netty would have been voted `UNHEALTHY` and dropped from routing without `MultiStageBrokerRequestHandler#retryUnhealthyServer` ever being consulted.
   
   Pinned by four tests in `ServerChannelsTest`: pre-connect leaves `hasChannel()` false whether it succeeds or fails, `sendRequest` is what makes it true (including when the connect itself fails, as before), and the first query reuses the pre-connected channel rather than opening a second one.
   



##########
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:
   I'd like to keep the cross product, and I think the reasoning cuts the other way here.
   
   Pre-connect is one-shot at startup with no re-run on routing change. Connecting only the table types a server serves *today* leaves a table added later cold — so the narrower set is worse on exactly the case where a broker is long-lived, which is the common one.
   
   The `hasChannel()` side effect you linked to this is fixed independently (see the `QueryRouter` thread), so it no longer argues for narrowing.
   
   Deriving the pairs from routing also needs a new routable-table enumeration on the `RoutingManager` interface — `getServingInstances(tableNameWithType)` exists but there is no way to enumerate the tables — which felt like more surface than this PR should take on for the benefit.
   
   That said, the cost you identify is real and I don't want to wave it away: `getRoutableServerInstanceMap()` is built from every enabled server instance config in the cluster with no tenant filter, so on a large multi-tenant cluster this is 2 x (all servers), not 2 x (my tenant's servers), and the entries are never evicted. If you think that outweighs the table-added-later case I'm happy to switch — it's a contained change.
   



##########
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:
   Both points taken.
   
   The bootstrap is now cloned per pre-connect with `ChannelOption.CONNECT_TIMEOUT_MILLIS` set from the remaining budget, so a single connect cannot outlive it, and the query path keeps Netty's default. Implementing that surfaced a trap worth flagging: Netty only schedules its connect-timeout task when the value is `> 0` (`AbstractNioUnsafe#connect`), so a spent budget passed straight through would have meant *no* timeout at all. There is an explicit guard before the option is set.
   
   Being straight about the limit, since I don't want to overclaim: this bounds a single connect, not the aggregate. Sixteen concurrently black-holed servers still occupy all sixteen workers for the budget, because each starts with the full remaining time. Making the cliff go away entirely would need a per-connect cap well below the budget so workers recycle, and that penalises legitimately slow-but-reachable servers under a TLS handshake — happy to add one if you'd prefer that trade. The readiness gate opens on schedule regardless.
   
   Comment reworded too — you're right that the completion service removes head-of-line blocking from the *counting*, not from execution.
   



-- 
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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add broker startup pre-connect for broker-to-server channels (SSE) [pinot]

Posted by "gortiz (via GitHub)" <gi...@apache.org>.

gortiz merged PR #19407:
URL: https://github.com/apache/pinot/pull/19407


-- 
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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add broker startup pre-connect for broker-to-server channels (SSE) [pinot]

Posted by "gortiz (via GitHub)" <gi...@apache.org>.

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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add broker startup pre-connect for broker-to-server channels (SSE) [pinot]

Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.

xiangfu0 commented on PR #19407:
URL: https://github.com/apache/pinot/pull/19407#issuecomment-5572962536

   Documentation follow-up: https://github.com/pinot-contrib/pinot-docs/pull/1035 (merged). It documents the new default-on broker startup pre-connect settings, readiness and timeout behavior, transport scope, tuning examples, and metrics.


-- 
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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add broker startup pre-connect for broker-to-server channels (SSE) [pinot]

Posted by "jineshparakh (via GitHub)" <gi...@apache.org>.

jineshparakh commented on code in PR #19407:
URL: https://github.com/apache/pinot/pull/19407#discussion_r3948973811


##########
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:
   Done



-- 
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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org