You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "xiangfu0 (via GitHub)" <gi...@apache.org> on 2026/10/23 22:09:01 UTC
[PR] Make the failure detector probe the channels that carry queries (DATA-3342) [pinot]
xiangfu0 opened a new pull request, #20022:
URL: https://github.com/apache/pinot/pull/20022
Stacked on #20021 (the first two commits are that PR; this PR is the last commit). Second of the DATA-3342 series; see #20021 for the incident write-up.
## Problem
Once a broker-to-server channel is half-open, the failure detector cannot get the server off it:
1. **The retry probe never saw the channel that carries the queries.** `QueryRouter#hasChannel` and `QueryRouter#connect` hard-code `TableType.OFFLINE`, and `ServerRoutingInstance#equals` includes the table type, so on a realtime-only cluster the single-stage retrier found no channel, answered `UNKNOWN`, and `BaseExponentialBackoffRetryFailureDetector` (which treats a round with no `UNHEALTHY` vote as recovered) readmitted the server onto whatever channel was cached. In the 2026-09-05 incident the health mark came 84 s after Helix readmission and changed nothing, because the probe had never looked at the REALTIME channel.
2. **The probe could not fail on a half-open channel anyway.** It only reconnects when the cached channel is `null` or `!isActive()`; a half-open socket reports active until the kernel gives up on retransmits.
3. **Timeouts never reached the detector.** `AsyncQueryResponse` only sets `_failedServer` on a send failure or a channel-inactive event. A query that timed out because the server never answered reported nothing, so a channel that swallowed 1.07M queries over 37 minutes (2026-10-09) never produced a single unhealthy mark.
## Fix
**Probe every channel the server was queried over.** `QueryRouter#hasChannel` answers whether the broker has sent a query to the server over any of its channels (OFFLINE or REALTIME, TLS or plaintext), and `QueryRouter#connect` re-establishes each such channel through the new `ServerChannels#probe`: a closed channel is reconnected as before, and a channel marked suspect is replaced with a fresh connection instead of being trusted on `isActive()`. Channels that only startup pre-connect opened are still left alone, so a multi-stage-only cluster keeps answering `UNKNOWN`.
**Report a server after consecutive unanswered timeouts.** `QueryRouter` keeps, per channel, the number of queries that timed out without the server answering anything since. Any response from the server resets it, even a late one for a query that already timed out, so a slow server that still answers some queries never reaches the threshold; a dead channel answers nothing. When a query times out, `AsyncQueryResponse` records every server the request reached but that never answered; the first one to cross the threshold becomes the failed server, the request handler reports it to the failure detector (which excludes it from routing), and its channel is marked suspect so the probe replaces it. New broker config key `pinot.broker.failure.detector.consecutive.timeouts.threshold`, default 5, non-positive disables. A timeout stays a timeout: no send exception is set, the query response is unchanged.
**All-UNKNOWN is recovered, deliberately.** The detector keeps treating a retry round where every retrier answers `UNKNOWN` as recovered, now with a log line saying so. After an eviction (readmission in #20021, or `DirectOOMHandler`) no transport holds a channel to the server, so there is nothing to verify; keeping it unhealthy would leave it out of routing for the rest of the backoff schedule (about 85 minutes with the defaults), and because `markServerUnhealthy` is `computeIfAbsent`, a later failure could not flag it again. With the probe fix, `UNKNOWN` is rare and honest.
### How the incidents play out with this change
- Realtime-only cluster, dead channel after a node flap: the first five unanswered queries report the server; it leaves routing; five seconds later the probe replaces the REALTIME channel with a fresh connection to the restarted process and the server is readmitted onto a working channel. Previously the probe answered `UNKNOWN` and readmitted it onto the dead one.
- Path stall on three connections: each stalled channel's server is reported after five unanswered queries (within one query timeout of the stall at that query rate), leaves routing, and is probed over a fresh connection, which takes a different path or fails loudly. Queries stop being routed into the stall within seconds instead of 37 minutes.
## Compatibility
No wire format or serialization change. `QueryRouter#hasChannel`/`connect` keep their signatures; the old five-argument `QueryRouter` constructor remains. The one behaviour change for the detector is that the single-stage retrier now votes on realtime-only clusters, which is the fix.
## Tests
- `QueryRoutingTest` against a real `QueryServer`:
- `testRetryProbeCoversRealtimeOnlyChannel`: after a REALTIME-only query `hasChannel` is true, the probe succeeds and keeps the healthy channel.
- `testConsecutiveUnansweredTimeoutsReportServerAndProbeReplacesChannel`: with a server that answers nothing (threshold 3, 300 ms timeout) the third timeout reports the REALTIME routing instance with no send exception, the count resets, and the probe closes the suspect channel and opens a fresh one that the next probe keeps.
- `testServerResponseResetsConsecutiveTimeouts`: a late answer to an already timed-out query resets the count, so the next timeouts start over.
- `AsyncQueryResponseTest`: a timeout reports the server the router flags and never a server the request did not reach; below the threshold nothing is reported; an answered query records no timeout.
- `ServerChannelsTest#testProbeReplacesSuspectChannelAndKeepsHealthyOne`: pre-connect-only channels are not probed; a healthy channel is kept, a suspect one replaced, a closed one reconnected, an unreachable server fails the probe and leaves no channel behind.
- `ConnectionFailureDetectorTest#testAllRetriersUnknownIsTreatedAsRecovered`: pins the all-UNKNOWN semantics and that the server can be flagged again afterwards.
Verified locally with the transport and broker routing suites touched by #20021 plus the above, and checkstyle and license checks on `pinot-spi`, `pinot-common`, `pinot-core`, `pinot-broker`.
## Follow-ups (DATA-3342)
- Cancel queued writes for queries that already timed out, and have the server drop requests whose broker deadline has passed (`InstanceRequestHandler` sets the arrival time at socket read), so a recovered channel does not execute a stale backlog.
- Evict channels on a `LiveInstance` session change, for pod restarts that leave the instance enabled in Helix.
--
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] Make the failure detector probe the channels that carry queries [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #20022:
URL: https://github.com/apache/pinot/pull/20022#discussion_r4369185086
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java:
##########
@@ -235,11 +245,32 @@ int getPartitionPruningPreparationThreshold() {
return _partitionPruningPreparationThreshold;
}
- /// Sets a callback to be invoked when a server becomes routable: when it starts, and when it is enabled again after
- /// it was disabled (for example while it restarted) or excluded. This is useful for resetting gRPC channel state to
- /// avoid exponential backoff delays and connections to a stale address.
- public void setServerEnabledCallback(Consumer<ServerInstance> callback) {
+ /// Registers a callback invoked with the [ServerInstance] of a server that becomes routable: when it starts, and when
+ /// it is enabled again after it was disabled (for example while it restarted) or excluded. Callers use it to drop
+ /// state that may be tied to a previous process of the server: gRPC reconnection backoff and connections to a stale
+ /// address, cached Netty channels that may be half-open. Callbacks run on the Helix callback thread while the routing
+ /// manager holds its write lock, so they must be quick and must not block.
+ public void addServerEnabledCallback(Consumer<ServerInstance> callback) {
+ _serverEnabledCallbacks.add(callback);
+ }
+
+ /// Sets a callback to be invoked when a server becomes routable, see [#addServerEnabledCallback(Consumer)]. Setting
+ /// it again replaces the callback set before through this method, and only that one, so callbacks registered
+ /// through `addServerEnabledCallback` are never dropped.
+ public synchronized void setServerEnabledCallback(Consumer<ServerInstance> callback) {
Review Comment:
Small asymmetry: `setServerEnabledCallback` is `synchronized` but `addServerEnabledCallback`/`addServerRemovedCallback` are not. The CopyOnWriteArrayList is thread-safe, but `set`'s remove-then-add isn't atomic against a concurrent `add`. In practice all registration happens during `start()` so this is benign — just flagging it in case callbacks ever get registered dynamically later.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/AsyncQueryResponse.java:
##########
@@ -116,6 +121,20 @@ public Map<ServerRoutingInstance, ServerResponse> getFinalResponses()
}
}
+ /// On a timeout, hands every server the request reached but that never answered to the transport, with the time
+ /// and the channel the request was written to; the transport judges after a grace period whether the server has gone
+ /// silent on that channel (see [ServerChannels#recordUnansweredQuery]).
+ private void recordUnansweredServers() {
+ for (Map.Entry<ServerRoutingInstance, ServerResponse> entry : _responseMap.entrySet()) {
+ ServerResponse response = entry.getValue();
+ int submitDelayMs = response.getSubmitDelayMs();
+ // Only servers the request was actually submitted to can be blamed for not answering it
+ if (response.getDataTable() == null && submitDelayMs >= 0) {
Review Comment:
When a query times out while still blocked acquiring the channel lock, `response.getChannel()` is null here, so the judgment counts against whatever channel the entry currently holds rather than the one this query targeted. That looks intentional (and is documented on `judgeUnansweredQuery`), just confirming it's the behavior you want for the stuck-on-lock case.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -182,7 +293,90 @@ public void preConnect(ServerRoutingInstance serverRoutingInstance, long timeout
_serverToChannelMap.computeIfAbsent(serverRoutingInstance, ServerChannel::new).preConnect(timeoutMs);
}
+ /// 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, when pings are off. 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 at `submitTimeMs` to `channel` (when known) timed out unanswered. The verdict waits
+ /// out the grace period: the query only counts if the entry still holds that channel and 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,
+ @Nullable Channel channel) {
+ if (_livenessExecutor == null) {
+ return;
+ }
+ ServerChannel serverChannel = _serverToChannelMap.get(serverRoutingInstance);
+ if (serverChannel == null) {
+ return;
+ }
+ if (serverChannel._numPendingJudgments.incrementAndGet() > _maxPendingJudgmentsPerChannel) {
Review Comment:
The cap is applied with an increment-then-decrement, so `_numPendingJudgments` can transiently read `cap+1` from another thread. It's purely a bound and self-corrects, so no real impact, but a `>= cap` pre-check or `updateAndGet` would avoid the brief overshoot if you want it exact.
--
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] Make the failure detector probe the channels that carry queries [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #20022:
URL: https://github.com/apache/pinot/pull/20022#discussion_r4368892259
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -264,17 +565,39 @@ 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();
+ resetLiveness();
Review Comment:
Fixed, and closed from both sides. The replacement now starts a clean liveness record (not suspect, nothing unanswered, last response set to now) under `_livenessLock` before it connects, and again once connected, so a judgment that runs while the connect is in progress sees the reset record instead of the stale one. On top of that, each timed-out query now carries the channel it was written to, and a judgment only counts a query whose channel is still the one the entry holds, so a query written to the dropped channel can never count against its successor, however the timing falls. `ServerChannelsTest#testJudgmentOfAQueryOnAReplacedChannelDoesNotCount` covers it with a threshold of 1, where a single counted judgment would mark the fresh channel suspect at once.
--
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] Make the failure detector probe the channels that carry queries [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #20022:
URL: https://github.com/apache/pinot/pull/20022#discussion_r4366307952
##########
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:
Fair, and the honest answer was that it depended on link speed: a 50 MB segment list at 50 Mbit/s takes 8 s to hand to the kernel, so a fixed completion timeout could fire on a healthy but slow link. The handler is now progress-based instead: every write goes down the pipeline with a progressive promise, Netty's outbound buffer advances it each time part of the message reaches the socket, and the timer re-arms from the last progress. A large request that keeps moving is never timed out, however long it takes in all; only a socket that accepts nothing for 10 s is closed, which is the half-open signature regardless of request size. `ServerChannelsTest#testSlowButProgressingWriteIsNotTimedOut` drives an 8 MB request through a peer draining about 3 MB/s with a 1 s timeout, so the write outlives the timeout by more than a second and must still complete with the channel open; the stall test still closes within the timeout. (One caveat worth knowing: the signal is bytes the kernel accept
ed, so a *receiver application* that reads very slowly can make its kernel batch window updates and look idle for stretches; a slow *network path* does not, since acknowledgements free the send buffer continuously. A Pinot server reads requests on its event loop, so the former only happens when that loop is stuck, which is a case we want closed.)
##########
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:
Addressed in #20021 (this branch picks it up on rebase): a disabled server whose instance config disappears is dropped from the map and reported through a new `addServerRemovedCallback`, which the single-stage handler uses to evict its channels, so the map is bounded by the servers currently disabled rather than by every id the broker has ever seen.
--
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] Make the failure detector probe the channels that carry queries [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #20022:
URL: https://github.com/apache/pinot/pull/20022#discussion_r4369189152
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java:
##########
@@ -235,11 +245,32 @@ int getPartitionPruningPreparationThreshold() {
return _partitionPruningPreparationThreshold;
}
- /// Sets a callback to be invoked when a server becomes routable: when it starts, and when it is enabled again after
- /// it was disabled (for example while it restarted) or excluded. This is useful for resetting gRPC channel state to
- /// avoid exponential backoff delays and connections to a stale address.
- public void setServerEnabledCallback(Consumer<ServerInstance> callback) {
+ /// Registers a callback invoked with the [ServerInstance] of a server that becomes routable: when it starts, and when
+ /// it is enabled again after it was disabled (for example while it restarted) or excluded. Callers use it to drop
+ /// state that may be tied to a previous process of the server: gRPC reconnection backoff and connections to a stale
+ /// address, cached Netty channels that may be half-open. Callbacks run on the Helix callback thread while the routing
+ /// manager holds its write lock, so they must be quick and must not block.
+ public void addServerEnabledCallback(Consumer<ServerInstance> callback) {
+ _serverEnabledCallbacks.add(callback);
+ }
+
+ /// Sets a callback to be invoked when a server becomes routable, see [#addServerEnabledCallback(Consumer)]. Setting
+ /// it again replaces the callback set before through this method, and only that one, so callbacks registered
+ /// through `addServerEnabledCallback` are never dropped.
+ public synchronized void setServerEnabledCallback(Consumer<ServerInstance> callback) {
Review Comment:
It is benign even with dynamic registration, which is why only the setter is synchronized. Each `CopyOnWriteArrayList` operation is atomic on its own, so an `add` that interleaves between the setter's `remove` and `add` simply ends up registered next to the setter's new callback: no callback is lost and none is left behind. What `synchronized` protects is the setter's own slot bookkeeping (`_serverEnabledCallback`) against two concurrent `set` calls, which is the only case where remove-then-add could leave the wrong callback in place. Registration order between a concurrent `add` and `set` is unspecified either way, and nothing depends on it.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -182,7 +293,90 @@ public void preConnect(ServerRoutingInstance serverRoutingInstance, long timeout
_serverToChannelMap.computeIfAbsent(serverRoutingInstance, ServerChannel::new).preConnect(timeoutMs);
}
+ /// 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, when pings are off. 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 at `submitTimeMs` to `channel` (when known) timed out unanswered. The verdict waits
+ /// out the grace period: the query only counts if the entry still holds that channel and 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,
+ @Nullable Channel channel) {
+ if (_livenessExecutor == null) {
+ return;
+ }
+ ServerChannel serverChannel = _serverToChannelMap.get(serverRoutingInstance);
+ if (serverChannel == null) {
+ return;
+ }
+ if (serverChannel._numPendingJudgments.incrementAndGet() > _maxPendingJudgmentsPerChannel) {
Review Comment:
Right, and I left it that way on purpose: the bound that matters is exact. A caller only schedules a judgment when its own increment came out at or below the cap, and a caller whose increment overshoots undoes it without scheduling, so the number of queued or running judgments per channel never exceeds the cap. Only a concurrent reader of the counter can see it briefly above, by at most the number of callers racing at that instant, and the counter is read only by the cap check itself and a test getter. A CAS loop would make that transient reading exact at the cost of retries under exactly the burst the cap exists for.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/AsyncQueryResponse.java:
##########
@@ -116,6 +121,20 @@ public Map<ServerRoutingInstance, ServerResponse> getFinalResponses()
}
}
+ /// On a timeout, hands every server the request reached but that never answered to the transport, with the time
+ /// and the channel the request was written to; the transport judges after a grace period whether the server has gone
+ /// silent on that channel (see [ServerChannels#recordUnansweredQuery]).
+ private void recordUnansweredServers() {
+ for (Map.Entry<ServerRoutingInstance, ServerResponse> entry : _responseMap.entrySet()) {
+ ServerResponse response = entry.getValue();
+ int submitDelayMs = response.getSubmitDelayMs();
+ // Only servers the request was actually submitted to can be blamed for not answering it
+ if (response.getDataTable() == null && submitDelayMs >= 0) {
Review Comment:
That case does not reach the judgment. `QueryRouter#submitQuery` writes every request synchronously before the caller starts waiting on the response, and a request is only marked submitted (`submitDelayMs >= 0`, the condition `recordUnansweredServers` checks) after `ServerChannels#sendRequest` returned, by which point `markRequestChannel` has recorded its channel. A query still blocked on the channel lock is therefore not submitted yet, and if the lock wait runs out the send throws, the query is marked FAILED rather than TIMED_OUT, and `recordUnansweredServers` never runs. So with the current router every judged query carries its channel; the time-only fallback for a null channel only covers callers that submit without recording one.
--
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] Make the failure detector probe the channels that carry queries [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #20022:
URL: https://github.com/apache/pinot/pull/20022#discussion_r4369050243
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/QueryRouter.java:
##########
@@ -177,18 +203,17 @@ public boolean hasChannel(ServerInstance serverInstance) {
return !channelRoutingInstances(serverInstance).isEmpty();
}
- /// Connects to the given server, returns `true` if the server is successfully connected.
- ///
- /// This is the reachability probe the failure detector uses when pings are off, opening the OFFLINE channel with a
- /// TCP connect only. Startup pre-connect uses [#preConnect] instead.
+ /// The reachability probe the failure detector uses when pings are off. Re-establishes every channel this broker
+ /// has queried the server over, one per table type: reconnects the ones that are down and replaces the ones marked
+ /// suspect (see [ServerChannels#recordUnansweredQuery]) with a fresh connection, so that a half-open socket that
+ /// still reports active is never vouched for. Returns `true` if every such channel is connected afterwards. It used
+ /// to open the OFFLINE channel only, so on a realtime-only cluster a handshake to a freshly restarted process passed
+ /// the probe while the channel that carried the queries was dead. Startup pre-connect uses [#preConnect] instead.
public boolean connect(ServerInstance serverInstance) {
try {
- if (_serverChannelsTls != null) {
- _serverChannelsTls.connect(
- serverInstance.toServerRoutingInstance(TableType.OFFLINE, ServerInstance.RoutingType.NETTY_TLS));
- } else {
- _serverChannels.connect(
- serverInstance.toServerRoutingInstance(TableType.OFFLINE, ServerInstance.RoutingType.NETTY));
+ ServerChannels serverChannels = channels();
+ for (ServerRoutingInstance routingInstance : channelRoutingInstances(serverInstance)) {
+ serverChannels.probe(routingInstance);
}
return true;
Review Comment:
Minor: when `channelRoutingInstances` is empty this returns `true` vacuously. It's safe today because the only caller (`retryUnhealthyServer`) guards with `hasChannel` and returns UNKNOWN first, but the javadoc says "true if every such channel is connected" — maybe note that no channels also yields true, so a future caller doesn't read it as "reachable".
##########
pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java:
##########
@@ -1389,6 +1389,21 @@ public enum Type {
/// stop-the-world GC pause expected on a server, or a pause will briefly take that server out of routing.
public static final String CONFIG_OF_PING_TIMEOUT_MS = "pinot.broker.failure.detector.ping.timeout.ms";
public static final long DEFAULT_PING_TIMEOUT_MS = 5_000L;
+
+ /// Single-stage queries a server leaves unanswered are also counted per channel, with or without pings. A query
+ /// counts once its timeout has passed and, a further grace period later, the server has still sent nothing on
+ /// the channel since the query was written to it. The grace lets the server's own late timeout response through,
+ /// so a burst of slow queries or a broker pause does not look like a dead channel; a half-open channel answers
+ /// nothing, ever. At the threshold the channel is marked suspect, so the next query to it (or the failure
+ /// detector's retry) replaces it with a fresh connection. With pings disabled, the server is also reported to
+ /// the failure detector then; with pings enabled, the ping that follows every timeout makes that call instead.
+ /// Any response from the server resets the count. A non-positive threshold disables the check.
+ public static final String CONFIG_OF_UNANSWERED_QUERIES_THRESHOLD =
+ "pinot.broker.failure.detector.unanswered.queries.threshold";
+ public static final int DEFAULT_UNANSWERED_QUERIES_THRESHOLD = 5;
Review Comment:
Confirming intent: threshold defaults to 5 (> 0), so this check is on by default. With pings off, a channel that leaves 5 graced timeouts unanswered now gets the server reported to the failure detector (out of routing until the retry probe). That's the feature, just flagging it's a default-on behavior change for existing ping-off brokers.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/QueryRouter.java:
##########
@@ -330,12 +380,38 @@ public boolean preConnect(ServerInstance serverInstance, TableType tableType, lo
}
}
+ /// Closes and forgets every cached channel to the given server: both table types, over TLS and plaintext. Returns
+ /// the number of channels evicted.
+ ///
+ /// Called through `BrokerRequestHandler#onServerEnabled` when the routing manager enables a server, which is how a
+ /// restart shows up through Helix, and through `BrokerRequestHandler#onServerRemoved` when a server leaves the
+ /// cluster. See [ServerChannels#evictChannel] for why the cached channel cannot be trusted at that point.
+ public int evictChannels(ServerInstance serverInstance) {
+ int numEvicted = 0;
+ for (TableType tableType : TableType.values()) {
+ if (serverInstance.getPort() > 0 && _serverChannels.evictChannel(
+ serverInstance.toServerRoutingInstance(tableType, ServerInstance.RoutingType.NETTY))) {
+ numEvicted++;
+ }
+ if (_serverChannelsTls != null && serverInstance.getNettyTlsPort() > 0 && _serverChannelsTls.evictChannel(
+ serverInstance.toServerRoutingInstance(tableType, ServerInstance.RoutingType.NETTY_TLS))) {
+ numEvicted++;
+ }
+ }
+ return numEvicted;
+ }
+
public void shutDown() {
_serverChannels.shutDown();
+ if (_serverChannelsTls != null) {
+ _serverChannelsTls.shutDown();
+ }
}
void receiveDataTable(ServerRoutingInstance serverRoutingInstance, DataTable dataTable, int responseSize,
int deserializationTimeMs) {
+ // Any response proves the server and its channel alive, whether or not the query it answers is still waiting
+ serverChannelsFor(serverRoutingInstance).recordResponse(serverRoutingInstance);
Review Comment:
Minor/heads-up: `recordResponse` takes `_livenessLock` on the event loop for every response now, where before this path was lock-free. The section is tiny and per-channel (single event-loop thread per channel), so contention is negligible, but worth being aware of on the hot response path.
--
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] Make the failure detector probe the channels that carry queries [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #20022:
URL: https://github.com/apache/pinot/pull/20022#discussion_r4369091554
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/QueryRouter.java:
##########
@@ -177,18 +203,17 @@ public boolean hasChannel(ServerInstance serverInstance) {
return !channelRoutingInstances(serverInstance).isEmpty();
}
- /// Connects to the given server, returns `true` if the server is successfully connected.
- ///
- /// This is the reachability probe the failure detector uses when pings are off, opening the OFFLINE channel with a
- /// TCP connect only. Startup pre-connect uses [#preConnect] instead.
+ /// The reachability probe the failure detector uses when pings are off. Re-establishes every channel this broker
+ /// has queried the server over, one per table type: reconnects the ones that are down and replaces the ones marked
+ /// suspect (see [ServerChannels#recordUnansweredQuery]) with a fresh connection, so that a half-open socket that
+ /// still reports active is never vouched for. Returns `true` if every such channel is connected afterwards. It used
+ /// to open the OFFLINE channel only, so on a realtime-only cluster a handshake to a freshly restarted process passed
+ /// the probe while the channel that carried the queries was dead. Startup pre-connect uses [#preConnect] instead.
public boolean connect(ServerInstance serverInstance) {
try {
- if (_serverChannelsTls != null) {
- _serverChannelsTls.connect(
- serverInstance.toServerRoutingInstance(TableType.OFFLINE, ServerInstance.RoutingType.NETTY_TLS));
- } else {
- _serverChannels.connect(
- serverInstance.toServerRoutingInstance(TableType.OFFLINE, ServerInstance.RoutingType.NETTY));
+ ServerChannels serverChannels = channels();
+ for (ServerRoutingInstance routingInstance : channelRoutingInstances(serverInstance)) {
+ serverChannels.probe(routingInstance);
}
return true;
Review Comment:
Good point; the Javadoc now says so (6778a0f850): `true` includes the vacuous case of a server with no queried channel, so a caller that needs "reachable" has to check `hasChannel` first, as `retryUnhealthyServer` does.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/QueryRouter.java:
##########
@@ -330,12 +380,38 @@ public boolean preConnect(ServerInstance serverInstance, TableType tableType, lo
}
}
+ /// Closes and forgets every cached channel to the given server: both table types, over TLS and plaintext. Returns
+ /// the number of channels evicted.
+ ///
+ /// Called through `BrokerRequestHandler#onServerEnabled` when the routing manager enables a server, which is how a
+ /// restart shows up through Helix, and through `BrokerRequestHandler#onServerRemoved` when a server leaves the
+ /// cluster. See [ServerChannels#evictChannel] for why the cached channel cannot be trusted at that point.
+ public int evictChannels(ServerInstance serverInstance) {
+ int numEvicted = 0;
+ for (TableType tableType : TableType.values()) {
+ if (serverInstance.getPort() > 0 && _serverChannels.evictChannel(
+ serverInstance.toServerRoutingInstance(tableType, ServerInstance.RoutingType.NETTY))) {
+ numEvicted++;
+ }
+ if (_serverChannelsTls != null && serverInstance.getNettyTlsPort() > 0 && _serverChannelsTls.evictChannel(
+ serverInstance.toServerRoutingInstance(tableType, ServerInstance.RoutingType.NETTY_TLS))) {
+ numEvicted++;
+ }
+ }
+ return numEvicted;
+ }
+
public void shutDown() {
_serverChannels.shutDown();
+ if (_serverChannelsTls != null) {
+ _serverChannelsTls.shutDown();
+ }
}
void receiveDataTable(ServerRoutingInstance serverRoutingInstance, DataTable dataTable, int responseSize,
int deserializationTimeMs) {
+ // Any response proves the server and its channel alive, whether or not the query it answers is still waiting
+ serverChannelsFor(serverRoutingInstance).recordResponse(serverRoutingInstance);
Review Comment:
Right, and I kept it deliberately. The lock is per channel entry, and the response side always runs on that channel's single event loop, so the only other takers are the connect path when a connection is replaced and the liveness thread's judgment, both rare. The critical section is two field writes. In exchange it is what makes a response that lands while a judgment runs impossible to lose, which the earlier unsynchronized version could do. An uncontended monitor on that path costs a few nanoseconds against a response that has just been deserialized.
##########
pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java:
##########
@@ -1389,6 +1389,21 @@ public enum Type {
/// stop-the-world GC pause expected on a server, or a pause will briefly take that server out of routing.
public static final String CONFIG_OF_PING_TIMEOUT_MS = "pinot.broker.failure.detector.ping.timeout.ms";
public static final long DEFAULT_PING_TIMEOUT_MS = 5_000L;
+
+ /// Single-stage queries a server leaves unanswered are also counted per channel, with or without pings. A query
+ /// counts once its timeout has passed and, a further grace period later, the server has still sent nothing on
+ /// the channel since the query was written to it. The grace lets the server's own late timeout response through,
+ /// so a burst of slow queries or a broker pause does not look like a dead channel; a half-open channel answers
+ /// nothing, ever. At the threshold the channel is marked suspect, so the next query to it (or the failure
+ /// detector's retry) replaces it with a fresh connection. With pings disabled, the server is also reported to
+ /// the failure detector then; with pings enabled, the ping that follows every timeout makes that call instead.
+ /// Any response from the server resets the count. A non-positive threshold disables the check.
+ public static final String CONFIG_OF_UNANSWERED_QUERIES_THRESHOLD =
+ "pinot.broker.failure.detector.unanswered.queries.threshold";
+ public static final int DEFAULT_UNANSWERED_QUERIES_THRESHOLD = 5;
Review Comment:
Confirmed, that is the intent, and it is now called out in the description's release note: on by default (threshold 5, grace 2 s); on brokers with pings off, a channel that leaves five graced timeouts unanswered gets the server reported to the failure detector, which with the `CONNECTION` detector takes it out of routing until the retry probe re-establishes the channel, and with the default `NO_OP` detector only replaces the channel. A threshold of 0 or less turns it off.
--
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] Make the failure detector probe the channels that carry queries (DATA-3342) [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #20022:
URL: https://github.com/apache/pinot/pull/20022#discussion_r4364963128
##########
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;
+
+ /// Linux `TCP_USER_TIMEOUT`, in milliseconds: how long transmitted data may remain unacknowledged before the
+ /// kernel closes the connection, replacing the default retransmission timeout of several minutes. Covers the
+ /// case the write timeout cannot see: small writes that the kernel accepted into a send buffer that the peer never
+ /// drains. Only applied when the broker uses the Epoll transport (`native.transports.enabled` on Linux); NIO and
+ /// KQueue have no such socket option. Non-positive leaves the kernel default. Broker key:
+ /// `pinot.broker.netty.tcp.user.timeout.ms`.
+ public static final String TCP_USER_TIMEOUT_MS = "tcp.user.timeout.ms";
+ public static final int DEFAULT_TCP_USER_TIMEOUT_MS = 10_000;
Review Comment:
Minor/process, non-blocking: this and the write timeout and the write-buffer watermark rejection are all new and default ON for every broker-to-server channel (on master these channels had none of them). The defaults look sound and each is configurable with non-positive to disable, so no change requested — but it's worth an explicit release note and a rolling-upgrade mention, since on a high-RTT or lossy broker-to-server link a 10s TCP_USER_TIMEOUT could close connections a prior deployment tolerated, and operators there may want to raise or disable it.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -227,35 +418,99 @@ protected void initChannel(SocketChannel ch) {
_serverRoutingInstance));
}
});
+ if (isTcpUserTimeoutApplied()) {
+ _bootstrap.option(EpollChannelOption.TCP_USER_TIMEOUT, _tcpUserTimeoutMs);
+ }
}
void closeChannel() {
- if (_channel != null) {
- _channel.close();
+ Channel channel = _channel;
+ if (channel != null) {
+ channel.close();
}
}
+ /// See [ServerChannels#evictChannel]. Flags the entry first, then closes whatever channel it holds; no lock is
+ /// taken so that the routing manager's Helix callback is never blocked behind a connect or write in progress.
+ void evict() {
+ _evicted = true;
+ Channel channel = _channel;
+ if (channel != null) {
+ closeLocally(channel);
+ }
+ }
+
+ /// Closes a channel the broker decided to drop (eviction after a restart, replacement of a suspect channel).
+ /// The queries still in flight on it fail through `channelInactive` as on any closed channel, but the server is
+ /// not reported for it: the close was the broker's decision, not evidence against the server.
+ private void closeLocally(Channel channel) {
+ DataTableHandler dataTableHandler = channel.pipeline().get(DataTableHandler.class);
+ if (dataTableHandler != null) {
+ dataTableHandler.markClosedByBroker();
+ }
+ channel.close();
+ }
+
+ void recordResponse() {
+ _lastResponseTimeMs = System.currentTimeMillis();
+ _unansweredQueries.set(0);
+ }
+
+ /// Runs on the liveness thread a grace period after a query written at `submitTimeMs` timed out. See
+ /// [ServerChannels#recordUnansweredQuery].
+ void judgeUnansweredQuery(long submitTimeMs) {
+ // Answered something since the query was written (a fresh channel counts as answered from its connect time, so
+ // queries written to the channel it replaced do not count against it), or no longer in use. Strictly after:
+ // the connect and the first write can share a millisecond, while a real answer never shares one with its
+ // request.
+ if (_evicted || _lastResponseTimeMs > submitTimeMs) {
+ return;
+ }
+ int unansweredQueries = _unansweredQueries.incrementAndGet();
+ if (unansweredQueries < _unansweredQueriesThreshold) {
+ return;
+ }
+ _unansweredQueries.set(0);
+ _suspect = true;
Review Comment:
Minor, non-blocking: there's a narrow race here. The `_lastResponseTimeMs > submitTimeMs` guard is read once at the top of the method, but `recordResponse()` (event loop) and `connectWithoutLocking()` (query thread) both reset `_unansweredQueries` and bump `_lastResponseTimeMs` without ordering against this task. If a genuine response lands between the threshold check and this `_suspect = true`, the channel can be marked suspect and the server reported even though it just proved itself alive — a spurious reconnect plus one spurious failure-detector report.
The impact is bounded and self-correcting (the fresh channel is healthy, the next probe keeps it), so this is a MINOR. A cheap fix would be to re-read `_lastResponseTimeMs` right before setting `_suspect` and bail if it now exceeds `submitTimeMs`.
--
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] Make the failure detector probe the channels that carry queries [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #20022:
URL: https://github.com/apache/pinot/pull/20022#discussion_r4366342562
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -264,17 +565,39 @@ 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();
+ resetLiveness();
Review Comment:
Small window here: `_suspect = false` on line 596 is cleared before the (potentially slow) `connect().sync()`, and `resetLiveness()` only bumps `_lastResponseTimeMs` afterward. A stale `judgeUnansweredQuery` for an old query landing during the connect could see `!_suspect` + an un-reset `_lastResponseTimeMs <= submitTimeMs` and re-mark the brand-new channel suspect (and even report the server). It self-corrects on the next send, but if it crosses the threshold it costs a spurious exclusion. Worth clearing `_suspect` under `_livenessLock` together with the reset?
--
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] Make the failure detector probe the channels that carry queries [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #20022:
URL: https://github.com/apache/pinot/pull/20022#discussion_r4366307520
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -227,35 +418,99 @@ protected void initChannel(SocketChannel ch) {
_serverRoutingInstance));
}
});
+ if (isTcpUserTimeoutApplied()) {
+ _bootstrap.option(EpollChannelOption.TCP_USER_TIMEOUT, _tcpUserTimeoutMs);
+ }
}
void closeChannel() {
- if (_channel != null) {
- _channel.close();
+ Channel channel = _channel;
+ if (channel != null) {
+ channel.close();
}
}
+ /// See [ServerChannels#evictChannel]. Flags the entry first, then closes whatever channel it holds; no lock is
+ /// taken so that the routing manager's Helix callback is never blocked behind a connect or write in progress.
+ void evict() {
+ _evicted = true;
+ Channel channel = _channel;
+ if (channel != null) {
+ closeLocally(channel);
+ }
+ }
+
+ /// Closes a channel the broker decided to drop (eviction after a restart, replacement of a suspect channel).
+ /// The queries still in flight on it fail through `channelInactive` as on any closed channel, but the server is
+ /// not reported for it: the close was the broker's decision, not evidence against the server.
+ private void closeLocally(Channel channel) {
+ DataTableHandler dataTableHandler = channel.pipeline().get(DataTableHandler.class);
+ if (dataTableHandler != null) {
+ dataTableHandler.markClosedByBroker();
+ }
+ channel.close();
+ }
+
+ void recordResponse() {
+ _lastResponseTimeMs = System.currentTimeMillis();
+ _unansweredQueries.set(0);
+ }
+
+ /// Runs on the liveness thread a grace period after a query written at `submitTimeMs` timed out. See
+ /// [ServerChannels#recordUnansweredQuery].
+ void judgeUnansweredQuery(long submitTimeMs) {
+ // Answered something since the query was written (a fresh channel counts as answered from its connect time, so
+ // queries written to the channel it replaced do not count against it), or no longer in use. Strictly after:
+ // the connect and the first write can share a millisecond, while a real answer never shares one with its
+ // request.
+ if (_evicted || _lastResponseTimeMs > submitTimeMs) {
+ return;
+ }
+ int unansweredQueries = _unansweredQueries.incrementAndGet();
+ if (unansweredQueries < _unansweredQueriesThreshold) {
+ return;
+ }
+ _unansweredQueries.set(0);
+ _suspect = true;
Review Comment:
Fixed, and a little more than the re-read: the liveness bookkeeping (`_lastResponseTimeMs`, the unanswered count, the pending judgment) now sits behind a small lock on the channel entry, taken briefly by `recordResponse` on the event loop, by the connect path when it resets a fresh channel, and by the judgment on the liveness thread. The judgment re-checks the last response time and flips `_suspect` under that lock, so a response landing while it runs is never lost: either it is seen and the judgment bails, or it runs after and zeroes the count. The lock is uncontended in practice (one short critical section per response).
##########
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;
+
+ /// Linux `TCP_USER_TIMEOUT`, in milliseconds: how long transmitted data may remain unacknowledged before the
+ /// kernel closes the connection, replacing the default retransmission timeout of several minutes. Covers the
+ /// case the write timeout cannot see: small writes that the kernel accepted into a send buffer that the peer never
+ /// drains. Only applied when the broker uses the Epoll transport (`native.transports.enabled` on Linux); NIO and
+ /// KQueue have no such socket option. Non-positive leaves the kernel default. Broker key:
+ /// `pinot.broker.netty.tcp.user.timeout.ms`.
+ public static final String TCP_USER_TIMEOUT_MS = "tcp.user.timeout.ms";
+ public static final int DEFAULT_TCP_USER_TIMEOUT_MS = 10_000;
Review Comment:
Agreed that this deserves to be explicit, so #20021 now carries a "Release note" section covering the three new defaults, what each does, when it fires, and how to raise or disable it, plus the rolling-upgrade note (broker-only, no server or wire change, so brokers can be upgraded in any order and rolled back independently).
Why on by default, briefly: all three only act on a socket whose peer has stopped taking bytes. The write timeout is now on progress rather than completion (see the thread above): it fires only when the kernel has accepted nothing for 10 s, which on a healthy link needs a receiver that stopped reading, not a slow one. `TCP_USER_TIMEOUT` (Epoll only) fires only when transmitted data has gone unacknowledged for 10 s; on a lossy link Linux retransmits well within that, and a path that cannot deliver an ACK in 10 s has already failed every query on it at the default 10 s broker timeout. The watermark rejection needs 16 MB queued beyond what the kernel accepted, which a healthy peer drains in milliseconds. The failure each prevents is silent and long (minutes of lost queries, direct-memory exhaustion that takes healthy channels down with it), while a false positive costs one query and a reconnect, so the defaults err towards acting. Operators on unusual links can raise each one, or set
it non-positive to disable it, as the release note says.
##########
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:
Good call. The judgments queued per channel are now capped at `max(16, 2 × threshold)`: a timed-out query beyond that is simply not scheduled, because the ones already queued are enough to cross the threshold on their own, so the outcome under a dead server at high QPS is unchanged while the liveness queue stays at a few dozen tasks per channel instead of the inflight count. Each judgment still gets its own full grace period, which is what keeps the burst case honest, and a channel already marked suspect is not reported again until it has been replaced, so one burst means one report. Covered by `ServerChannelsTest#testPendingJudgmentsPerChannelAreCapped`.
##########
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:
Right, and I kept it that way deliberately. The lazy path has always paid the connect under `_channelLock` (first query, and every reconnect after a channel went inactive), and a suspect replacement is the same reconnect taken a little earlier. Concurrent senders wait at most their own `tryLock(timeoutMs)`, after which they fail with the channel-lock timeout rather than hanging, and since the server has usually been reported and taken out of routing by then there are few of them. Moving the connect out from under the lock would need a second connection-establishment path with its own races for a case that lasts one round trip; not worth it in this PR. Bounding `CONNECT_TIMEOUT_MILLIS` on the lazy path (Netty's 30 s default today) would shorten the worst case for everyone and is a reasonable small follow-up.
--
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] Make the failure detector probe the channels that carry queries (DATA-3342) [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 closed pull request #20022: Make the failure detector probe the channels that carry queries (DATA-3342)
URL: https://github.com/apache/pinot/pull/20022
--
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] Make the failure detector probe the channels that carry queries (DATA-3342) [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #20022:
URL: https://github.com/apache/pinot/pull/20022#issuecomment-6377067540
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/20022?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 `84.58904%` with `45 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 58.97%. Comparing base ([`910f9d5`](https://app.codecov.io/gh/apache/pinot/commit/910f9d5e4942010621c7d791557e0b4c4ffddcfd?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`36395c4`](https://app.codecov.io/gh/apache/pinot/commit/36395c4a46b01305bccea21eefe1b7bef9bd91fc?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 146 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pinot/pull/20022?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...rg/apache/pinot/core/transport/ServerChannels.java](https://app.codecov.io/gh/apache/pinot/pull/20022?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==) | 84.93% | [11 Missing and 11 partials :warning: ](https://app.codecov.io/gh/apache/pinot/pull/20022?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...a/org/apache/pinot/core/transport/QueryRouter.java](https://app.codecov.io/gh/apache/pinot/pull/20022?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==) | 65.30% | [10 Missing and 7 partials :warning: ](https://app.codecov.io/gh/apache/pinot/pull/20022?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...pache/pinot/core/transport/AsyncQueryResponse.java](https://app.codecov.io/gh/apache/pinot/pull/20022?src=pr&el=tree&filepath=pinot-core%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcore%2Ftransport%2FAsyncQueryResponse.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS90cmFuc3BvcnQvQXN5bmNRdWVyeVJlc3BvbnNlLmphdmE=) | 90.00% | [1 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/pinot/pull/20022?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...not/core/transport/RequestWriteTimeoutHandler.java](https://app.codecov.io/gh/apache/pinot/pull/20022?src=pr&el=tree&filepath=pinot-core%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcore%2Ftransport%2FRequestWriteTimeoutHandler.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS90cmFuc3BvcnQvUmVxdWVzdFdyaXRlVGltZW91dEhhbmRsZXIuamF2YQ==) | 82.35% | [1 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/pinot/pull/20022?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 (910f9d5) and HEAD (36395c4). Click for more details.
>
> <details><summary>HEAD has 24 uploads less than BASE</summary>
>
>| Flag | BASE (910f9d5) | HEAD (36395c4) |
>|------|------|------|
>|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 #20022 +/- ##
============================================
- Coverage 68.63% 58.97% -9.67%
+ Complexity 1486 1 -1485
============================================
Files 3526 2734 -792
Lines 230143 172831 -57312
Branches 36562 28540 -8022
============================================
- Hits 157963 101929 -56034
- Misses 59837 62256 +2419
+ Partials 12343 8646 -3697
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/20022/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/20022/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/20022/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/20022/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/20022/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `58.97% <84.58%> (-9.67%)` | :arrow_down: |
| [lane-a](https://app.codecov.io/gh/apache/pinot/pull/20022/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/20022/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/20022/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `58.97% <84.58%> (-9.67%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/20022/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `58.97% <84.58%> (-9.66%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/20022/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `58.97% <84.58%> (+0.63%)` | :arrow_up: |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/20022/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/20022?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] Make the failure detector probe the channels that carry queries [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
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: 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