xiangfu0 commented on code in PR #20022:
URL: https://github.com/apache/pinot/pull/20022#discussion_r4366206643
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java:
##########
@@ -520,6 +546,7 @@ private void processInstanceConfigChangeInternal() {
// always contains the selected servers
for (String newDisabledInstance : newDisabledServers) {
_enabledServerInstanceMap.remove(newDisabledInstance);
+ _previouslyEnabledServers.add(newDisabledInstance);
Review Comment:
`_previouslyEnabledServers` only ever has entries removed on re-enable, so a
server permanently removed from the cluster lingers here. It's bounded by the
set of instance IDs the broker has seen (same shape as `_excludedServers`), so
not a real leak — just noting it never gets pruned on permanent removal.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -264,17 +519,40 @@ void sendRequest(String rawTableName, AsyncQueryResponse
asyncQueryResponse,
}
}
+ private void checkNotEvicted() {
+ if (_evicted) {
+ closeChannel();
+ throw new IllegalStateException("Channel to server: " +
_serverRoutingInstance + " has been evicted");
+ }
+ }
+
/// Lazy query path: opens the TCP connection only. 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. Startup pre-connect uses
[#preConnectWithoutLocking(long)]
/// instead, which additionally pays the handshake.
+ ///
+ /// Also replaces a channel marked suspect: the server stopped answering
on it while the socket still looked
+ /// active, which is what a half-open socket looks like until the kernel
gives up on it, so `isActive()` is not
+ /// trusted. The suspect channel is dropped before the connect, so a
failed connect does not leave it in place,
+ /// and whatever is still in flight on it fails without the server being
reported (nothing on it was being
+ /// answered anyway).
void connectWithoutLocking()
throws InterruptedException {
- if (_channel == null || !_channel.isActive()) {
- long startTime = System.currentTimeMillis();
- _channel = _bootstrap.connect().sync().channel();
- recordConnectTime(System.currentTimeMillis() - startTime);
+ Channel channel = _channel;
+ if (channel != null && channel.isActive() && !_suspect) {
+ return;
}
+ if (channel != null && _suspect) {
+ LOGGER.info("Replacing suspect channel to server: {} with a fresh
connection", _serverRoutingInstance);
+ _channel = null;
+ closeLocally(channel);
+ }
+ _suspect = false;
+ long startTime = System.currentTimeMillis();
+ _channel = _bootstrap.connect().sync().channel();
Review Comment:
`_bootstrap.connect().sync()` here runs while holding `_channelLock`, same
as before this PR, so no new concern. Just noting a suspect-replacement now
always pays a fresh connect under the lock on the query path, which serializes
other queries to the same server briefly during recovery — expected given the
note about not awaiting TLS.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -169,6 +273,71 @@ public void connect(ServerRoutingInstance
serverRoutingInstance)
_serverToChannelMap.computeIfAbsent(serverRoutingInstance,
ServerChannel::new).connect();
}
+ /// Marks the cached channel to the given routing instance suspect: the
server stopped answering queries on it
+ /// while the socket still looked active. The next query to it, or the next
[#probe], replaces it with a fresh
+ /// connection instead of trusting `isActive()`.
+ public void markSuspect(ServerRoutingInstance serverRoutingInstance) {
+ ServerChannel serverChannel =
_serverToChannelMap.get(serverRoutingInstance);
+ if (serverChannel != null) {
+ serverChannel._suspect = true;
+ }
+ }
+
+ /// The failure detector's retry probe for one channel. Does nothing and
returns `false` when this broker has not
+ /// queried the server over this channel (see [#hasChannel]); otherwise
reconnects the channel if it is down,
+ /// replaces it with a fresh connection if it was marked suspect, and
returns `true`, throwing if the connection
+ /// cannot be established.
+ public boolean probe(ServerRoutingInstance serverRoutingInstance)
+ throws InterruptedException, TimeoutException {
+ ServerChannel serverChannel =
_serverToChannelMap.get(serverRoutingInstance);
+ if (serverChannel == null || !serverChannel._openedByQuery) {
+ return false;
+ }
+ serverChannel.connect();
+ return true;
+ }
+
+ /// Records a response from the server on the given channel: it is alive,
whatever it answered.
+ public void recordResponse(ServerRoutingInstance serverRoutingInstance) {
+ ServerChannel serverChannel =
_serverToChannelMap.get(serverRoutingInstance);
+ if (serverChannel != null) {
+ serverChannel.recordResponse();
+ }
+ }
+
+ /// Records that a query written to the given channel at `submitTimeMs`
timed out unanswered. The verdict waits
+ /// out the grace period: the query only counts against the channel if by
then the server has still sent nothing
+ /// on it since the query was written. That lets the server's own late
timeout response through, so a burst of
+ /// slow queries, or a broker pause that delays reading the responses, does
not look like a dead channel: only a
+ /// channel the server has gone silent on does. At the threshold the channel
is marked suspect and the server is
+ /// reported through [QueryRouter#reportUnansweredServer]. No-op when the
check is disabled.
+ public void recordUnansweredQuery(ServerRoutingInstance
serverRoutingInstance, long submitTimeMs) {
Review Comment:
Nice design on the grace-period judging. One thought: on a server that has
genuinely died under high QPS, every timed-out query schedules a judge task on
the single liveness thread, so the executor's queue can briefly grow with the
inflight count until the first threshold trip flips the channel suspect. It
self-heals once the channel is replaced, so not blocking — just flagging in
case you want a cap or a dedup per channel.
##########
pinot-common/src/main/java/org/apache/pinot/common/config/NettyConfig.java:
##########
@@ -18,13 +18,56 @@
*/
package org.apache.pinot.common.config;
+import com.google.common.base.Preconditions;
import org.apache.pinot.spi.env.PinotConfiguration;
/// Container object for netty configuration of pinot clients and servers
(netty, grizzly, etc.)
+///
+/// Every setting is read as `<namespace>.<suffix>`, where the namespace is
`pinot.broker.netty` on the broker and
+/// `pinot.server.netty` on the server. The channel liveness settings below
only apply to the client side, i.e. the
+/// broker-to-server channels opened by `ServerChannels`; the server-side
`QueryServer` ignores them.
public class NettyConfig {
private static final String NATIVE_TRANSPORTS_ENABLED =
"native.transports.enabled";
+
+ /// Upper bound on how long a request write may stay pending before the
channel is closed, in milliseconds. A write
+ /// is pending while the kernel has not accepted the bytes, which on a
half-open or stalled socket is forever (the
+ /// peer stops acknowledging, the send buffer fills, and `isActive()` keeps
reporting `true` until the kernel gives
+ /// up on retransmits, typically after many minutes). Closing the channel
fails the in-flight queries on it fast
+ /// and lets the failure detector quarantine the server. Non-positive
disables the check. Broker key:
+ /// `pinot.broker.netty.write.timeout.ms`.
+ public static final String WRITE_TIMEOUT_MS = "write.timeout.ms";
+ public static final long DEFAULT_WRITE_TIMEOUT_MS = 10_000L;
Review Comment:
The 10s write timeout is a new default that will now close a channel whose
write doesn't reach the kernel in time. For a table with a very large segment
list (multi-MB request) on a slow-but-healthy link this could occasionally
fire. The default seems safe, but worth confirming it comfortably clears your
largest real request sizes, since a false close costs the in-flight queries on
that channel.
--
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]
