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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
