This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 6d97d3226a7 Make the failure detector probe the channels that carry
queries (#20022)
6d97d3226a7 is described below
commit 6d97d3226a73d9d43ba3b313da8dc11427861e3c
Author: Xiang Fu <[email protected]>
AuthorDate: Sun Nov 8 01:10:25 2026 -0800
Make the failure detector probe the channels that carry queries (#20022)
With pings off (the default), the failure detector's retry probe opened the
OFFLINE channel only and trusted
isActive(), so a server could be readmitted onto a half-open REALTIME
channel, and a channel that swallowed every
query was never replaced.
- QueryRouter#connect probes every channel the server was queried over
(ServerChannels#probe): a closed channel is
reconnected and a suspect one is replaced instead of being trusted on
isActive(). Channels opened only by startup
pre-connect are left alone.
- A timed-out query hands each server it reached but that never answered to
the transport, which judges it
pinot.broker.failure.detector.unanswered.query.grace.ms (default 2000)
later, per channel: it counts only if the
server sent nothing on that channel since the request was written. Any
response frame resets the count.
- At pinot.broker.failure.detector.unanswered.queries.threshold consecutive
unanswered queries (default 5,
non-positive disables) the channel is marked suspect and the next send
replaces it with a fresh connection. With
pings off and
pinot.broker.failure.detector.report.unanswered.servers=true (default false),
the server is also
reported to the failure detector.
- An unwritable-channel rejection does not mark the channel suspect.
- A retry round where every retrier answers UNKNOWN still counts as
recovered, now pinned by a test.
- Add the NETTY_CONNECTION_SUSPECT_CHANNELS meter.
No wire or serialization change.
---
.../SingleConnectionBrokerRequestHandler.java | 18 +-
...ionBrokerRequestHandlerFailureDetectorTest.java | 37 +++-
.../apache/pinot/common/metrics/BrokerMeter.java | 6 +
.../ConnectionFailureDetectorTest.java | 19 ++
.../pinot/core/transport/AsyncQueryResponse.java | 20 +-
.../pinot/core/transport/DataTableHandler.java | 2 +
.../apache/pinot/core/transport/QueryRouter.java | 83 ++++++-
.../pinot/core/transport/ServerChannels.java | 242 +++++++++++++++++++--
.../core/transport/AsyncQueryResponseTest.java | 45 ++++
.../pinot/core/transport/QueryRoutingTest.java | 192 ++++++++++++++++
.../pinot/core/transport/ServerChannelsTest.java | 239 +++++++++++++++++++-
.../apache/pinot/spi/utils/CommonConstants.java | 27 +++
12 files changed, 891 insertions(+), 39 deletions(-)
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/SingleConnectionBrokerRequestHandler.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/SingleConnectionBrokerRequestHandler.java
index 364751d74d4..68d9af07e9f 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/SingleConnectionBrokerRequestHandler.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/SingleConnectionBrokerRequestHandler.java
@@ -140,7 +140,11 @@ public class SingleConnectionBrokerRequestHandler extends
BaseSingleStageBrokerR
super(config, brokerId, requestIdGenerator, routingManager,
accessControlFactory, queryQuotaManager, tableCache,
threadAccountant, multiClusterRoutingContext, materializedViewHandler);
_brokerReduceService = new BrokerReduceService(_config);
- _queryRouter = new QueryRouter(_brokerId, nettyConfig, tlsConfig,
serverRoutingStatsManager, threadAccountant);
+ _queryRouter = new QueryRouter(_brokerId, nettyConfig, tlsConfig,
serverRoutingStatsManager, threadAccountant,
+
_config.getProperty(CommonConstants.Broker.FailureDetector.CONFIG_OF_UNANSWERED_QUERIES_THRESHOLD,
+
CommonConstants.Broker.FailureDetector.DEFAULT_UNANSWERED_QUERIES_THRESHOLD),
+
_config.getProperty(CommonConstants.Broker.FailureDetector.CONFIG_OF_UNANSWERED_QUERY_GRACE_MS,
+
CommonConstants.Broker.FailureDetector.DEFAULT_UNANSWERED_QUERY_GRACE_MS));
_failureDetector = failureDetector;
_pingOnTimeout =
_config.getProperty(CommonConstants.Broker.FailureDetector.CONFIG_OF_ENABLE_PING_ON_TIMEOUT,
CommonConstants.Broker.FailureDetector.DEFAULT_ENABLE_PING_ON_TIMEOUT);
@@ -151,6 +155,14 @@ public class SingleConnectionBrokerRequestHandler extends
BaseSingleStageBrokerR
Preconditions.checkArgument(_pingTimeoutMs > 0, "'%s' must be positive,
got: %s",
CommonConstants.Broker.FailureDetector.CONFIG_OF_PING_TIMEOUT_MS,
_pingTimeoutMs);
_failureDetector.registerServerNotRespondedChecker(this::checkServerNotResponded);
+ } else if
(_config.getProperty(CommonConstants.Broker.FailureDetector.CONFIG_OF_REPORT_UNANSWERED_SERVERS,
+
CommonConstants.Broker.FailureDetector.DEFAULT_REPORT_UNANSWERED_SERVERS)) {
+ // Opted in: without pings nothing else acts on a timeout, so a server
that has gone silent on one of its channels
+ // is reported like a server the transport saw go down, and leaves
routing until the retry probe re-establishes
+ // the channel. Otherwise, and always with pings (the ping that follows
every timeout makes that call), the
+ // silent channel is only replaced.
+ _queryRouter.setUnansweredServerReporter(serverRoutingInstance ->
_failureDetector.markServerUnhealthy(
+ serverRoutingInstance.getInstanceId(),
serverRoutingInstance.getHostname()));
}
}
@@ -1070,7 +1082,7 @@ public class SingleConnectionBrokerRequestHandler extends
BaseSingleStageBrokerR
}
/// Check if a server that was previously detected as unhealthy is now
healthy: by pinging it when pings are enabled,
- /// otherwise by connecting to it.
+ /// otherwise by re-establishing every channel it was queried over (see
[QueryRouter#connect]).
public FailureDetector.ServerState retryUnhealthyServer(String instanceId) {
LOGGER.info("Retrying unhealthy server: {}", instanceId);
ServerInstance serverInstance =
_routingManager.getEnabledServerInstanceMap().get(instanceId);
@@ -1100,7 +1112,7 @@ public class SingleConnectionBrokerRequestHandler extends
BaseSingleStageBrokerR
}
if (_queryRouter.connect(serverInstance)) {
- LOGGER.info("Successfully connect to server: {}, marking it healthy",
instanceId);
+ LOGGER.info("Re-established the channel(s) to server: {}, marking it
healthy", instanceId);
return FailureDetector.ServerState.HEALTHY;
} else {
LOGGER.warn("Still cannot connect to server: {}, retry later",
instanceId);
diff --git
a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/SingleConnectionBrokerRequestHandlerFailureDetectorTest.java
b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/SingleConnectionBrokerRequestHandlerFailureDetectorTest.java
index 3ff080b1b3e..28e94e36e98 100644
---
a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/SingleConnectionBrokerRequestHandlerFailureDetectorTest.java
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/SingleConnectionBrokerRequestHandlerFailureDetectorTest.java
@@ -114,10 +114,15 @@ public class
SingleConnectionBrokerRequestHandlerFailureDetectorTest {
/// are recorded, mirroring the wiring `BaseBrokerStarter` performs in
production. Without `pingOnTimeout` the
/// default applies.
private void startBroker(boolean pingOnTimeout, long retryInitialDelayMs) {
+ startBroker(pingOnTimeout, retryInitialDelayMs, Map.of());
+ }
+
+ /// As above, with extra broker settings on top.
+ private void startBroker(boolean pingOnTimeout, long retryInitialDelayMs,
Map<String, Object> extraConfig) {
// BrokerMetrics is a JVM-wide singleton, so the gauge carries over from
whatever ran before in this fork.
BrokerMetrics.get().setValueOfGlobalGauge(BrokerGauge.UNHEALTHY_SERVERS,
0);
- PinotConfiguration config = new PinotConfiguration();
+ PinotConfiguration config = new PinotConfiguration(new
HashMap<>(extraConfig));
config.setProperty(Broker.FailureDetector.CONFIG_OF_TYPE,
Broker.FailureDetector.Type.CONNECTION.name());
if (pingOnTimeout) {
config.setProperty(Broker.FailureDetector.CONFIG_OF_ENABLE_PING_ON_TIMEOUT,
true);
@@ -380,6 +385,36 @@ public class
SingleConnectionBrokerRequestHandlerFailureDetectorTest {
assertEquals(_excludedServers, List.of());
}
+ /// With pings off, a server that leaves the threshold of queries unanswered
on its channel is reported to the failure
+ /// detector only when the operator opted in. By default only its channel is
replaced, so a healthy server whose
+ /// heavy queries overrun their timeout is never taken out of routing for it.
+ @Test(dataProvider = "reportUnanswered")
+ public void testUnansweredServerIsReportedOnlyWhenEnabled(boolean report)
+ throws Exception {
+ stopBroker();
+ startBroker(false, RETRIES_OFF_MS, Map.of(
+ Broker.FailureDetector.CONFIG_OF_UNANSWERED_QUERIES_THRESHOLD, 1,
+ Broker.FailureDetector.CONFIG_OF_UNANSWERED_QUERY_GRACE_MS, 100L,
+ Broker.FailureDetector.CONFIG_OF_REPORT_UNANSWERED_SERVERS, report));
+ ServerInstance frozenServer = startFrozenServer();
+
+ assertTrue(scatter(offlineRouteTo(frozenServer)).isTimedOut());
+ if (report) {
+ awaitUnhealthyServers(frozenServer);
+ assertEquals(_excludedServers, List.of(frozenServer.getInstanceId()));
+ } else {
+ // The judgment runs a grace period after the timeout; give it far
longer than that before concluding none ran
+ Thread.sleep(1_000L);
+ assertEquals(_failureDetector.getUnhealthyServers(), Set.of());
+ assertEquals(_excludedServers, List.of());
+ }
+ }
+
+ @DataProvider(name = "reportUnanswered")
+ public Object[][] reportUnanswered() {
+ return new Object[][]{{false}, {true}};
+ }
+
/// With pings off, retries connect to the server, as before this change,
rather than ping it. A connect succeeds
/// even against a frozen server, whose kernel still completes the
handshake, so this one is put back at once.
@Test
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMeter.java
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMeter.java
index b037d73321e..cfd1a6b5cbe 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMeter.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMeter.java
@@ -247,6 +247,12 @@ public class BrokerMeter implements AbstractMetrics.Meter {
// cluster or moved to another address. Entries whose channel had already
closed are dropped without being counted.
public static final BrokerMeter NETTY_CONNECTION_CHANNEL_EVICTIONS = create(
"NETTY_CONNECTION_CHANNEL_EVICTIONS", "nettyConnection", false);
+ // Channels marked suspect because the server left the configured number of
consecutive queries unanswered on them
+ // (see
CommonConstants.Broker.FailureDetector.CONFIG_OF_UNANSWERED_QUERIES_THRESHOLD);
the next query replaces each.
+ // With CONFIG_OF_REPORT_UNANSWERED_SERVERS on and pings off, each is also a
report to the failure detector. Tagged
+ // with the server's short name.
+ public static final BrokerMeter NETTY_CONNECTION_SUSPECT_CHANNELS = create(
+ "NETTY_CONNECTION_SUSPECT_CHANNELS", "nettyConnection", false);
public static final BrokerMeter PROACTIVE_CLUSTER_CHANGE_CHECK = create(
"PROACTIVE_CLUSTER_CHANGE_CHECK", "proactiveClusterChangeCheck", true);
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/failuredetector/ConnectionFailureDetectorTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/failuredetector/ConnectionFailureDetectorTest.java
index 45ab8ea1212..0fe95504605 100644
---
a/pinot-common/src/test/java/org/apache/pinot/common/failuredetector/ConnectionFailureDetectorTest.java
+++
b/pinot-common/src/test/java/org/apache/pinot/common/failuredetector/ConnectionFailureDetectorTest.java
@@ -453,6 +453,25 @@ public class ConnectionFailureDetectorTest {
}
}
+ /// When no retrier can vouch for or against a server (every transport
answers UNKNOWN, e.g. because the server's
+ /// channels were evicted), the server is let go rather than kept unhealthy
for the whole backoff schedule, so
+ /// that a later failure can flag it again. Pins the semantics the retriers
rely on.
+ @Test
+ public void testAllRetriersUnknownIsTreatedAsRecovered() {
+ _failureDetector.registerUnhealthyServerRetrier(instanceId ->
FailureDetector.ServerState.UNKNOWN);
+ _failureDetector.registerUnhealthyServerRetrier(instanceId ->
FailureDetector.ServerState.UNKNOWN);
+
+ _failureDetector.markServerUnhealthy(INSTANCE_ID, HOST_NAME);
+ verify(Set.of(INSTANCE_ID), 1, 0);
+
+ TestUtils.waitForCondition(aVoid ->
_failureDetector.getUnhealthyServers().isEmpty()
+ && _healthyServerNotifier._notifyHealthyServerCalled == 1, 5_000L,
"Server was not let go on all-UNKNOWN");
+
+ // ... and can be flagged again
+ _failureDetector.markServerUnhealthy(INSTANCE_ID, HOST_NAME);
+ verify(Set.of(INSTANCE_ID), 2, 1);
+ }
+
private void verify(Set<String> expectedUnhealthyServers, int
expectedNotifyUnhealthyServerCalled,
int expectedNotifyHealthyServerCalled) {
assertEquals(_failureDetector.getUnhealthyServers(),
expectedUnhealthyServers);
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/transport/AsyncQueryResponse.java
b/pinot-core/src/main/java/org/apache/pinot/core/transport/AsyncQueryResponse.java
index cb88d8e146d..ce74674e9b2 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/transport/AsyncQueryResponse.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/transport/AsyncQueryResponse.java
@@ -43,6 +43,7 @@ public class AsyncQueryResponse implements QueryResponse {
private final AtomicInteger _numServersResponded = new AtomicInteger();
private final ConcurrentHashMap<ServerRoutingInstance, ServerResponse>
_responseMap;
private final CountDownLatch _countDownLatch;
+ private final long _startTimeMs;
private final long _maxEndTimeMs;
private final long _timeoutMs;
private final ServerRoutingStatsManager _serverRoutingStatsManager;
@@ -72,6 +73,7 @@ public class AsyncQueryResponse implements QueryResponse {
}
_countDownLatch = new CountDownLatch(numServersQueried);
_timeoutMs = timeoutMs;
+ _startTimeMs = startTimeMs;
_maxEndTimeMs = startTimeMs + timeoutMs;
}
@@ -95,7 +97,9 @@ public class AsyncQueryResponse implements QueryResponse {
throws InterruptedException {
try {
boolean finish = _countDownLatch.await(_maxEndTimeMs -
System.currentTimeMillis(), TimeUnit.MILLISECONDS);
- _status.compareAndSet(Status.IN_PROGRESS, finish ? Status.COMPLETED :
Status.TIMED_OUT);
+ if (_status.compareAndSet(Status.IN_PROGRESS, finish ? Status.COMPLETED
: Status.TIMED_OUT) && !finish) {
+ recordUnansweredServers();
+ }
return _responseMap;
} finally {
// Update ServerRoutingStats for query completion. This is done here to
ensure that the stats are updated for
@@ -117,6 +121,20 @@ public class AsyncQueryResponse implements QueryResponse {
}
}
+ /// 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) {
+ _queryRouter.recordServerTimeout(entry.getKey(), _startTimeMs +
submitDelayMs, response.getChannel());
+ }
+ }
+ }
+
private boolean hasServerReturnedExceptions(ServerResponse response) {
if (response.getDataTable() != null &&
response.getDataTable().getExceptions().size() > 0) {
DataTable dataTable = response.getDataTable();
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/transport/DataTableHandler.java
b/pinot-core/src/main/java/org/apache/pinot/core/transport/DataTableHandler.java
index 8c97c99d22e..da104bb62b0 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/transport/DataTableHandler.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/transport/DataTableHandler.java
@@ -85,6 +85,8 @@ public class DataTableHandler extends
SimpleChannelInboundHandler<ByteBuf> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
+ // Before decoding: a frame proves the server alive even if it then fails
to decode
+ _queryRouter.recordResponse(_serverRoutingInstance);
int responseSize = msg.readableBytes();
_brokerMetrics.addMeteredGlobalValue(BrokerMeter.NETTY_CONNECTION_BYTES_RECEIVED,
responseSize);
try {
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/transport/QueryRouter.java
b/pinot-core/src/main/java/org/apache/pinot/core/transport/QueryRouter.java
index 381def8c5e2..ae719a1629f 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/transport/QueryRouter.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/transport/QueryRouter.java
@@ -28,6 +28,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiPredicate;
+import java.util.function.Consumer;
import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;
import org.apache.pinot.common.config.NettyConfig;
@@ -46,6 +47,7 @@ import
org.apache.pinot.core.transport.server.routing.stats.ServerRoutingStatsMa
import org.apache.pinot.spi.accounting.ThreadAccountant;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.apache.pinot.sql.parsers.CalciteSqlCompiler;
import org.slf4j.Logger;
@@ -76,6 +78,11 @@ public class QueryRouter {
private final BrokerMetrics _brokerMetrics = BrokerMetrics.get();
private final ConcurrentHashMap<Long, AsyncQueryResponse>
_asyncQueryResponseMap = new ConcurrentHashMap<>();
+ // Told about a server that stopped answering queries on one of its channels
(see ServerChannels). The request
+ // handler points it at the failure detector when pings are off and
reporting is enabled; with or without it, the
+ // suspect channel is replaced by the next query to it, so recovery does not
depend on the failure detector.
+ private volatile Consumer<ServerRoutingInstance> _unansweredServerReporter =
serverRoutingInstance -> {
+ };
/// Ping request ids count down from -1, so they can never collide with
query request ids, which are non-negative
/// (see `BrokerRequestIdGenerator`).
private final AtomicLong _pingRequestIdGenerator = new AtomicLong();
@@ -90,12 +97,31 @@ public class QueryRouter {
/// @param tlsConfig TLS config
public QueryRouter(String brokerId, @Nullable NettyConfig nettyConfig,
@Nullable TlsConfig tlsConfig,
ServerRoutingStatsManager serverRoutingStatsManager, ThreadAccountant
threadAccountant) {
+ this(brokerId, nettyConfig, tlsConfig, serverRoutingStatsManager,
threadAccountant,
+ Broker.FailureDetector.DEFAULT_UNANSWERED_QUERIES_THRESHOLD,
+ Broker.FailureDetector.DEFAULT_UNANSWERED_QUERY_GRACE_MS);
+ }
+
+ /// @param unansweredQueriesThreshold see
[Broker.FailureDetector#CONFIG_OF_UNANSWERED_QUERIES_THRESHOLD];
+ /// non-positive disables the check
+ /// @param unansweredQueryGraceMs see
[Broker.FailureDetector#CONFIG_OF_UNANSWERED_QUERY_GRACE_MS]
+ public QueryRouter(String brokerId, @Nullable NettyConfig nettyConfig,
@Nullable TlsConfig tlsConfig,
+ ServerRoutingStatsManager serverRoutingStatsManager, ThreadAccountant
threadAccountant,
+ int unansweredQueriesThreshold, long unansweredQueryGraceMs) {
_brokerId = brokerId;
- _serverChannels = new ServerChannels(this, nettyConfig, null,
threadAccountant);
- _serverChannelsTls = tlsConfig != null ? new ServerChannels(this,
nettyConfig, tlsConfig, threadAccountant) : null;
+ _serverChannels = new ServerChannels(this, nettyConfig, null,
threadAccountant, unansweredQueriesThreshold,
+ unansweredQueryGraceMs);
+ _serverChannelsTls = tlsConfig != null ? new ServerChannels(this,
nettyConfig, tlsConfig, threadAccountant,
+ unansweredQueriesThreshold, unansweredQueryGraceMs) : null;
_serverRoutingStatsManager = serverRoutingStatsManager;
}
+ /// Registers the consumer told about a server that stopped answering
queries on one of its channels, typically the
+ /// failure detector. Called from the transport's liveness thread, never
from an event loop.
+ public void setUnansweredServerReporter(Consumer<ServerRoutingInstance>
unansweredServerReporter) {
+ _unansweredServerReporter = unansweredServerReporter;
+ }
+
public AsyncQueryResponse submitQuery(long requestId, String rawTableName,
@Nullable BrokerRequest offlineBrokerRequest,
@Nullable Map<ServerInstance, SegmentsToQuery> offlineRoutingTable,
@@ -180,18 +206,19 @@ public class QueryRouter {
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, which
+ /// includes the vacuous case of a server with no such channel: callers that
need "reachable" must check
+ /// [#hasChannel] first, as `retryUnhealthyServer` does. 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;
} catch (Exception e) {
@@ -295,6 +322,31 @@ public class QueryRouter {
return routingInstances;
}
+ /// Records that a query written at `submitTimeMs` to `channel` (when known)
timed out unanswered; the transport
+ /// judges it after a grace period (see
[ServerChannels#recordUnansweredQuery]).
+ void recordServerTimeout(ServerRoutingInstance serverRoutingInstance, long
submitTimeMs, @Nullable Channel channel) {
+
serverChannelsFor(serverRoutingInstance).recordUnansweredQuery(serverRoutingInstance,
submitTimeMs, channel);
+ }
+
+ /// Called by the transport once a server has left the configured number of
consecutive queries unanswered on a
+ /// channel; hands the server to the registered reporter.
+ void reportUnansweredServer(ServerRoutingInstance serverRoutingInstance) {
+ try {
+ _unansweredServerReporter.accept(serverRoutingInstance);
+ } catch (Exception e) {
+ LOGGER.error("Caught exception while reporting unanswered server: {}",
serverRoutingInstance, e);
+ }
+ }
+
+ @VisibleForTesting
+ int getUnansweredQueries(ServerRoutingInstance serverRoutingInstance) {
+ return
serverChannelsFor(serverRoutingInstance).getUnansweredQueries(serverRoutingInstance);
+ }
+
+ private ServerChannels serverChannelsFor(ServerRoutingInstance
serverRoutingInstance) {
+ return serverRoutingInstance.isTlsEnabled() && _serverChannelsTls != null
? _serverChannelsTls : _serverChannels;
+ }
+
@VisibleForTesting
ServerChannels getServerChannels() {
return _serverChannels;
@@ -390,6 +442,13 @@ public class QueryRouter {
}
}
+ /// Records that a response frame arrived from the server, before it is
decoded: any response proves the server and
+ /// its channel alive, whether or not the query it answers is still waiting,
and whether or not it decodes (a heap
+ /// OOM while deserializing is the broker's problem, not a silent server).
Called by [DataTableHandler].
+ void recordResponse(ServerRoutingInstance serverRoutingInstance) {
+
serverChannelsFor(serverRoutingInstance).recordResponse(serverRoutingInstance);
+ }
+
void receiveDataTable(ServerRoutingInstance serverRoutingInstance, DataTable
dataTable, int responseSize,
int deserializationTimeMs) {
long requestId =
Long.parseLong(dataTable.getMetadata().get(MetadataKey.REQUEST_ID.getName()));
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java
b/pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java
index 75992359ffc..9d58f8df08c 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java
@@ -19,6 +19,7 @@
package org.apache.pinot.core.transport;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.PooledByteBufAllocatorMetric;
@@ -42,8 +43,12 @@ import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.ssl.SslHandler;
import io.netty.util.concurrent.Future;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.ToLongFunction;
import javax.annotation.Nullable;
@@ -57,6 +62,7 @@ import org.apache.pinot.common.metrics.BrokerTimer;
import org.apache.pinot.common.request.InstanceRequest;
import org.apache.pinot.core.util.OsCheck;
import org.apache.pinot.spi.accounting.ThreadAccountant;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
import org.apache.thrift.TSerializer;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.transport.TTransportException;
@@ -79,7 +85,10 @@ import org.slf4j.LoggerFactory;
/// - the routing manager evicts the cached channels of a server when it
enables it, as after a restart, and when it
/// sees the server removed from the cluster or moved to another address
([#evictChannel]); a channel that still
/// carries queries when its server is enabled is retired instead: the
next query opens a fresh connection, and
-/// the old one is closed once the queries on it are past their deadline
([#evictChannelOfEnabledServer]).
+/// the old one is closed once the queries on it are past their deadline
([#evictChannelOfEnabledServer]);
+/// - a channel on which the server left
[Broker.FailureDetector#CONFIG_OF_UNANSWERED_QUERIES_THRESHOLD] consecutive
+/// queries unanswered, each judged a grace period after its timeout, is
marked suspect: the next query to it, or
+/// the failure detector's retry probe, replaces it with a fresh
connection ([#recordUnansweredQuery]).
@ThreadSafe
public class ServerChannels {
private static final Logger LOGGER =
LoggerFactory.getLogger(ServerChannels.class);
@@ -111,6 +120,16 @@ public class ServerChannels {
private final long _writeTimeoutMs;
private final int _tcpUserTimeoutMs;
private final WriteBufferWaterMark _writeBufferWaterMark;
+ private final int _unansweredQueriesThreshold;
+ private final long _unansweredQueryGraceMs;
+ // Bound on the judgments queued per channel at a time. Under a dead server
at high QPS every timed-out query would
+ // otherwise queue a task for the whole grace period; the first few
judgments already cross the threshold, so the
+ // rest add nothing but queue depth.
+ private final int _maxPendingJudgmentsPerChannel;
+ // Judges unanswered queries once their grace period is over, and reports
the server from there; null when the
+ // check is disabled. Its own thread, because reporting ends in a routing
update that must not run on an event loop.
+ @Nullable
+ private final ScheduledExecutorService _livenessExecutor;
private final BrokerMetrics _brokerMetrics = BrokerMetrics.get();
private final ConcurrentHashMap<ServerRoutingInstance, ServerChannel>
_serverToChannelMap = new ConcurrentHashMap<>();
@@ -121,6 +140,16 @@ public class ServerChannels {
/// @param tlsConfig TLS/SSL config
public ServerChannels(QueryRouter queryRouter, @Nullable NettyConfig
nettyConfig, @Nullable TlsConfig tlsConfig,
ThreadAccountant threadAccountant) {
+ this(queryRouter, nettyConfig, tlsConfig, threadAccountant,
+ Broker.FailureDetector.DEFAULT_UNANSWERED_QUERIES_THRESHOLD,
+ Broker.FailureDetector.DEFAULT_UNANSWERED_QUERY_GRACE_MS);
+ }
+
+ /// @param unansweredQueriesThreshold see
[Broker.FailureDetector#CONFIG_OF_UNANSWERED_QUERIES_THRESHOLD];
+ /// non-positive disables the check
+ /// @param unansweredQueryGraceMs see
[Broker.FailureDetector#CONFIG_OF_UNANSWERED_QUERY_GRACE_MS]
+ public ServerChannels(QueryRouter queryRouter, @Nullable NettyConfig
nettyConfig, @Nullable TlsConfig tlsConfig,
+ ThreadAccountant threadAccountant, int unansweredQueriesThreshold, long
unansweredQueryGraceMs) {
if (nettyConfig == null) {
nettyConfig = new NettyConfig();
}
@@ -159,9 +188,16 @@ public class ServerChannels {
_tcpUserTimeoutMs = nettyConfig.getTcpUserTimeoutMs();
_writeBufferWaterMark = new
WriteBufferWaterMark(nettyConfig.getWriteBufferLowWaterMarkBytes(),
nettyConfig.getWriteBufferHighWaterMarkBytes());
+ _unansweredQueriesThreshold = unansweredQueriesThreshold;
+ _unansweredQueryGraceMs = Math.max(0L, unansweredQueryGraceMs);
+ _maxPendingJudgmentsPerChannel = Math.max(16, 2 *
unansweredQueriesThreshold);
+ _livenessExecutor = unansweredQueriesThreshold > 0 ?
Executors.newSingleThreadScheduledExecutor(
+ new
ThreadFactoryBuilder().setNameFormat("server-channel-liveness-%d").setDaemon(true).build())
: null;
LOGGER.info("Server channel liveness settings: write timeout: {}ms, TCP
user timeout: {}ms ({}), write buffer "
- + "water mark: {}", _writeTimeoutMs, _tcpUserTimeoutMs,
- isTcpUserTimeoutApplied() ? "applied" : "not applied, requires the
Epoll transport", _writeBufferWaterMark);
+ + "water mark: {}, unanswered queries threshold: {} ({}),
unanswered query grace: {}ms", _writeTimeoutMs,
+ _tcpUserTimeoutMs, isTcpUserTimeoutApplied() ? "applied" : "not
applied, requires the Epoll transport",
+ _writeBufferWaterMark, unansweredQueriesThreshold, _livenessExecutor
!= null ? "enabled" : "disabled",
+ _unansweredQueryGraceMs);
_bufAllocatorWithLimits =
PooledByteBufAllocatorWithLimits.getSharedBufferAllocatorWithLimits();
PooledByteBufAllocatorMetric metric = _bufAllocatorWithLimits.metric();
@@ -309,7 +345,90 @@ public class ServerChannels {
_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) {
+ // Enough judgments queued to cross the threshold already; this query
adds nothing
+ serverChannel._numPendingJudgments.decrementAndGet();
+ return;
+ }
+ try {
+ _livenessExecutor.schedule(() -> {
+ try {
+ serverChannel.judgeUnansweredQuery(submitTimeMs, channel);
+ } catch (Exception e) {
+ LOGGER.error("Caught exception while judging an unanswered query to
server: {}", serverRoutingInstance, e);
+ } finally {
+ serverChannel._numPendingJudgments.decrementAndGet();
+ }
+ }, _unansweredQueryGraceMs, TimeUnit.MILLISECONDS);
+ } catch (RejectedExecutionException e) {
+ // Shutting down: nothing left to judge
+ serverChannel._numPendingJudgments.decrementAndGet();
+ }
+ }
+
+ @VisibleForTesting
+ int getUnansweredQueries(ServerRoutingInstance serverRoutingInstance) {
+ ServerChannel serverChannel =
_serverToChannelMap.get(serverRoutingInstance);
+ return serverChannel != null ? serverChannel.getUnansweredQueries() : 0;
+ }
+
+ @VisibleForTesting
+ int getNumPendingJudgments(ServerRoutingInstance serverRoutingInstance) {
+ ServerChannel serverChannel =
_serverToChannelMap.get(serverRoutingInstance);
+ return serverChannel != null ? serverChannel._numPendingJudgments.get() :
0;
+ }
+
public void shutDown() {
+ if (_livenessExecutor != null) {
+ _livenessExecutor.shutdownNow();
+ }
// Shut down immediately
_eventLoopGroup.shutdownGracefully(0, 0, TimeUnit.SECONDS);
}
@@ -350,6 +469,20 @@ public class ServerChannels {
// then replaces it with a fresh connection rather than reuse it, and it
stays open for those queries until it is
// closed once they are past their deadline. Compared by identity, so it
never applies to a connection opened since.
volatile Channel _retiredChannel;
+ // Set when the server stopped answering queries on this channel while the
socket still looked active (see
+ // judgeUnansweredQuery). connectWithoutLocking() then replaces the
channel rather than trusting it. A full outbound
+ // queue does not set it: a healthy server draining a large request fills
it too, and replacing the channel would
+ // fail that request; a stall that lasts is the write timeout's to judge.
+ volatile boolean _suspect;
+ // Liveness bookkeeping, guarded by _livenessLock: when the server last
sent anything on this channel (the start of
+ // the current connection for a fresh one) and how many consecutive
queries it has left unanswered since. The lock
+ // is taken briefly on the event loop (per response), on the query thread
(when a connection is replaced) and on the
+ // liveness thread (judgment), so that a response or a replacement landing
while a judgment runs is never lost.
+ private final Object _livenessLock = new Object();
+ private long _lastResponseTimeMs;
+ private int _unansweredQueries;
+ // Judgments queued for this channel, bounded by
ServerChannels#_maxPendingJudgmentsPerChannel
+ final AtomicInteger _numPendingJudgments = new AtomicInteger();
ServerChannel(ServerRoutingInstance serverRoutingInstance) {
_serverRoutingInstance = serverRoutingInstance;
@@ -400,9 +533,9 @@ public class ServerChannels {
}
}
- /// Closes a channel the broker decided to drop (eviction, a retired
channel once its queries are past their
- /// deadline): the queries still in flight on it fail through
`channelInactive` as on any closed channel, but the
- /// server is not reported for it.
+ /// Closes a channel the broker decided to drop (eviction, replacement of
a suspect channel, a retired channel
+ /// once its queries are past their deadline): the queries still in flight
on it fail through `channelInactive` as
+ /// on any closed channel, but the server is not reported for it.
private void closeByBroker(Channel channel) {
DataTableHandler handler =
channel.pipeline().get(DataTableHandler.class);
if (handler != null) {
@@ -443,6 +576,58 @@ public class ServerChannels {
}
}
+ void recordResponse() {
+ synchronized (_livenessLock) {
+ _lastResponseTimeMs = System.currentTimeMillis();
+ _unansweredQueries = 0;
+ }
+ }
+
+ int getUnansweredQueries() {
+ synchronized (_livenessLock) {
+ return _unansweredQueries;
+ }
+ }
+
+ /// Starts a clean record for a new connection: not suspect, nothing
unanswered, counted as answered from now. Done
+ /// before a replacement connects as well as after, so that a judgment of
a query written to the dropped channel
+ /// that runs while the connect is in progress finds nothing to count.
+ private void resetLiveness() {
+ synchronized (_livenessLock) {
+ _suspect = false;
+ _lastResponseTimeMs = System.currentTimeMillis();
+ _unansweredQueries = 0;
+ }
+ }
+
+ /// Runs on the liveness thread a grace period after a query written at
`submitTimeMs` to `channel` timed out. See
+ /// [ServerChannels#recordUnansweredQuery].
+ void judgeUnansweredQuery(long submitTimeMs, @Nullable Channel channel) {
+ int unansweredQueries;
+ synchronized (_livenessLock) {
+ // Nothing to add once the channel is already suspect (it is reported
once, until replaced) or no longer in use,
+ // for a query written to a channel since replaced or closed (it says
nothing about the current one), or if the
+ // server answered something since the query was written. Strictly
after: a connect and the first write can
+ // share a millisecond, while a real answer never shares one with its
request. Checked and acted on under the
+ // lock, so a response or a replacement that lands meanwhile either
makes this judgment bail or resets the
+ // count right after it.
+ if (_evicted || _suspect || (channel != null && channel != _channel)
|| _lastResponseTimeMs > submitTimeMs) {
+ return;
+ }
+ unansweredQueries = ++_unansweredQueries;
+ if (unansweredQueries < _unansweredQueriesThreshold) {
+ return;
+ }
+ _unansweredQueries = 0;
+ _suspect = true;
+ }
+ LOGGER.warn("Server: {} has left {} consecutive queries unanswered on
its channel, each judged {}ms past its "
+ + "timeout; marking the channel suspect", _serverRoutingInstance,
unansweredQueries, _unansweredQueryGraceMs);
+
_brokerMetrics.addMeteredValue(BrokerMeter.NETTY_CONNECTION_SUSPECT_CHANNELS, 1,
+ _serverRoutingInstance.getShortName());
+ _queryRouter.reportUnansweredServer(_serverRoutingInstance);
+ }
+
@VisibleForTesting
void setChannel(Channel channel) {
_channel = channel;
@@ -502,25 +687,37 @@ public class ServerChannels {
///
/// Without the bound, a connect to a host that silently drops packets --
the very host the failure detector pings
/// -- blocks for Netty's 30 s default, however long the caller can
actually wait.
+ ///
+ /// 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).
void connectWithoutLocking(long timeoutMs)
throws InterruptedException, TimeoutException {
Channel channel = _channel;
- if (channel != null && channel == _retiredChannel) {
+ if (channel != null && channel.isActive() && !_suspect && channel !=
_retiredChannel) {
+ return;
+ }
+ if (channel != null && _suspect) {
+ LOGGER.info("Replacing suspect channel to server: {} with a fresh
connection", _serverRoutingInstance);
+ _channel = null;
+ closeByBroker(channel);
+ } else if (channel != null && channel == _retiredChannel) {
// Left open for the queries already on it, and closed once they are
past their deadline (see retire())
LOGGER.info("Replacing retired channel to server: {} with a fresh
connection", _serverRoutingInstance);
_channel = null;
}
- if (_channel == null || !_channel.isActive()) {
- long startTime = System.currentTimeMillis();
- ChannelFuture connectFuture = _bootstrap.connect();
- if (!connectFuture.await(timeoutMs, TimeUnit.MILLISECONDS)) {
- connectFuture.channel().close();
- throw new TimeoutException("Timed out connecting to server: " +
_serverRoutingInstance);
- }
- // Completed by now: returns the channel, or rethrows the connect
failure as sync() always did.
- _channel = connectFuture.sync().channel();
- recordConnectTime(System.currentTimeMillis() - startTime);
+ resetLiveness();
+ long startTime = System.currentTimeMillis();
+ ChannelFuture connectFuture = _bootstrap.connect();
+ if (!connectFuture.await(timeoutMs, TimeUnit.MILLISECONDS)) {
+ connectFuture.channel().close();
+ throw new TimeoutException("Timed out connecting to server: " +
_serverRoutingInstance);
}
+ // Completed by now: returns the channel, or rethrows the connect
failure as sync() always did.
+ _channel = connectFuture.sync().channel();
+ resetLiveness();
+ recordConnectTime(System.currentTimeMillis() - startTime);
}
/// Like [#connectWithoutLocking(long)] but additionally waits out the TLS
handshake, and bounds both
@@ -560,6 +757,7 @@ public class ServerChannels {
throw t;
}
_channel = channel;
+ resetLiveness();
recordConnectTime(System.currentTimeMillis() - startTime);
}
@@ -657,10 +855,12 @@ public class ServerChannels {
}
/// Opens the TCP connection, as the failure detector's reconnect probe
- /// (`SingleConnectionBrokerRequestHandler#retryUnhealthyServer`) has
always done. Deliberately does
- /// **not** await the TLS handshake: this runs at steady state under live
traffic, and holding
- /// `_channelLock` across a handshake would make concurrent queries to a
recovering server queue
- /// behind it -- the very serialization startup pre-connect exists to
remove.
+ /// (`SingleConnectionBrokerRequestHandler#retryUnhealthyServer`) has
always done, replacing the channel when it is
+ /// suspect (see [#connectWithoutLocking(long)]): a probe that only looked
at `isActive()` would vouch for a
+ /// half-open channel and put the server straight back into routing on it.
Deliberately does **not** await the TLS
+ /// handshake: this runs at steady state under live traffic, and holding
`_channelLock` across a handshake would
+ /// make concurrent queries to a recovering server queue behind it -- the
very serialization startup pre-connect
+ /// exists to remove.
void connect()
throws InterruptedException, TimeoutException {
if (_channelLock.tryLock(TRY_CONNECT_CHANNEL_LOCK_TIMEOUT_MS,
TimeUnit.MILLISECONDS)) {
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/transport/AsyncQueryResponseTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/transport/AsyncQueryResponseTest.java
index 03488ea37ae..a4305c768e3 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/transport/AsyncQueryResponseTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/transport/AsyncQueryResponseTest.java
@@ -21,13 +21,19 @@ package org.apache.pinot.core.transport;
import io.netty.channel.Channel;
import java.util.Map;
import java.util.Set;
+import org.apache.pinot.common.datatable.DataTable;
import org.apache.pinot.core.common.datatable.DataTableBuilderFactory;
import
org.apache.pinot.core.transport.server.routing.stats.ServerRoutingStatsManager;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.exception.QueryCancelledException;
import org.testng.annotations.Test;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
@@ -216,4 +222,43 @@ public class AsyncQueryResponseTest {
response.markServerUnavailable(SERVER_2, requestChannel, new
RuntimeException("inactive"), true);
assertFalse(response.isAwaitingResponseOn(SERVER_2, requestChannel),
"Counted down");
}
+
+ /// A timeout hands every server the request reached but that never answered
to the transport, with the time the
+ /// request was written, and blames nobody on the spot. Servers the request
never reached are not handed over.
+ @Test
+ public void testTimeoutRecordsUnansweredServers()
+ throws Exception {
+ QueryRouter queryRouter = mock(QueryRouter.class);
+ long startTimeMs = System.currentTimeMillis();
+ AsyncQueryResponse response =
+ new AsyncQueryResponse(queryRouter, 1L, Set.of(SERVER_1, SERVER_2),
startTimeMs, 50L,
+ mock(ServerRoutingStatsManager.class), false);
+ // SERVER_2 is never marked submitted: its send failed before anything was
written
+ response.markRequestSubmitted(SERVER_1);
+
+ response.getFinalResponses();
+
+ assertEquals(response.getStatus(), QueryResponse.Status.TIMED_OUT);
+ assertNull(response.getFailedServer());
+ assertNull(response.getException());
+ verify(queryRouter).recordServerTimeout(eq(SERVER_1), anyLong(), any());
+ verify(queryRouter, never()).recordServerTimeout(eq(SERVER_2), anyLong(),
any());
+ }
+
+ @Test
+ public void testAnsweredQueryRecordsNoTimeout()
+ throws Exception {
+ QueryRouter queryRouter = mock(QueryRouter.class);
+ AsyncQueryResponse response =
+ new AsyncQueryResponse(queryRouter, 1L, Set.of(SERVER_1),
System.currentTimeMillis(), 10_000L,
+ mock(ServerRoutingStatsManager.class), false);
+ response.markRequestSubmitted(SERVER_1);
+ response.receiveDataTable(SERVER_1, mock(DataTable.class), 10, 1);
+
+ response.getFinalResponses();
+
+ assertEquals(response.getStatus(), QueryResponse.Status.COMPLETED);
+ verify(queryRouter, never()).recordServerTimeout(eq(SERVER_1), anyLong(),
any());
+ assertNull(response.getFailedServer());
+ }
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/transport/QueryRoutingTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/transport/QueryRoutingTest.java
index 9d883f68f02..be2ee941d98 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/transport/QueryRoutingTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/transport/QueryRoutingTest.java
@@ -31,6 +31,7 @@ import io.netty.channel.ChannelOutboundHandlerAdapter;
import io.netty.channel.ChannelPromise;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.channel.socket.SocketChannel;
import io.netty.util.ReferenceCountUtil;
import java.io.IOException;
import java.io.InputStream;
@@ -45,6 +46,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -84,6 +86,7 @@ import org.testng.annotations.Test;
import static
org.apache.pinot.core.transport.QueryServerTestUtils.startAndGetPort;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
@@ -175,6 +178,43 @@ public class QueryRoutingTest {
return new QueryRouter("testBroker", null, null, statsManager,
ThreadAccountantUtils.getNoOpAccountant());
}
+ private QueryRouter newIsolatedQueryRouter(int unansweredQueriesThreshold,
long unansweredQueryGraceMs) {
+ Map<String, Object> properties = new HashMap<>();
+
properties.put(CommonConstants.Broker.AdaptiveServerSelector.CONFIG_OF_ENABLE_STATS_COLLECTION,
true);
+ ServerRoutingStatsManager statsManager =
+ new ServerRoutingStatsManager(new PinotConfiguration(properties),
mock(BrokerMetrics.class));
+ statsManager.init();
+ return new QueryRouter("testBroker", null, null, statsManager,
ThreadAccountantUtils.getNoOpAccountant(),
+ unansweredQueriesThreshold, unansweredQueryGraceMs);
+ }
+
+ private AsyncQueryResponse submitRealtimeQuery(QueryRouter queryRouter, long
requestId, long timeoutMs)
+ throws Exception {
+ AsyncQueryResponse asyncQueryResponse =
+ queryRouter.submitQuery(requestId, "testTable", null, null,
BROKER_REQUEST, _routingTable, timeoutMs);
+ asyncQueryResponse.getFinalResponses();
+ return asyncQueryResponse;
+ }
+
+ private static void waitForUnansweredQueries(QueryRouter queryRouter,
ServerRoutingInstance instance, int expected) {
+ TestUtils.waitForCondition(aVoid ->
queryRouter.getUnansweredQueries(instance) == expected, 10L, 5_000L,
+ "Expected " + expected + " unanswered queries counted against " +
instance);
+ }
+
+ /// A server that answers nothing until the test completes the current gate,
and then answers every request it has
+ /// queued; the gate can be swapped to hold the next requests again. The
scheduler returns the gate itself, so the
+ /// server's event loop is never blocked and keeps reading requests (and
noticing closed connections).
+ private QueryServer
getGatedQueryServer(AtomicReference<SettableFuture<DataBuffer>> gate) {
+ return QueryServerTestUtils.newQueryServer(queryRequest -> gate.get());
+ }
+
+ private static byte[] responseBytesForRequest(long requestId)
+ throws Exception {
+ DataTable dataTable = DataTableBuilderFactory.getEmptyDataTable();
+ dataTable.getMetadata().put(MetadataKey.REQUEST_ID.getName(),
Long.toString(requestId));
+ return dataTable.toBytes();
+ }
+
@Test
public void testValidResponse()
throws Exception {
@@ -1010,6 +1050,158 @@ public class QueryRoutingTest {
}
}
+ /// The retry probe used to look at the OFFLINE channel only. On a
realtime-only cluster it therefore had no
+ /// channel to check, answered UNKNOWN, and the failure detector readmitted
the server onto whatever channel was
+ /// cached.
+ @Test
+ public void testRetryProbeCoversRealtimeOnlyChannel()
+ throws Exception {
+ long requestId = 123;
+ _queryServer = getQueryServer(0, responseBytesForRequest(requestId));
+ initializeTestFixtures(startAndGetPort(_queryServer));
+ QueryRouter queryRouter = newIsolatedQueryRouter();
+ try {
+ assertFalse(queryRouter.hasChannel(_serverInstance), "No query sent
yet");
+
+ AsyncQueryResponse asyncQueryResponse =
+ queryRouter.submitQuery(requestId, "testTable", null, null,
BROKER_REQUEST, _routingTable, 10_000L);
+
assertNotNull(asyncQueryResponse.getFinalResponses().get(_realtimeServerRoutingInstance).getDataTable());
+ assertTrue(queryRouter.hasChannel(_serverInstance), "The REALTIME
channel carries queries");
+ TestUtils.waitForCondition(aVoid ->
_queryServer.getConnectedChannelCount() == 1, 10L, 5_000L,
+ "Expected the REALTIME channel on the server");
+ SocketChannel healthyChannel =
_queryServer.getConnectedChannels().iterator().next();
+
+ // A healthy channel passes the probe and is kept, not replaced
+ assertTrue(queryRouter.connect(_serverInstance));
+ assertEquals(_queryServer.getConnectedChannelCount(), 1);
+ assertSame(_queryServer.getConnectedChannels().iterator().next(),
healthyChannel);
+ } finally {
+ queryRouter.shutDown();
+ }
+ }
+
+ /// A dead channel that still reports active answers nothing. Each
unanswered query is judged a grace period after
+ /// its timeout; at the threshold the server is reported, and the channel is
replaced by the next query to it
+ /// (so recovery does not depend on a failure detector being configured) or
by the detector's probe. Timeouts
+ /// themselves never fail the query or blame the server on the spot.
+ @Test
+ public void testUnansweredQueriesReportServerAndReplaceChannel()
+ throws Exception {
+ long requestId = 123;
+ // Never completed: the server answers nothing for the whole test
+ AtomicReference<SettableFuture<DataBuffer>> gate = new
AtomicReference<>(SettableFuture.create());
+ _queryServer = getGatedQueryServer(gate);
+ initializeTestFixtures(startAndGetPort(_queryServer));
+ int threshold = 3;
+ long timeoutMs = 300L;
+ long graceMs = 200L;
+ QueryRouter queryRouter = newIsolatedQueryRouter(threshold, graceMs);
+ List<ServerRoutingInstance> reported = new CopyOnWriteArrayList<>();
+ queryRouter.setUnansweredServerReporter(reported::add);
+ try {
+ for (int i = 1; i <= threshold; i++) {
+ AsyncQueryResponse asyncQueryResponse =
submitRealtimeQuery(queryRouter, requestId, timeoutMs);
+ assertEquals(asyncQueryResponse.getStatus(),
QueryResponse.Status.TIMED_OUT);
+ assertNull(asyncQueryResponse.getFailedServer(), "A timeout blames
nobody on the spot");
+ assertNull(asyncQueryResponse.getException(), "A timeout is not a send
failure");
+ // Judged a grace period later; the count resets once the threshold is
reached
+ waitForUnansweredQueries(queryRouter, _realtimeServerRoutingInstance,
i < threshold ? i : 0);
+ }
+ TestUtils.waitForCondition(aVoid ->
reported.equals(List.of(_realtimeServerRoutingInstance)), 10L, 5_000L,
+ "The server was not reported after " + threshold + " unanswered
queries");
+ TestUtils.waitForCondition(aVoid ->
_queryServer.getConnectedChannelCount() == 1, 10L, 5_000L,
+ "Expected the REALTIME channel on the server");
+ SocketChannel suspectChannel =
_queryServer.getConnectedChannels().iterator().next();
+
+ // The next query is written to a fresh connection and the suspect
channel is closed, without the server being
+ // reported again for the close
+ AsyncQueryResponse asyncQueryResponse = submitRealtimeQuery(queryRouter,
requestId, timeoutMs);
+ assertEquals(asyncQueryResponse.getStatus(),
QueryResponse.Status.TIMED_OUT);
+ TestUtils.waitForCondition(aVoid ->
_queryServer.getConnectedChannelCount() == 1
+ && _queryServer.getConnectedChannels().iterator().next() !=
suspectChannel, 10L, 5_000L,
+ "The suspect channel was not replaced by the next query");
+ assertEquals(reported.size(), 1);
+ assertTrue(queryRouter.hasChannel(_serverInstance));
+
+ // The probe keeps the fresh channel ...
+ SocketChannel freshChannel =
_queryServer.getConnectedChannels().iterator().next();
+ assertTrue(queryRouter.connect(_serverInstance));
+ assertEquals(_queryServer.getConnectedChannelCount(), 1);
+ assertSame(_queryServer.getConnectedChannels().iterator().next(),
freshChannel);
+
+ // ... and replaces it once it is suspect in turn, which is the path the
failure detector takes after a report
+ // (the first unanswered query above already counted once against the
fresh channel)
+ for (int i = 2; i <= threshold; i++) {
+ submitRealtimeQuery(queryRouter, requestId, timeoutMs);
+ }
+ TestUtils.waitForCondition(aVoid -> reported.size() == 2, 10L, 5_000L,
"The server was not reported again");
+ assertTrue(queryRouter.connect(_serverInstance));
+ TestUtils.waitForCondition(aVoid ->
_queryServer.getConnectedChannelCount() == 1
+ && _queryServer.getConnectedChannels().iterator().next() !=
freshChannel, 10L, 5_000L,
+ "The suspect channel was not replaced by the probe");
+ } finally {
+ gate.get().set(PinotByteBuffer.wrap(new byte[0]));
+ queryRouter.shutDown();
+ }
+ }
+
+ /// Any response from the server within the grace period, even a late one
for a query that already timed out,
+ /// clears the query: a slow server that still answers is never mistaken for
a dead channel, and a broker that was
+ /// late reading the responses does not blame the servers.
+ @Test
+ public void testLateResponseClearsUnansweredQuery()
+ throws Exception {
+ long requestId = 123;
+ AtomicReference<SettableFuture<DataBuffer>> gate = new
AtomicReference<>(SettableFuture.create());
+ _queryServer = getGatedQueryServer(gate);
+ initializeTestFixtures(startAndGetPort(_queryServer));
+ int threshold = 2;
+ long timeoutMs = 300L;
+ long graceMs = 1_000L;
+ QueryRouter queryRouter = newIsolatedQueryRouter(threshold, graceMs);
+ List<ServerRoutingInstance> reported = new CopyOnWriteArrayList<>();
+ queryRouter.setUnansweredServerReporter(reported::add);
+ try {
+ AsyncQueryResponse asyncQueryResponse = submitRealtimeQuery(queryRouter,
requestId, timeoutMs);
+ assertEquals(asyncQueryResponse.getStatus(),
QueryResponse.Status.TIMED_OUT);
+ // The server answers the first query late, inside the grace period:
useless to the query, proof of life for
+ // the channel
+
gate.getAndSet(SettableFuture.create()).set(PinotByteBuffer.wrap(responseBytesForRequest(requestId)));
+ Thread.sleep(graceMs + 500L);
+
assertEquals(queryRouter.getUnansweredQueries(_realtimeServerRoutingInstance),
0,
+ "A query answered within the grace period must not count");
+
+ // Two queries the server never answers do count, and reach the threshold
+ submitRealtimeQuery(queryRouter, requestId, timeoutMs);
+ waitForUnansweredQueries(queryRouter, _realtimeServerRoutingInstance, 1);
+ assertTrue(reported.isEmpty());
+ submitRealtimeQuery(queryRouter, requestId, timeoutMs);
+ TestUtils.waitForCondition(aVoid ->
reported.equals(List.of(_realtimeServerRoutingInstance)), 10L, 5_000L,
+ "The server was not reported");
+ } finally {
+ gate.get().set(PinotByteBuffer.wrap(new byte[0]));
+ queryRouter.shutDown();
+ }
+ }
+
+ /// A response frame counts as an answer from the server as soon as it
arrives, before it is decoded: one that fails
+ /// to deserialize, a heap OOM included, still proves the server alive, so
it must not count towards the unanswered
+ /// queries that get a channel replaced.
+ @Test
+ public void testUndecodableResponseStillCountsAsAnswer() {
+ QueryRouter queryRouter = mock(QueryRouter.class);
+ ServerRoutingInstance routingInstance = new
ServerRoutingInstance("localhost", 12345, TableType.OFFLINE);
+ EmbeddedChannel channel = new EmbeddedChannel(
+ new DataTableHandler(queryRouter,
ThreadAccountantUtils.getNoOpAccountant(), routingInstance));
+ try {
+ channel.writeInbound(Unpooled.wrappedBuffer(new byte[]{1, 2, 3}));
+ verify(queryRouter).recordResponse(routingInstance);
+ verify(queryRouter, never()).receiveDataTable(any(), any(), anyInt(),
anyInt());
+ } finally {
+ channel.finishAndReleaseAll();
+ }
+ }
+
@Test
public void testChannelActiveInactiveEmitPerServerTaggedMeters() {
PinotMetricUtils.init(new PinotConfiguration());
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/transport/ServerChannelsTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/transport/ServerChannelsTest.java
index 587074ef75a..c54ebe4ae92 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/transport/ServerChannelsTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/transport/ServerChannelsTest.java
@@ -499,6 +499,9 @@ public class ServerChannelsTest {
verify(mockChannel, never()).close();
verify(asyncQueryResponse, never()).markRequestSent(any(), anyInt());
verify(asyncQueryResponse, never()).markServerUnavailable(any(), any(),
any(Exception.class), anyBoolean());
+ // Not marked suspect: a healthy server draining a large request fills
the queue too, and replacing the channel
+ // would fail that request
+ assertFalse(serverChannel._suspect);
} finally {
serverChannels.shutDown();
}
@@ -1030,9 +1033,10 @@ public class ServerChannelsTest {
assertTrue(e.getMessage().contains("not writable"), e.getMessage());
assertEquals(meterCount(BrokerMeter.NETTY_CONNECTION_UNWRITABLE_REJECTIONS,
instance), rejectionsBefore + 1);
verify(rejectedResponse, never()).markRequestSent(any(), anyInt());
- // Rejecting does not close the channel; the write timeout handler owns
that decision
+ // Rejecting neither closes nor replaces the channel; the write timeout
handler owns that decision
assertTrue(channel.isActive());
assertSame(serverChannel._channel, channel);
+ assertFalse(serverChannel._suspect);
} finally {
serverChannels.shutDown();
}
@@ -1199,6 +1203,226 @@ public class ServerChannelsTest {
}
}
+ /// A channel the server stopped answering on is replaced by the next query
to it, so recovery does not wait for
+ /// a failure detector, and the queries still in flight on it fail without
blaming the server.
+ @Test
+ public void testNextRequestReplacesSuspectChannel()
+ throws Exception {
+ ServerChannels serverChannels =
+ new ServerChannels(mock(QueryRouter.class), null, null,
ThreadAccountantUtils.getNoOpAccountant());
+ try (StalledServer server = new StalledServer()) {
+ ServerRoutingInstance instance = new ServerRoutingInstance("localhost",
server.getPort(), TableType.OFFLINE);
+ AsyncQueryResponse firstResponse = mock(AsyncQueryResponse.class);
+ Channel suspectChannel =
+ serverChannels.sendRequest("testTable", firstResponse, instance,
smallInstanceRequest(), 10_000L);
+ assertNotNull(suspectChannel);
+ ServerChannels.ServerChannel entry =
serverChannels.getOrCreateServerChannel(instance);
+ assertSame(entry._channel, suspectChannel);
+
+ serverChannels.markSuspect(instance);
+ AsyncQueryResponse nextResponse = mock(AsyncQueryResponse.class);
+ Channel freshChannel =
+ serverChannels.sendRequest("testTable", nextResponse, instance,
smallInstanceRequest(), 10_000L);
+ assertNotSame(freshChannel, suspectChannel);
+ assertSame(entry._channel, freshChannel);
+ assertTrue(freshChannel.isActive());
+ TestUtils.waitForCondition(aVoid -> !suspectChannel.isActive(), 10L,
5_000L, "Suspect channel was not closed");
+ verify(nextResponse, timeout(5_000L)).markRequestSent(eq(instance),
anyInt());
+ } finally {
+ serverChannels.shutDown();
+ }
+ }
+
+ /// Unanswered queries are judged a grace period after their timeout: a
response in between clears them, a server
+ /// that stays silent for the threshold gets the channel marked suspect and
is reported through the router.
+ @Test
+ public void testUnansweredQueriesAreJudgedAfterTheGracePeriod()
+ throws Exception {
+ QueryRouter queryRouter = mock(QueryRouter.class);
+ long graceMs = 200L;
+ ServerChannels serverChannels =
+ new ServerChannels(queryRouter, null, null,
ThreadAccountantUtils.getNoOpAccountant(), 2, graceMs);
+ try (StalledServer server = new StalledServer()) {
+ ServerRoutingInstance instance = new ServerRoutingInstance("localhost",
server.getPort(), TableType.REALTIME);
+ long suspectsBefore =
meterCount(BrokerMeter.NETTY_CONNECTION_SUSPECT_CHANNELS, instance);
+ // Unknown channel: nothing to judge
+ serverChannels.recordUnansweredQuery(instance,
System.currentTimeMillis(), null);
+ serverChannels.sendRequest("testTable", mock(AsyncQueryResponse.class),
instance, smallInstanceRequest(),
+ 10_000L);
+ ServerChannels.ServerChannel entry =
serverChannels.getOrCreateServerChannel(instance);
+ Thread.sleep(5L);
+
+ // A query written after the last response, still unanswered a grace
period after its timeout: counts
+ serverChannels.recordUnansweredQuery(instance,
System.currentTimeMillis(), null);
+ assertEquals(serverChannels.getUnansweredQueries(instance), 0, "Not
judged before the grace period");
+ TestUtils.waitForCondition(aVoid ->
serverChannels.getUnansweredQueries(instance) == 1, 10L, 5_000L,
+ "The unanswered query was not counted after the grace period");
+ assertFalse(entry._suspect);
+ verify(queryRouter, never()).reportUnansweredServer(any());
+ assertEquals(meterCount(BrokerMeter.NETTY_CONNECTION_SUSPECT_CHANNELS,
instance), suspectsBefore);
+
+ // A response from the server, even to some other query, clears the
count ...
+ serverChannels.recordResponse(instance);
+ assertEquals(serverChannels.getUnansweredQueries(instance), 0);
+ // ... and a query the server answered within the grace period does not
count
+ long submitTimeMs = System.currentTimeMillis();
+ Thread.sleep(5L);
+ serverChannels.recordResponse(instance);
+ serverChannels.recordUnansweredQuery(instance, submitTimeMs, null);
+ Thread.sleep(graceMs + 200L);
+ assertEquals(serverChannels.getUnansweredQueries(instance), 0);
+
+ // Silence for the threshold marks the channel suspect and reports the
server
+ Thread.sleep(5L);
+ serverChannels.recordUnansweredQuery(instance,
System.currentTimeMillis(), null);
+ serverChannels.recordUnansweredQuery(instance,
System.currentTimeMillis(), null);
+ verify(queryRouter, timeout(5_000L)).reportUnansweredServer(instance);
+ assertTrue(entry._suspect);
+ assertEquals(serverChannels.getUnansweredQueries(instance), 0, "Reset
once reported");
+ assertEquals(meterCount(BrokerMeter.NETTY_CONNECTION_SUSPECT_CHANNELS,
instance), suspectsBefore + 1);
+ } finally {
+ serverChannels.shutDown();
+ }
+ }
+
+ /// Under a dead server at high QPS the judgments queued per channel are
capped; the ones that do queue are enough
+ /// to cross the threshold, so the outcome is the same with a bounded queue.
+ @Test
+ public void testPendingJudgmentsPerChannelAreCapped()
+ throws Exception {
+ QueryRouter queryRouter = mock(QueryRouter.class);
+ int threshold = 3;
+ ServerChannels serverChannels =
+ new ServerChannels(queryRouter, null, null,
ThreadAccountantUtils.getNoOpAccountant(), threshold, 300L);
+ try (StalledServer server = new StalledServer()) {
+ ServerRoutingInstance instance = new ServerRoutingInstance("localhost",
server.getPort(), TableType.REALTIME);
+ serverChannels.sendRequest("testTable", mock(AsyncQueryResponse.class),
instance, smallInstanceRequest(),
+ 10_000L);
+ Thread.sleep(5L);
+ long submitTimeMs = System.currentTimeMillis();
+ for (int i = 0; i < 1_000; i++) {
+ serverChannels.recordUnansweredQuery(instance, submitTimeMs, null);
+ }
+ int numPendingJudgments =
serverChannels.getNumPendingJudgments(instance);
+ assertTrue(numPendingJudgments >= threshold && numPendingJudgments <= 16,
+ "Expected between the threshold and the cap of pending judgments,
got " + numPendingJudgments);
+ verify(queryRouter, timeout(5_000L)).reportUnansweredServer(instance);
+ TestUtils.waitForCondition(aVoid ->
serverChannels.getNumPendingJudgments(instance) == 0, 10L, 5_000L,
+ "Pending judgments were not drained");
+ } finally {
+ serverChannels.shutDown();
+ }
+ }
+
+ /// With the check disabled nothing is scheduled or counted.
+ @Test
+ public void testUnansweredQueryCheckDisabled()
+ throws Exception {
+ QueryRouter queryRouter = mock(QueryRouter.class);
+ ServerChannels serverChannels =
+ new ServerChannels(queryRouter, null, null,
ThreadAccountantUtils.getNoOpAccountant(), 0, 10L);
+ try (StalledServer server = new StalledServer()) {
+ ServerRoutingInstance instance = new ServerRoutingInstance("localhost",
server.getPort(), TableType.REALTIME);
+ serverChannels.sendRequest("testTable", mock(AsyncQueryResponse.class),
instance, smallInstanceRequest(),
+ 10_000L);
+ Thread.sleep(5L);
+ serverChannels.recordUnansweredQuery(instance,
System.currentTimeMillis(), null);
+ Thread.sleep(200L);
+ assertEquals(serverChannels.getUnansweredQueries(instance), 0);
+ assertFalse(serverChannels.getOrCreateServerChannel(instance)._suspect);
+ verify(queryRouter, never()).reportUnansweredServer(any());
+ } finally {
+ serverChannels.shutDown();
+ }
+ }
+
+ /// The failure detector's probe must not vouch for a channel the server
stopped answering on just because the
+ /// socket still reports active: a suspect channel is replaced, a healthy
one kept, a closed one reconnected, and
+ /// an unreachable server fails the probe.
+ @Test
+ public void testProbeReplacesSuspectChannelAndKeepsHealthyOne()
+ throws Exception {
+ ServerChannels serverChannels =
+ new ServerChannels(mock(QueryRouter.class), null, null,
ThreadAccountantUtils.getNoOpAccountant());
+ StalledServer server = new StalledServer();
+ try {
+ ServerRoutingInstance instance = new ServerRoutingInstance("localhost",
server.getPort(), TableType.REALTIME);
+ // Nothing to vouch for before a query was sent, even once pre-connect
opened a channel
+ assertFalse(serverChannels.probe(instance));
+ serverChannels.preConnect(instance, 5_000L);
+ assertFalse(serverChannels.probe(instance), "Pre-connected only: not
query-carrying");
+
+ serverChannels.sendRequest("testTable", mock(AsyncQueryResponse.class),
instance, smallInstanceRequest(),
+ 10_000L);
+ ServerChannels.ServerChannel entry =
serverChannels.getOrCreateServerChannel(instance);
+ Channel healthyChannel = entry._channel;
+ assertTrue(serverChannels.probe(instance));
+ assertSame(entry._channel, healthyChannel, "A healthy channel is kept");
+ assertTrue(healthyChannel.isActive());
+
+ serverChannels.markSuspect(instance);
+ assertTrue(serverChannels.probe(instance));
+ Channel freshChannel = entry._channel;
+ assertNotSame(freshChannel, healthyChannel, "A suspect channel is
replaced");
+ assertTrue(freshChannel.isActive());
+ TestUtils.waitForCondition(aVoid -> !healthyChannel.isActive(), 10L,
5_000L, "Suspect channel was not closed");
+ // No longer suspect: the replacement is kept by the next probe
+ assertTrue(serverChannels.probe(instance));
+ assertSame(entry._channel, freshChannel);
+
+ // A closed channel is reconnected, as the probe always did
+ freshChannel.close().sync();
+ assertTrue(serverChannels.probe(instance));
+ assertNotSame(entry._channel, freshChannel);
+ assertTrue(entry._channel.isActive());
+
+ // An unreachable server fails the probe and leaves no channel behind
+ serverChannels.markSuspect(instance);
+ server.close();
+ assertThrows(Exception.class, () -> serverChannels.probe(instance));
+ assertNull(entry._channel);
+ } finally {
+ server.close();
+ serverChannels.shutDown();
+ }
+ }
+
+ /// A query written to a channel the entry has since replaced says nothing
about the new channel: its judgment does
+ /// not count, even when it runs after the replacement, and so cannot mark
the fresh channel suspect.
+ @Test
+ public void testJudgmentOfAQueryOnAReplacedChannelDoesNotCount()
+ throws Exception {
+ QueryRouter queryRouter = mock(QueryRouter.class);
+ ServerChannels serverChannels =
+ new ServerChannels(queryRouter, null, null,
ThreadAccountantUtils.getNoOpAccountant(), 1, 100L);
+ try (StalledServer server = new StalledServer()) {
+ ServerRoutingInstance instance = new ServerRoutingInstance("localhost",
server.getPort(), TableType.REALTIME);
+ Channel oldChannel = serverChannels.sendRequest("testTable",
mock(AsyncQueryResponse.class), instance,
+ smallInstanceRequest(), 10_000L);
+ long oldSubmitTimeMs = System.currentTimeMillis();
+ serverChannels.markSuspect(instance);
+ Channel freshChannel = serverChannels.sendRequest("testTable",
mock(AsyncQueryResponse.class), instance,
+ smallInstanceRequest(), 10_000L);
+ assertNotSame(freshChannel, oldChannel);
+
+ // With a threshold of 1 a counted judgment would mark the fresh channel
suspect at once
+ serverChannels.recordUnansweredQuery(instance, oldSubmitTimeMs,
oldChannel);
+ TestUtils.waitForCondition(aVoid ->
serverChannels.getNumPendingJudgments(instance) == 0, 10L, 5_000L,
+ "The judgment did not run");
+ assertEquals(serverChannels.getUnansweredQueries(instance), 0);
+ assertFalse(serverChannels.getOrCreateServerChannel(instance)._suspect);
+ verify(queryRouter, never()).reportUnansweredServer(any());
+
+ // A query on the current channel does count
+ Thread.sleep(5L);
+ serverChannels.recordUnansweredQuery(instance,
System.currentTimeMillis(), freshChannel);
+ verify(queryRouter, timeout(5_000L)).reportUnansweredServer(instance);
+ assertTrue(serverChannels.getOrCreateServerChannel(instance)._suspect);
+ } finally {
+ serverChannels.shutDown();
+ }
+ }
+
private static long meterCount(BrokerMeter meter) {
return BrokerMetrics.get().getMeteredValue(meter).count();
}
@@ -1324,9 +1548,22 @@ public class ServerChannelsTest {
serverChannel._bootstrap.option(ChannelOption.SO_SNDBUF,
SOCKET_BUFFER_BYTES);
}
+ /// Returns once the port refuses connections. On Linux a socket closed
while the acceptor is blocked in
+ /// `accept()` keeps listening until that call returns, so a connect right
after closing could still succeed.
@Override
public void close() {
IOUtils.closeQuietly(_serverSocket);
+ closeAccepted();
+ try {
+ _acceptor.join(5_000L);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ // Anything the acceptor took in before it returned
+ closeAccepted();
+ }
+
+ private void closeAccepted() {
for (Socket socket : _accepted) {
IOUtils.closeQuietly(socket);
}
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
index 45c947fad6b..271ce143163 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
@@ -1414,6 +1414,33 @@ public class CommonConstants {
/// 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. That never
takes the server out of routing; see
+ /// [#CONFIG_OF_REPORT_UNANSWERED_SERVERS] for that. Any response from
the server, decodable or not, 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;
+ public static final String CONFIG_OF_UNANSWERED_QUERY_GRACE_MS =
+ "pinot.broker.failure.detector.unanswered.query.grace.ms";
+ public static final long DEFAULT_UNANSWERED_QUERY_GRACE_MS = 2_000L;
+
+ /// With pings disabled, whether a server that crosses
[#CONFIG_OF_UNANSWERED_QUERIES_THRESHOLD] on a channel is
+ /// also reported to the failure detector, which with the `CONNECTION`
detector takes it out of routing until the
+ /// retry probe re-establishes its channels. Off by default: a tenant
whose heavy queries routinely overrun their
+ /// timeout, or a long GC pause, can cross the threshold on a healthy
server. Without pings the retry probe is a
+ /// TCP connect, which a frozen JVM's kernel still completes, so such a
server is readmitted at every retry and
+ /// can leave up to the threshold of queries per channel unanswered
again before it is reported once more;
+ /// enabling pings, where every server supports them, is the better way
to catch it. With pings enabled this
+ /// setting has no effect: the ping that follows every timeout decides.
+ public static final String CONFIG_OF_REPORT_UNANSWERED_SERVERS =
+ "pinot.broker.failure.detector.report.unanswered.servers";
+ public static final boolean DEFAULT_REPORT_UNANSWERED_SERVERS = false;
}
// Configs related to AdaptiveServerSelection.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
