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: [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]
