This is an automated email from the ASF dual-hosted git repository.
yashmayya 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 aea5680a40b Detect servers that stop answering queries but keep their
connection open (#19452)
aea5680a40b is described below
commit aea5680a40b15a03414876bcad7c5a15ce95eef4
Author: Yash Mayya <[email protected]>
AuthorDate: Sat Oct 24 15:24:11 2026 -0700
Detect servers that stop answering queries but keep their connection open
(#19452)
---
.../SingleConnectionBrokerRequestHandler.java | 73 +++-
...ionBrokerRequestHandlerFailureDetectorTest.java | 450 +++++++++++++++++++++
...BaseExponentialBackoffRetryFailureDetector.java | 132 +++++-
.../failuredetector/ConnectionFailureDetector.java | 9 +-
.../common/failuredetector/FailureDetector.java | 23 ++
.../pinot/common/request/InstanceRequest.java | 106 ++++-
.../ConnectionFailureDetectorTest.java | 284 ++++++++++++-
pinot-common/src/thrift/request.thrift | 4 +
.../core/transport/InstanceRequestHandler.java | 19 +
.../apache/pinot/core/transport/QueryRouter.java | 149 ++++++-
.../pinot/core/transport/ServerChannels.java | 47 ++-
.../core/transport/InstanceRequestHandlerTest.java | 56 +++
.../pinot/core/transport/QueryRoutingTest.java | 270 +++++++++++++
.../pinot/core/transport/QueryServerTestUtils.java | 69 ++++
.../pinot/core/transport/ServerChannelsTest.java | 39 +-
.../apache/pinot/spi/utils/CommonConstants.java | 20 +-
16 files changed, 1693 insertions(+), 57 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 ea9364d8af4..cceced717cc 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
@@ -19,6 +19,7 @@
package org.apache.pinot.broker.requesthandler;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.util.ArrayList;
@@ -113,6 +114,8 @@ public class SingleConnectionBrokerRequestHandler extends
BaseSingleStageBrokerR
protected final BrokerReduceService _brokerReduceService;
protected final QueryRouter _queryRouter;
protected final FailureDetector _failureDetector;
+ private final boolean _pingOnTimeout;
+ private final long _pingTimeoutMs;
/// Legacy constructor without an MV handler — see
[BaseSingleStageBrokerRequestHandler]'s
/// legacy ctor for the rationale. Delegates with `materializedViewHandler
= null`.
@@ -139,7 +142,16 @@ public class SingleConnectionBrokerRequestHandler extends
BaseSingleStageBrokerR
_brokerReduceService = new BrokerReduceService(_config);
_queryRouter = new QueryRouter(_brokerId, nettyConfig, tlsConfig,
serverRoutingStatsManager, threadAccountant);
_failureDetector = failureDetector;
+ _pingOnTimeout =
_config.getProperty(CommonConstants.Broker.FailureDetector.CONFIG_OF_ENABLE_PING_ON_TIMEOUT,
+ CommonConstants.Broker.FailureDetector.DEFAULT_ENABLE_PING_ON_TIMEOUT);
+ _pingTimeoutMs =
_config.getProperty(CommonConstants.Broker.FailureDetector.CONFIG_OF_PING_TIMEOUT_MS,
+ CommonConstants.Broker.FailureDetector.DEFAULT_PING_TIMEOUT_MS);
_failureDetector.registerUnhealthyServerRetrier(this::retryUnhealthyServer);
+ if (_pingOnTimeout) {
+ Preconditions.checkArgument(_pingTimeoutMs > 0, "'%s' must be positive,
got: %s",
+ CommonConstants.Broker.FailureDetector.CONFIG_OF_PING_TIMEOUT_MS,
_pingTimeoutMs);
+
_failureDetector.registerServerNotRespondedChecker(this::checkServerNotResponded);
+ }
}
@Override
@@ -739,11 +751,34 @@ public class SingleConnectionBrokerRequestHandler extends
BaseSingleStageBrokerR
serversNotResponded.add(entry.getKey());
}
}
+ reportServersNotResponded(asyncQueryResponse, finalResponses);
+
ScatterResultStats stats = new ScatterResultStats(
dataTableMap.size() + serversNotResponded.size(), dataTableMap.size(),
totalResponseSize);
return new ScatterResult(dataTableMap, serversNotResponded, stats,
timedOut, asyncQueryResponse.getException());
}
+ /// Reports the servers that left a timed-out query unanswered to the
failure detector, which, with pings enabled,
+ /// pings each of them and marks the ones that do not answer unhealthy. This
is how a server whose node is gone or
+ /// whose JVM is frozen is detected at all -- see
[FailureDetector#notifyServerNotResponded].
+ ///
+ /// Nothing is reported unless the query timed out: a `FAILED` response was
aborted early by a send failure or a
+ /// channel teardown, both already reported through
[AsyncQueryResponse#getFailedServer()], and aborting leaves
+ /// servers that were never given a chance to answer without a response. For
the same reason the split
+ /// materialized-view path reports each of its two responses on its own: one
can time out while the other aborts.
+ private void reportServersNotResponded(AsyncQueryResponse asyncQueryResponse,
+ Map<ServerRoutingInstance, ServerResponse> finalResponses) {
+ if (asyncQueryResponse.getStatus() == QueryResponse.Status.TIMED_OUT) {
+ for (Map.Entry<ServerRoutingInstance, ServerResponse> entry :
finalResponses.entrySet()) {
+ if (entry.getValue().getDataTable() == null) {
+ ServerRoutingInstance serverRoutingInstance = entry.getKey();
+
_failureDetector.notifyServerNotResponded(serverRoutingInstance.getInstanceId(),
+ serverRoutingInstance.getHostname());
+ }
+ }
+ }
+ }
+
/// Executes the reduce step on the scatter result and populates the
response with server stats.
/// Subclasses may override to perform custom reduce logic, or construct a
[ScatterResult]
/// with a substituted data table map using [ScatterResultStats] to preserve
server stats.
@@ -855,6 +890,9 @@ public class SingleConnectionBrokerRequestHandler extends
BaseSingleStageBrokerR
long totalResponseSize = totalResponseSizeHolder[0];
int numServersResponded = dataTableMap.size();
+ reportServersNotResponded(baseAsyncResponse, baseFinalResponses);
+ reportServersNotResponded(materializedViewAsyncResponse,
viewFinalResponses);
+
/// On a SPLIT query, base and MV scatter-gathers cover DISJOINT halves of
the timeline
/// (base covers `ts < boundary`, MV covers `ts >= boundary`). Partial
failure on either
/// side therefore leaves whole time-ranges missing — semantically
distinct from a non-split
@@ -1004,7 +1042,8 @@ public class SingleConnectionBrokerRequestHandler extends
BaseSingleStageBrokerR
}
}
- /// Check if a server that was previously detected as unhealthy is now
healthy.
+ /// 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.
public FailureDetector.ServerState retryUnhealthyServer(String instanceId) {
LOGGER.info("Retrying unhealthy server: {}", instanceId);
ServerInstance serverInstance =
_routingManager.getEnabledServerInstanceMap().get(instanceId);
@@ -1019,6 +1058,18 @@ public class SingleConnectionBrokerRequestHandler
extends BaseSingleStageBrokerR
return FailureDetector.ServerState.UNKNOWN;
}
+ if (_pingOnTimeout) {
+ // Not a connect: a frozen JVM's kernel still completes the handshake,
which would put a server taken out for an
+ // unanswered ping straight back into routing.
+ if (_queryRouter.ping(serverInstance, _pingTimeoutMs)) {
+ LOGGER.info("Server: {} answered a ping, marking it healthy",
instanceId);
+ return FailureDetector.ServerState.HEALTHY;
+ } else {
+ LOGGER.warn("Server: {} did not answer a ping within {}ms, retry
later", instanceId, _pingTimeoutMs);
+ return FailureDetector.ServerState.UNHEALTHY;
+ }
+ }
+
if (_queryRouter.connect(serverInstance)) {
LOGGER.info("Successfully connect to server: {}, marking it healthy",
instanceId);
return FailureDetector.ServerState.HEALTHY;
@@ -1028,6 +1079,26 @@ public class SingleConnectionBrokerRequestHandler
extends BaseSingleStageBrokerR
}
}
+ /// Checks a server that left a query unanswered, by pinging it. Registered
with the failure detector only when pings
+ /// are enabled, as the only check a timeout runs -- see
[FailureDetector#registerServerNotRespondedChecker].
+ private FailureDetector.ServerState checkServerNotResponded(String
instanceId) {
+ ServerInstance serverInstance =
_routingManager.getEnabledServerInstanceMap().get(instanceId);
+ if (serverInstance == null) {
+ // Not routed by this broker -- a server of another cluster, say, or one
Helix has just disabled -- so there is
+ // nothing to take out of routing.
+ return FailureDetector.ServerState.UNKNOWN;
+ }
+ if (!_queryRouter.hasChannel(serverInstance)) {
+ // Nothing to ping over, which says nothing about the server.
+ return FailureDetector.ServerState.UNKNOWN;
+ }
+ if (_queryRouter.ping(serverInstance, _pingTimeoutMs)) {
+ LOGGER.debug("Server: {} left a query unanswered but answered a ping",
instanceId);
+ return FailureDetector.ServerState.HEALTHY;
+ }
+ return FailureDetector.ServerState.UNHEALTHY;
+ }
+
/// Counts responses that successfully returned a DataTable. Production
callers use this in the
/// MV-split path to detect "all of one side's servers failed" — if
`viewFinalResponses` is
/// non-empty AND the count is zero, the split would silently undercount the
historical half
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
new file mode 100644
index 00000000000..7574bc25d94
--- /dev/null
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/SingleConnectionBrokerRequestHandlerFailureDetectorTest.java
@@ -0,0 +1,450 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.broker.requesthandler;
+
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.SettableFuture;
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.pinot.broker.broker.AllowAllAccessControlFactory;
+import org.apache.pinot.broker.queryquota.QueryQuotaManager;
+import org.apache.pinot.broker.routing.manager.BrokerRoutingManager;
+import org.apache.pinot.common.config.provider.TableCache;
+import org.apache.pinot.common.datatable.DataTable;
+import org.apache.pinot.common.datatable.DataTable.MetadataKey;
+import org.apache.pinot.common.failuredetector.FailureDetector;
+import org.apache.pinot.common.failuredetector.FailureDetectorFactory;
+import org.apache.pinot.common.metrics.BrokerGauge;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.common.metrics.MetricValueUtils;
+import org.apache.pinot.common.request.BrokerRequest;
+import org.apache.pinot.core.common.datatable.DataTableBuilderFactory;
+import org.apache.pinot.core.routing.ImplicitHybridTableRouteInfo;
+import org.apache.pinot.core.routing.MultiClusterRoutingContext;
+import org.apache.pinot.core.routing.SegmentsToQuery;
+import org.apache.pinot.core.routing.TableRouteInfo;
+import org.apache.pinot.core.transport.QueryServer;
+import org.apache.pinot.core.transport.QueryServerTestUtils;
+import org.apache.pinot.core.transport.ServerInstance;
+import
org.apache.pinot.core.transport.server.routing.stats.ServerRoutingStatsManager;
+import org.apache.pinot.spi.accounting.ThreadAccountantUtils;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import
org.apache.pinot.spi.eventlistener.query.BrokerQueryEventListenerFactory;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.apache.pinot.sql.parsers.CalciteSqlCompiler;
+import org.apache.pinot.util.TestUtils;
+import org.mockito.stubbing.Answer;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+
+
+/// End-to-end coverage, over real sockets, of the broker removing a server
whose node is gone or whose JVM is frozen.
+///
+/// Such a server keeps its TCP connection open and never answers, so it
produces query timeouts and nothing else: no
+/// send exception, no channel teardown. Before this change the broker kept
routing queries to it until Helix noticed,
+/// minutes later, while healthy replicas sat idle. Now a timed-out query
makes the broker ping the server, and a server
+/// that does not answer the ping is taken out of routing.
+///
+/// These tests drive the real [SingleConnectionBrokerRequestHandler] and a
real [FailureDetector] against real
+/// servers, so the whole chain -- timeout, ping, routing exclusion, retry --
runs over real sockets.
+public class SingleConnectionBrokerRequestHandlerFailureDetectorTest {
+ private static final BrokerRequest BROKER_REQUEST =
+ CalciteSqlCompiler.compileToBrokerRequest("SELECT * FROM testTable");
+ private static final String RAW_TABLE_NAME = "testTable";
+ private static final long REQUEST_ID = 123L;
+ private static final long QUERY_TIMEOUT_MS = 1_000L;
+ private static final long PING_TIMEOUT_MS = 500L;
+ /// Long enough that the retry loop never runs during a test that is not
about it.
+ private static final long RETRIES_OFF_MS = 600_000L;
+ private static final long FAST_RETRY_DELAY_MS = 100L;
+
+ private final List<QueryServer> _queryServers = new ArrayList<>();
+ private final List<ServerSocket> _frozenServers = new ArrayList<>();
+ /// What the routing manager reports as enabled. The ping-based check treats
a server missing from here as
+ /// unhealthy, so every server a test starts must be in it.
+ private final Map<String, ServerInstance> _enabledServers = new
ConcurrentHashMap<>();
+
+ private BrokerRoutingManager _routingManager;
+ private SingleConnectionBrokerRequestHandler _requestHandler;
+ private FailureDetector _failureDetector;
+ private List<String> _excludedServers;
+ private List<String> _reincludedServers;
+
+ @BeforeMethod
+ public void setUp() {
+ startBroker(true, RETRIES_OFF_MS);
+ }
+
+ /// Builds the broker-side stack under test: a real request handler wired to
a real failure detector whose notifiers
+ /// are recorded, mirroring the wiring `BaseBrokerStarter` performs in
production. Without `pingOnTimeout` the
+ /// default applies.
+ private void startBroker(boolean pingOnTimeout, long retryInitialDelayMs) {
+ // 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();
+ 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);
+ }
+ config.setProperty(Broker.FailureDetector.CONFIG_OF_PING_TIMEOUT_MS,
PING_TIMEOUT_MS);
+
config.setProperty(Broker.FailureDetector.CONFIG_OF_RETRY_INITIAL_DELAY_MS,
retryInitialDelayMs);
+ config.setProperty(Broker.FailureDetector.CONFIG_OF_RETRY_DELAY_FACTOR, 1);
+ // Past the max retries a server is put back regardless, which would let a
recovery test pass with no ping answered.
+ config.setProperty(Broker.FailureDetector.CONFIG_OF_MAX_RETRIES,
Integer.MAX_VALUE);
+
+ _failureDetector = FailureDetectorFactory.getFailureDetector(config,
BrokerMetrics.get());
+ _excludedServers = new CopyOnWriteArrayList<>();
+ _reincludedServers = new CopyOnWriteArrayList<>();
+ _failureDetector.registerUnhealthyServerNotifier(instanceId ->
_excludedServers.add(instanceId));
+ _failureDetector.registerHealthyServerNotifier(instanceId ->
_reincludedServers.add(instanceId));
+ _failureDetector.start();
+
+ BrokerQueryEventListenerFactory.init(new PinotConfiguration());
+ ServerRoutingStatsManager serverRoutingStatsManager =
+ new ServerRoutingStatsManager(new PinotConfiguration(),
BrokerMetrics.get());
+ serverRoutingStatsManager.init();
+
+ _routingManager = mock(BrokerRoutingManager.class);
+
when(_routingManager.getEnabledServerInstanceMap()).thenReturn(_enabledServers);
+ _requestHandler =
+ new SingleConnectionBrokerRequestHandler(config, "testBroker", new
BrokerRequestIdGenerator(), _routingManager,
+ new AllowAllAccessControlFactory(), mock(QueryQuotaManager.class),
mock(TableCache.class), null, null,
+ serverRoutingStatsManager, _failureDetector,
ThreadAccountantUtils.getNoOpAccountant(),
+ mock(MultiClusterRoutingContext.class));
+ }
+
+ @AfterMethod
+ public void tearDown()
+ throws IOException {
+ stopBroker();
+ for (QueryServer queryServer : _queryServers) {
+ queryServer.shutDown();
+ }
+ _queryServers.clear();
+ for (ServerSocket frozenServer : _frozenServers) {
+ frozenServer.close();
+ }
+ _frozenServers.clear();
+ _enabledServers.clear();
+ }
+
+ private void stopBroker() {
+ if (_requestHandler != null) {
+ _requestHandler.shutDown();
+ _requestHandler = null;
+ }
+ if (_failureDetector != null) {
+ _failureDetector.stop();
+ _failureDetector = null;
+ }
+ // The gauge is a JVM-wide singleton; do not leave it dirty for the next
test.
+ BrokerMetrics.get().setValueOfGlobalGauge(BrokerGauge.UNHEALTHY_SERVERS,
0);
+ }
+
+ //
---------------------------------------------------------------------------------------------------------------
+ // Fixtures
+ //
---------------------------------------------------------------------------------------------------------------
+
+ /// Starts a server whose node is gone or whose JVM is frozen: a listening
socket that is never accepted from. The
+ /// kernel completes the TCP handshake and buffers whatever is sent, but
nothing ever reads it or replies -- neither
+ /// queries nor pings. Shutting a server down would NOT model this: that
closes the socket, which takes the
+ /// already-working connection-failure path instead.
+ private ServerInstance startFrozenServer()
+ throws IOException {
+ ServerSocket frozenServer = new ServerSocket(0, 100);
+ _frozenServers.add(frozenServer);
+ return enable(QueryServerTestUtils.serverInstance("localhost",
frozenServer.getLocalPort()));
+ }
+
+ /// Starts a server whose JVM is alive but whose query engine never finishes
anything -- stuck, or so busy that every
+ /// query times out. Its network thread still answers pings, because they do
not go through the query engine.
+ private ServerInstance startStuckQueryEngineServer() {
+ return startQueryServer(0, invocation -> SettableFuture.<byte[]>create());
+ }
+
+ /// Starts a server that answers queries straight away.
+ private ServerInstance startRespondingServer(int port) {
+ DataTable dataTable = DataTableBuilderFactory.getEmptyDataTable();
+ dataTable.getMetadata().put(MetadataKey.REQUEST_ID.getName(),
Long.toString(REQUEST_ID));
+ byte[] responseBytes;
+ try {
+ responseBytes = dataTable.toBytes();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ return startQueryServer(port, invocation ->
Futures.immediateFuture(responseBytes));
+ }
+
+ private ServerInstance startQueryServer(int port, Answer<?> submitAnswer) {
+ QueryServer queryServer = QueryServerTestUtils.newQueryServer(port,
submitAnswer);
+ _queryServers.add(queryServer);
+ return enable(QueryServerTestUtils.serverInstance("localhost",
QueryServerTestUtils.startAndGetPort(queryServer)));
+ }
+
+ private ServerInstance enable(ServerInstance serverInstance) {
+ _enabledServers.put(serverInstance.getInstanceId(), serverInstance);
+ return serverInstance;
+ }
+
+ /// Routes a query to the given servers for the given table types; both for
a hybrid table.
+ private static TableRouteInfo routeTo(boolean offline, boolean realtime,
ServerInstance... servers) {
+ Map<ServerInstance, SegmentsToQuery> routingTable = new HashMap<>();
+ for (ServerInstance server : servers) {
+ routingTable.put(server, new SegmentsToQuery(List.of("segment0"),
List.of()));
+ }
+ return new ImplicitHybridTableRouteInfo(offline ? BROKER_REQUEST : null,
realtime ? BROKER_REQUEST : null,
+ offline ? routingTable : null, realtime ? routingTable : null);
+ }
+
+ private static TableRouteInfo offlineRouteTo(ServerInstance... servers) {
+ return routeTo(true, false, servers);
+ }
+
+ private SingleConnectionBrokerRequestHandler.ScatterResult
scatter(TableRouteInfo route)
+ throws Exception {
+ return _requestHandler.doScatter(REQUEST_ID, RAW_TABLE_NAME, route,
QUERY_TIMEOUT_MS,
+ new BaseSingleStageBrokerRequestHandler.ServerStats());
+ }
+
+ /// Waits until exactly `servers` are unhealthy, gauge included. Waits on
that final state rather than on the
+ /// notifier, which runs before the detector records the change.
+ private void awaitUnhealthyServers(ServerInstance... servers) {
+ Set<String> expected = new HashSet<>();
+ for (ServerInstance server : servers) {
+ expected.add(server.getInstanceId());
+ }
+ TestUtils.waitForCondition(
+ aVoid -> _failureDetector.getUnhealthyServers().equals(expected) &&
unhealthyServerGauge() == expected.size(),
+ 10_000L, "Unhealthy servers did not become: " + expected);
+ }
+
+ private static int unhealthyServerGauge() {
+ return (int) MetricValueUtils.getGlobalGaugeValue(BrokerMetrics.get(),
BrokerGauge.UNHEALTHY_SERVERS);
+ }
+
+ /// Waits until the check of `server` after a timeout has finished, not
merely started, given the count of its
+ /// lookups. A server is checked again only once its previous check is over,
so a second check proves the first
+ /// finished. Stops early if the server got marked unhealthy, which ends its
checks, and leaves that to the caller's
+ /// assertions.
+ private void awaitCheckFinished(ServerInstance server, AtomicInteger
numLookups) {
+ TestUtils.waitForCondition(aVoid -> {
+ _failureDetector.notifyServerNotResponded(server.getInstanceId(),
server.getHostname());
+ return numLookups.get() >= 2 ||
!_failureDetector.getUnhealthyServers().isEmpty();
+ }, 10_000L, "Check did not finish");
+ }
+
+ /// Counts lookups of enabled servers. The handler looks a server up once
per check or retry, so this counts those.
+ private AtomicInteger countServerLookups() {
+ AtomicInteger numLookups = new AtomicInteger();
+ when(_routingManager.getEnabledServerInstanceMap()).thenAnswer(invocation
-> {
+ numLookups.incrementAndGet();
+ return _enabledServers;
+ });
+ return numLookups;
+ }
+
+ //
---------------------------------------------------------------------------------------------------------------
+ // Tests
+ //
---------------------------------------------------------------------------------------------------------------
+
+ /// THE INCIDENT. A query to a server whose node is gone times out; the
broker pings the server, the ping goes
+ /// unanswered too, and the server is taken out of routing -- about one
query timeout plus one ping timeout after the
+ /// first failure, instead of minutes later.
+ ///
+ /// Covered for every table type: the broker keeps a separate channel per
table type, and the incident's tables were
+ /// REALTIME. Checking only the OFFLINE channel would never have pinged this
server at all.
+ @Test(dataProvider = "tableTypes")
+ public void testServerWhoseNodeIsGoneIsTakenOutOfRouting(boolean offline,
boolean realtime)
+ throws Exception {
+ ServerInstance frozenServer = startFrozenServer();
+
+ long startTimeMs = System.currentTimeMillis();
+ SingleConnectionBrokerRequestHandler.ScatterResult result =
scatter(routeTo(offline, realtime, frozenServer));
+ assertTrue(result.isTimedOut());
+ awaitUnhealthyServers(frozenServer);
+ long elapsedMs = System.currentTimeMillis() - startTimeMs;
+
+ assertEquals(_excludedServers, List.of(frozenServer.getInstanceId()));
+ assertTrue(elapsedMs < QUERY_TIMEOUT_MS + PING_TIMEOUT_MS + 3_000L,
+ "Server should be out of routing about one query timeout plus one ping
timeout in; took " + elapsedMs + "ms");
+ }
+
+ @DataProvider(name = "tableTypes")
+ public Object[][] tableTypes() {
+ return new Object[][]{{true, false}, {false, true}, {true, true}};
+ }
+
+ /// A server whose JVM is alive but whose queries time out -- stuck or
merely busy -- answers the ping and stays in
+ /// routing. This is the control for the test above: the same timeout and
the same check, but here the ping is
+ /// answered. It also pins the accepted gap: a live JVM with a stuck query
engine is not detected.
+ ///
+ /// A retrier that calls every server unhealthy is registered too, as the
multi-stage one does for an idle gRPC
+ /// channel: only the ping may decide whether a server that left a query
unanswered leaves routing.
+ @Test
+ public void testServerWhoseQueryEngineIsStuckStaysInRouting()
+ throws Exception {
+ ServerInstance stuckServer = startStuckQueryEngineServer();
+ _failureDetector.registerUnhealthyServerRetrier(instanceId ->
FailureDetector.ServerState.UNHEALTHY);
+ AtomicInteger numChecks = countServerLookups();
+
+ assertTrue(scatter(offlineRouteTo(stuckServer)).isTimedOut());
+ awaitCheckFinished(stuckServer, numChecks);
+
+ assertEquals(_failureDetector.getUnhealthyServers(), Set.of());
+ assertEquals(_excludedServers, List.of());
+ }
+
+ /// A server outside this broker's routing -- another cluster's, say, or one
Helix has just disabled -- is left
+ /// alone after a timeout: there is nothing to take out of routing.
+ @Test
+ public void testServerOutsideThisBrokersRoutingIsLeftAlone()
+ throws Exception {
+ ServerInstance frozenServer = startFrozenServer();
+ _enabledServers.remove(frozenServer.getInstanceId());
+ AtomicInteger numChecks = countServerLookups();
+
+ assertTrue(scatter(offlineRouteTo(frozenServer)).isTimedOut());
+ awaitCheckFinished(frozenServer, numChecks);
+
+ assertEquals(_failureDetector.getUnhealthyServers(), Set.of());
+ assertEquals(_excludedServers, List.of());
+ }
+
+ /// A server the broker has no query channel to has nothing to ping over,
which says nothing about the server, so it
+ /// is left alone too.
+ @Test
+ public void testServerWithoutAQueryChannelIsLeftAlone() {
+ ServerInstance respondingServer = startRespondingServer(0);
+ AtomicInteger numChecks = countServerLookups();
+
+ awaitCheckFinished(respondingServer, numChecks);
+
+ assertEquals(_failureDetector.getUnhealthyServers(), Set.of());
+ assertEquals(_excludedServers, List.of());
+ }
+
+ /// Pings are off by default, which leaves a timeout without consequence, as
before this change. Old servers queue a
+ /// ping behind their queries, so a busy one could miss it; the feature is
for clusters whose servers all answer
+ /// pings.
+ @Test
+ public void testTimeoutLeavesServerInRoutingWhenPingsAreOff()
+ throws Exception {
+ stopBroker();
+ startBroker(false, RETRIES_OFF_MS);
+ ServerInstance frozenServer = startFrozenServer();
+ AtomicInteger numLookups = countServerLookups();
+
+ assertTrue(scatter(offlineRouteTo(frozenServer)).isTimedOut());
+ // A check would look the server up within milliseconds; give one far
longer than that before concluding none ran.
+ Thread.sleep(PING_TIMEOUT_MS + 500L);
+
+ assertEquals(numLookups.get(), 0);
+ assertEquals(_failureDetector.getUnhealthyServers(), Set.of());
+ assertEquals(_excludedServers, List.of());
+ }
+
+ /// 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
+ public void testRetriesConnectWhenPingsAreOff()
+ throws Exception {
+ stopBroker();
+ startBroker(false, FAST_RETRY_DELAY_MS);
+ ServerInstance frozenServer = startFrozenServer();
+ // Opens the channel a retry looks for; without one the retrier has no
opinion.
+ assertTrue(scatter(offlineRouteTo(frozenServer)).isTimedOut());
+
+ _failureDetector.markServerUnhealthy(frozenServer.getInstanceId(),
frozenServer.getHostname());
+ awaitUnhealthyServers();
+ assertEquals(_reincludedServers, List.of(frozenServer.getInstanceId()));
+ }
+
+ /// Several servers dying together are all taken out of routing, while their
healthy peer is left alone.
+ @Test
+ public void testEveryFrozenServerIsTakenOutOfRouting()
+ throws Exception {
+ ServerInstance frozenServer1 = startFrozenServer();
+ ServerInstance frozenServer2 = startFrozenServer();
+ ServerInstance respondingServer = startRespondingServer(0);
+
+ SingleConnectionBrokerRequestHandler.ScatterResult result =
+ scatter(offlineRouteTo(frozenServer1, frozenServer2,
respondingServer));
+ assertTrue(result.isTimedOut());
+ assertEquals(result.getServersNotResponded().size(), 2);
+ awaitUnhealthyServers(frozenServer1, frozenServer2);
+ }
+
+ /// A server that keeps failing its pings stays out of routing, retry after
retry. Each retry reconnects -- the
+ /// previous channel was closed when its ping failed -- and pings over the
new channel.
+ @Test
+ public void testFrozenServerStaysOutOfRoutingWhileItKeepsFailingPings()
+ throws Exception {
+ stopBroker();
+ startBroker(true, FAST_RETRY_DELAY_MS);
+ ServerInstance frozenServer = startFrozenServer();
+ AtomicInteger numChecks = countServerLookups();
+
+ scatter(offlineRouteTo(frozenServer));
+ awaitUnhealthyServers(frozenServer);
+ // The first check marked it unhealthy; wait for retries on top of it, so
this cannot pass vacuously.
+ TestUtils.waitForCondition(aVoid -> numChecks.get() >= 4, 10_000L, "Server
was not retried");
+
+ assertEquals(_failureDetector.getUnhealthyServers(),
Set.of(frozenServer.getInstanceId()));
+ assertEquals(_reincludedServers, List.of());
+ }
+
+ /// A server that comes back -- after a long GC pause, or a restart in place
-- is put back into routing by the
+ /// retrier once it answers a ping again. Its old channel was closed when
the first ping failed, so the retry has to
+ /// reconnect to reach it.
+ @Test
+ public void testServerIsPutBackOnceItAnswersAgain()
+ throws Exception {
+ stopBroker();
+ startBroker(true, FAST_RETRY_DELAY_MS);
+ ServerInstance frozenServer = startFrozenServer();
+ int port = frozenServer.getPort();
+
+ scatter(offlineRouteTo(frozenServer));
+ awaitUnhealthyServers(frozenServer);
+
+ // The server recovers on the same port.
+ _frozenServers.remove(0).close();
+ startRespondingServer(port);
+
+ awaitUnhealthyServers();
+ assertEquals(_reincludedServers, List.of(frozenServer.getInstanceId()));
+ }
+}
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/failuredetector/BaseExponentialBackoffRetryFailureDetector.java
b/pinot-common/src/main/java/org/apache/pinot/common/failuredetector/BaseExponentialBackoffRetryFailureDetector.java
index 90b6dcf33c0..a6d2245eb77 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/failuredetector/BaseExponentialBackoffRetryFailureDetector.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/failuredetector/BaseExponentialBackoffRetryFailureDetector.java
@@ -18,13 +18,17 @@
*/
package org.apache.pinot.common.failuredetector;
-import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.function.Function;
import javax.annotation.Nullable;
@@ -39,15 +43,26 @@ import org.slf4j.LoggerFactory;
/// The `BaseExponentialBackoffRetryFailureDetector` is a base failure
detector implementation that retries the
/// unhealthy servers with exponential increasing delays.
+///
+/// Every check of a server -- a retry, or a check after a query to it timed
out -- runs on a small pool rather than on
+/// the retry thread, because a check can wait out a ping timeout and one dead
server must not hold up the checks of the
+/// others. A retry runs the registered retriers; a check after a timeout runs
the registered checkers only.
@ThreadSafe
public abstract class BaseExponentialBackoffRetryFailureDetector implements
FailureDetector {
private static final Logger LOGGER =
LoggerFactory.getLogger(BaseExponentialBackoffRetryFailureDetector.class);
+ private static final int MAX_CONCURRENT_CHECKS = 16;
protected final String _name = getClass().getSimpleName();
protected final ConcurrentHashMap<String, RetryInfo>
_unhealthyServerRetryInfoMap = new ConcurrentHashMap<>();
protected final DelayQueue<RetryInfo> _retryInfoDelayQueue = new
DelayQueue<>();
+ /// Servers with a check after a timeout in flight, so a burst of timeouts
against one server checks it once.
+ protected final Set<String> _serversBeingChecked =
ConcurrentHashMap.newKeySet();
+ /// Both iterated concurrently by the checks.
+ protected final List<Function<String, ServerState>> _unhealthyServerRetriers
= new CopyOnWriteArrayList<>();
+ protected final List<Function<String, ServerState>>
_serverNotRespondedCheckers = new CopyOnWriteArrayList<>();
+ protected final ThreadPoolExecutor _checkExecutor = createCheckExecutor();
+ private final Object _unhealthyServerGaugeLock = new Object();
- protected final List<Function<String, ServerState>> _unhealthyServerRetriers
= new ArrayList<>();
protected Consumer<String> _healthyServerNotifier;
protected Consumer<String> _unhealthyServerNotifier;
protected BrokerMetrics _brokerMetrics;
@@ -77,6 +92,11 @@ public abstract class
BaseExponentialBackoffRetryFailureDetector implements Fail
_unhealthyServerRetriers.add(unhealthyServerRetrier);
}
+ @Override
+ public void registerServerNotRespondedChecker(Function<String, ServerState>
serverNotRespondedChecker) {
+ _serverNotRespondedCheckers.add(serverNotRespondedChecker);
+ }
+
@Override
public void registerHealthyServerNotifier(Consumer<String>
healthyServerNotifier) {
_healthyServerNotifier = healthyServerNotifier;
@@ -107,24 +127,7 @@ public abstract class
BaseExponentialBackoffRetryFailureDetector implements Fail
markServerHealthy(instanceId, retryInfo._hostName);
continue;
}
- LOGGER.info("Retry unhealthy server: {}", instanceId);
- boolean recovered = true;
- for (Function<String, ServerState> unhealthyServerRetrier :
_unhealthyServerRetriers) {
- ServerState serverState = unhealthyServerRetrier.apply(instanceId);
- if (serverState == ServerState.UNHEALTHY) {
- recovered = false;
- break;
- }
- }
- if (recovered) {
- markServerHealthy(instanceId, retryInfo._hostName);
- } else {
- // Update the retry info and add it back to the delay queue
- retryInfo._retryDelayNs = (long) (retryInfo._retryDelayNs *
_retryDelayFactor);
- retryInfo._retryTimeNs = System.nanoTime() +
retryInfo._retryDelayNs;
- retryInfo._numRetries++;
- _retryInfoDelayQueue.offer(retryInfo);
- }
+ _checkExecutor.execute(() -> retry(retryInfo));
} catch (Exception e) {
if (_running) {
LOGGER.error("Caught exception in the retry thread, continuing
with errors", e);
@@ -137,26 +140,96 @@ public abstract class
BaseExponentialBackoffRetryFailureDetector implements Fail
_retryThread.start();
}
+ /// Retries one unhealthy server: marks it healthy if it checks out,
otherwise requeues it with a longer delay. A
+ /// retrier that throws counts as a failed check, so the server is retried
again rather than stranded as unhealthy.
+ private void retry(RetryInfo retryInfo) {
+ String instanceId = retryInfo._instanceId;
+ LOGGER.info("Retry unhealthy server: {}", instanceId);
+ boolean recovered = false;
+ try {
+ recovered = !isReportedUnhealthy(instanceId, _unhealthyServerRetriers);
+ } catch (Exception e) {
+ LOGGER.error("Caught exception while retrying unhealthy server: {},
retrying again later", instanceId, e);
+ }
+ if (recovered) {
+ markServerHealthy(instanceId, retryInfo._hostName);
+ } else {
+ retryInfo._retryDelayNs = (long) (retryInfo._retryDelayNs *
_retryDelayFactor);
+ retryInfo._retryTimeNs = System.nanoTime() + retryInfo._retryDelayNs;
+ retryInfo._numRetries++;
+ _retryInfoDelayQueue.offer(retryInfo);
+ }
+ }
+
+ /// Returns whether any of the given retriers or checkers reports the server
unhealthy.
+ private static boolean isReportedUnhealthy(String instanceId,
List<Function<String, ServerState>> checks) {
+ for (Function<String, ServerState> check : checks) {
+ if (check.apply(instanceId) == ServerState.UNHEALTHY) {
+ return true;
+ }
+ }
+ return false;
+ }
+
@Override
public void markServerHealthy(String instanceId, @Nullable String hostName) {
_unhealthyServerRetryInfoMap.computeIfPresent(instanceId, (id, retryInfo)
-> {
LOGGER.info("Mark server: {} {} as healthy", instanceId, hostName);
- _brokerMetrics.setValueOfGlobalGauge(BrokerGauge.UNHEALTHY_SERVERS,
_unhealthyServerRetryInfoMap.size() - 1);
_healthyServerNotifier.accept(instanceId);
return null;
});
+ updateUnhealthyServerGauge();
}
@Override
public void markServerUnhealthy(String instanceId, @Nullable String
hostName) {
_unhealthyServerRetryInfoMap.computeIfAbsent(instanceId, id -> {
LOGGER.warn("Mark server: {} {} as unhealthy", instanceId, hostName);
- _brokerMetrics.setValueOfGlobalGauge(BrokerGauge.UNHEALTHY_SERVERS,
_unhealthyServerRetryInfoMap.size() + 1);
_unhealthyServerNotifier.accept(instanceId);
RetryInfo retryInfo = new RetryInfo(id, hostName);
_retryInfoDelayQueue.offer(retryInfo);
return retryInfo;
});
+ updateUnhealthyServerGauge();
+ }
+
+ /// Sets the gauge from the map's size, rather than adjusting it by one
inside the compute: two servers changing state
+ /// at once would both read the same size there and leave the gauge off by
one. Under the lock the last writer reads
+ /// a size that already includes every change made before it.
+ private void updateUnhealthyServerGauge() {
+ synchronized (_unhealthyServerGaugeLock) {
+ _brokerMetrics.setValueOfGlobalGauge(BrokerGauge.UNHEALTHY_SERVERS,
_unhealthyServerRetryInfoMap.size());
+ }
+ }
+
+ /// {@inheritDoc}
+ ///
+ /// Checks the server with the registered checkers, off the caller's thread,
and marks it unhealthy if any of them
+ /// reports it unhealthy. A server already unhealthy, or already being
checked, is left alone.
+ @Override
+ public void notifyServerNotResponded(String instanceId, @Nullable String
hostName) {
+ if (!_running || _serverNotRespondedCheckers.isEmpty() ||
_unhealthyServerRetryInfoMap.containsKey(instanceId)
+ || !_serversBeingChecked.add(instanceId)) {
+ return;
+ }
+ try {
+ _checkExecutor.execute(() -> {
+ try {
+ // Skip marking once stopping: shutdownNow() interrupts the check,
which then looks like a failed one.
+ if (isReportedUnhealthy(instanceId, _serverNotRespondedCheckers) &&
_running) {
+ LOGGER.warn("Server: {} {} left a query unanswered and failed the
follow-up check", instanceId, hostName);
+ markServerUnhealthy(instanceId, hostName);
+ }
+ } catch (Exception e) {
+ LOGGER.error("Caught exception while checking server: {} after a
query timed out", instanceId, e);
+ } finally {
+ _serversBeingChecked.remove(instanceId);
+ }
+ });
+ } catch (RejectedExecutionException e) {
+ // Only when stopping.
+ _serversBeingChecked.remove(instanceId);
+ }
}
@Override
@@ -168,6 +241,7 @@ public abstract class
BaseExponentialBackoffRetryFailureDetector implements Fail
public void stop() {
LOGGER.info("Stopping {}", _name);
_running = false;
+ _checkExecutor.shutdownNow();
try {
_retryThread.interrupt();
@@ -177,6 +251,20 @@ public abstract class
BaseExponentialBackoffRetryFailureDetector implements Fail
}
}
+ /// Bounded, with its threads created on demand and reclaimed when idle, so
an idle broker holds none.
+ private static ThreadPoolExecutor createCheckExecutor() {
+ AtomicInteger threadIndex = new AtomicInteger();
+ ThreadPoolExecutor executor =
+ new ThreadPoolExecutor(MAX_CONCURRENT_CHECKS, MAX_CONCURRENT_CHECKS,
60L, TimeUnit.SECONDS,
+ new LinkedBlockingQueue<>(), runnable -> {
+ Thread thread = new Thread(runnable, "failure-detector-check-" +
threadIndex.getAndIncrement());
+ thread.setDaemon(true);
+ return thread;
+ });
+ executor.allowCoreThreadTimeOut(true);
+ return executor;
+ }
+
/// Encapsulates the retry related information.
protected class RetryInfo implements Delayed {
final String _instanceId;
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/failuredetector/ConnectionFailureDetector.java
b/pinot-common/src/main/java/org/apache/pinot/common/failuredetector/ConnectionFailureDetector.java
index 7daeaf45324..4f046f4324d 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/failuredetector/ConnectionFailureDetector.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/failuredetector/ConnectionFailureDetector.java
@@ -21,8 +21,13 @@ package org.apache.pinot.common.failuredetector;
import javax.annotation.concurrent.ThreadSafe;
-/// The `ConnectionFailureDetector` marks failed server (connection failure)
from query response as unhealthy, and
-/// retries the unhealthy servers with exponential increasing delays.
+/// The `ConnectionFailureDetector` marks a server as unhealthy when a query
response reports it as failed
+/// (connection failure), or, with pings enabled, when a query to it times out
and the server then fails to answer a
+/// ping (see [FailureDetector#notifyServerNotResponded]). It retries the
unhealthy servers with exponentially
+/// increasing delays.
+///
+/// Only the single-stage Netty query path reports timeouts today; the gRPC
and multi-stage paths report connection
+/// failures only.
///
/// This class doesn't currently implement any additional logic over
BaseExponentialBackoffRetryFailureDetector and is
/// retained for backward compatibility.
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/failuredetector/FailureDetector.java
b/pinot-common/src/main/java/org/apache/pinot/common/failuredetector/FailureDetector.java
index bc230c94acd..2344e685978 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/failuredetector/FailureDetector.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/failuredetector/FailureDetector.java
@@ -46,6 +46,16 @@ public interface FailureDetector {
/// does not know about this server.
void registerUnhealthyServerRetrier(Function<String, ServerState>
unhealthyServerRetrier);
+ /// Registers a function that checks a server which left a query unanswered
(see [#notifyServerNotResponded]). It is
+ /// called with the instanceId of the server and should return
[ServerState#UNHEALTHY] only if the server is not
+ /// answering at all, and [ServerState#HEALTHY] or [ServerState#UNKNOWN]
otherwise.
+ ///
+ /// Kept apart from the retriers on purpose. A retrier answers whether an
unhealthy server has recovered, and may err
+ /// towards keeping it out -- the multi-stage one reports an idle gRPC
channel as unhealthy. A checker decides whether
+ /// a server that is in routing comes out, so it must only report a server
that verifiably failed to answer.
+ default void registerServerNotRespondedChecker(Function<String, ServerState>
serverNotRespondedChecker) {
+ }
+
/// Registers a consumer that will be called with the instanceId of a server
that is detected as healthy.
void registerHealthyServerNotifier(Consumer<String> healthyServerNotifier);
@@ -73,6 +83,19 @@ public interface FailureDetector {
/// Marks a server as unhealthy.
void markServerUnhealthy(String instanceId, @Nullable String hostName);
+ /// Notifies the detector that a server left a query unanswered until the
query deadline, without its connection
+ /// ever breaking.
+ ///
+ /// A server whose node is gone, whose network path is blackholed, or whose
JVM is frozen produces no other signal.
+ /// Writing into an open socket succeeds even when the peer is gone, so
there is no send exception, the channel is
+ /// never torn down, and connection-failure detection cannot see it. The
query simply times out.
+ ///
+ /// A timeout alone cannot tell such a server from one that is merely busy,
so implementations must check the server
+ /// with the registered checkers (see [#registerServerNotRespondedChecker])
before acting on it, and do nothing when
+ /// none is registered.
+ default void notifyServerNotResponded(String instanceId, @Nullable String
hostName) {
+ }
+
/// Returns all the unhealthy servers.
Set<String> getUnhealthyServers();
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/request/InstanceRequest.java
b/pinot-common/src/main/java/org/apache/pinot/common/request/InstanceRequest.java
index 79ac876ba11..e673825ed16 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/request/InstanceRequest.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/request/InstanceRequest.java
@@ -37,6 +37,7 @@ public class InstanceRequest implements
org.apache.thrift.TBase<InstanceRequest,
private static final org.apache.thrift.protocol.TField
OPTIONAL_SEGMENTS_FIELD_DESC = new
org.apache.thrift.protocol.TField("optionalSegments",
org.apache.thrift.protocol.TType.LIST, (short)6);
private static final org.apache.thrift.protocol.TField CID_FIELD_DESC = new
org.apache.thrift.protocol.TField("cid",
org.apache.thrift.protocol.TType.STRING, (short)7);
private static final org.apache.thrift.protocol.TField
TABLE_SEGMENTS_INFO_LIST_FIELD_DESC = new
org.apache.thrift.protocol.TField("tableSegmentsInfoList",
org.apache.thrift.protocol.TType.LIST, (short)8);
+ private static final org.apache.thrift.protocol.TField PING_FIELD_DESC = new
org.apache.thrift.protocol.TField("ping",
org.apache.thrift.protocol.TType.BOOL, (short)9);
private static final org.apache.thrift.scheme.SchemeFactory
STANDARD_SCHEME_FACTORY = new InstanceRequestStandardSchemeFactory();
private static final org.apache.thrift.scheme.SchemeFactory
TUPLE_SCHEME_FACTORY = new InstanceRequestTupleSchemeFactory();
@@ -49,6 +50,7 @@ public class InstanceRequest implements
org.apache.thrift.TBase<InstanceRequest,
private @org.apache.thrift.annotation.Nullable
java.util.List<java.lang.String> optionalSegments; // optional
private @org.apache.thrift.annotation.Nullable java.lang.String cid; //
optional
private @org.apache.thrift.annotation.Nullable
java.util.List<TableSegmentsInfo> tableSegmentsInfoList; // optional
+ private boolean ping; // optional
/** The set of fields this struct contains, along with convenience methods
for finding and manipulating them. */
public enum _Fields implements org.apache.thrift.TFieldIdEnum {
@@ -59,7 +61,8 @@ public class InstanceRequest implements
org.apache.thrift.TBase<InstanceRequest,
BROKER_ID((short)5, "brokerId"),
OPTIONAL_SEGMENTS((short)6, "optionalSegments"),
CID((short)7, "cid"),
- TABLE_SEGMENTS_INFO_LIST((short)8, "tableSegmentsInfoList");
+ TABLE_SEGMENTS_INFO_LIST((short)8, "tableSegmentsInfoList"),
+ PING((short)9, "ping");
private static final java.util.Map<java.lang.String, _Fields> byName = new
java.util.HashMap<java.lang.String, _Fields>();
@@ -91,6 +94,8 @@ public class InstanceRequest implements
org.apache.thrift.TBase<InstanceRequest,
return CID;
case 8: // TABLE_SEGMENTS_INFO_LIST
return TABLE_SEGMENTS_INFO_LIST;
+ case 9: // PING
+ return PING;
default:
return null;
}
@@ -136,8 +141,9 @@ public class InstanceRequest implements
org.apache.thrift.TBase<InstanceRequest,
// isset id assignments
private static final int __REQUESTID_ISSET_ID = 0;
private static final int __ENABLETRACE_ISSET_ID = 1;
+ private static final int __PING_ISSET_ID = 2;
private byte __isset_bitfield = 0;
- private static final _Fields optionals[] =
{_Fields.SEARCH_SEGMENTS,_Fields.ENABLE_TRACE,_Fields.BROKER_ID,_Fields.OPTIONAL_SEGMENTS,_Fields.CID,_Fields.TABLE_SEGMENTS_INFO_LIST};
+ private static final _Fields optionals[] =
{_Fields.SEARCH_SEGMENTS,_Fields.ENABLE_TRACE,_Fields.BROKER_ID,_Fields.OPTIONAL_SEGMENTS,_Fields.CID,_Fields.TABLE_SEGMENTS_INFO_LIST,_Fields.PING};
public static final java.util.Map<_Fields,
org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
static {
java.util.Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap =
new java.util.EnumMap<_Fields,
org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
@@ -160,6 +166,8 @@ public class InstanceRequest implements
org.apache.thrift.TBase<InstanceRequest,
tmpMap.put(_Fields.TABLE_SEGMENTS_INFO_LIST, new
org.apache.thrift.meta_data.FieldMetaData("tableSegmentsInfoList",
org.apache.thrift.TFieldRequirementType.OPTIONAL,
new
org.apache.thrift.meta_data.ListMetaData(org.apache.thrift.protocol.TType.LIST,
new
org.apache.thrift.meta_data.StructMetaData(org.apache.thrift.protocol.TType.STRUCT,
TableSegmentsInfo.class))));
+ tmpMap.put(_Fields.PING, new
org.apache.thrift.meta_data.FieldMetaData("ping",
org.apache.thrift.TFieldRequirementType.OPTIONAL,
+ new
org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.BOOL)));
metaDataMap = java.util.Collections.unmodifiableMap(tmpMap);
org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(InstanceRequest.class,
metaDataMap);
}
@@ -208,6 +216,7 @@ public class InstanceRequest implements
org.apache.thrift.TBase<InstanceRequest,
}
this.tableSegmentsInfoList = __this__tableSegmentsInfoList;
}
+ this.ping = other.ping;
}
@Override
@@ -227,6 +236,8 @@ public class InstanceRequest implements
org.apache.thrift.TBase<InstanceRequest,
this.optionalSegments = null;
this.cid = null;
this.tableSegmentsInfoList = null;
+ setPingIsSet(false);
+ this.ping = false;
}
public long getRequestId() {
@@ -465,6 +476,28 @@ public class InstanceRequest implements
org.apache.thrift.TBase<InstanceRequest,
}
}
+ public boolean isPing() {
+ return this.ping;
+ }
+
+ public void setPing(boolean ping) {
+ this.ping = ping;
+ setPingIsSet(true);
+ }
+
+ public void unsetPing() {
+ __isset_bitfield =
org.apache.thrift.EncodingUtils.clearBit(__isset_bitfield, __PING_ISSET_ID);
+ }
+
+ /** Returns true if field ping is set (has been assigned a value) and false
otherwise */
+ public boolean isSetPing() {
+ return org.apache.thrift.EncodingUtils.testBit(__isset_bitfield,
__PING_ISSET_ID);
+ }
+
+ public void setPingIsSet(boolean value) {
+ __isset_bitfield =
org.apache.thrift.EncodingUtils.setBit(__isset_bitfield, __PING_ISSET_ID,
value);
+ }
+
@Override
public void setFieldValue(_Fields field,
@org.apache.thrift.annotation.Nullable java.lang.Object value) {
switch (field) {
@@ -532,6 +565,14 @@ public class InstanceRequest implements
org.apache.thrift.TBase<InstanceRequest,
}
break;
+ case PING:
+ if (value == null) {
+ unsetPing();
+ } else {
+ setPing((java.lang.Boolean)value);
+ }
+ break;
+
}
}
@@ -563,6 +604,9 @@ public class InstanceRequest implements
org.apache.thrift.TBase<InstanceRequest,
case TABLE_SEGMENTS_INFO_LIST:
return getTableSegmentsInfoList();
+ case PING:
+ return isPing();
+
}
throw new java.lang.IllegalStateException();
}
@@ -591,6 +635,8 @@ public class InstanceRequest implements
org.apache.thrift.TBase<InstanceRequest,
return isSetCid();
case TABLE_SEGMENTS_INFO_LIST:
return isSetTableSegmentsInfoList();
+ case PING:
+ return isSetPing();
}
throw new java.lang.IllegalStateException();
}
@@ -680,6 +726,15 @@ public class InstanceRequest implements
org.apache.thrift.TBase<InstanceRequest,
return false;
}
+ boolean this_present_ping = true && this.isSetPing();
+ boolean that_present_ping = true && that.isSetPing();
+ if (this_present_ping || that_present_ping) {
+ if (!(this_present_ping && that_present_ping))
+ return false;
+ if (this.ping != that.ping)
+ return false;
+ }
+
return true;
}
@@ -717,6 +772,10 @@ public class InstanceRequest implements
org.apache.thrift.TBase<InstanceRequest,
if (isSetTableSegmentsInfoList())
hashCode = hashCode * 8191 + tableSegmentsInfoList.hashCode();
+ hashCode = hashCode * 8191 + ((isSetPing()) ? 131071 : 524287);
+ if (isSetPing())
+ hashCode = hashCode * 8191 + ((ping) ? 131071 : 524287);
+
return hashCode;
}
@@ -808,6 +867,16 @@ public class InstanceRequest implements
org.apache.thrift.TBase<InstanceRequest,
return lastComparison;
}
}
+ lastComparison = java.lang.Boolean.compare(isSetPing(), other.isSetPing());
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ if (isSetPing()) {
+ lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.ping,
other.ping);
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ }
return 0;
}
@@ -899,6 +968,12 @@ public class InstanceRequest implements
org.apache.thrift.TBase<InstanceRequest,
}
first = false;
}
+ if (isSetPing()) {
+ if (!first) sb.append(", ");
+ sb.append("ping:");
+ sb.append(this.ping);
+ first = false;
+ }
sb.append(")");
return sb.toString();
}
@@ -1053,6 +1128,14 @@ public class InstanceRequest implements
org.apache.thrift.TBase<InstanceRequest,
org.apache.thrift.protocol.TProtocolUtil.skip(iprot,
schemeField.type);
}
break;
+ case 9: // PING
+ if (schemeField.type == org.apache.thrift.protocol.TType.BOOL) {
+ struct.ping = iprot.readBool();
+ struct.setPingIsSet(true);
+ } else {
+ org.apache.thrift.protocol.TProtocolUtil.skip(iprot,
schemeField.type);
+ }
+ break;
default:
org.apache.thrift.protocol.TProtocolUtil.skip(iprot,
schemeField.type);
}
@@ -1136,6 +1219,11 @@ public class InstanceRequest implements
org.apache.thrift.TBase<InstanceRequest,
oprot.writeFieldEnd();
}
}
+ if (struct.isSetPing()) {
+ oprot.writeFieldBegin(PING_FIELD_DESC);
+ oprot.writeBool(struct.ping);
+ oprot.writeFieldEnd();
+ }
oprot.writeFieldStop();
oprot.writeStructEnd();
}
@@ -1175,7 +1263,10 @@ public class InstanceRequest implements
org.apache.thrift.TBase<InstanceRequest,
if (struct.isSetTableSegmentsInfoList()) {
optionals.set(5);
}
- oprot.writeBitSet(optionals, 6);
+ if (struct.isSetPing()) {
+ optionals.set(6);
+ }
+ oprot.writeBitSet(optionals, 7);
if (struct.isSetSearchSegments()) {
{
oprot.writeI32(struct.searchSegments.size());
@@ -1212,6 +1303,9 @@ public class InstanceRequest implements
org.apache.thrift.TBase<InstanceRequest,
}
}
}
+ if (struct.isSetPing()) {
+ oprot.writeBool(struct.ping);
+ }
}
@Override
@@ -1222,7 +1316,7 @@ public class InstanceRequest implements
org.apache.thrift.TBase<InstanceRequest,
struct.query = new BrokerRequest();
struct.query.read(iprot);
struct.setQueryIsSet(true);
- java.util.BitSet incoming = iprot.readBitSet(6);
+ java.util.BitSet incoming = iprot.readBitSet(7);
if (incoming.get(0)) {
{
org.apache.thrift.protocol.TList _list31 =
iprot.readListBegin(org.apache.thrift.protocol.TType.STRING);
@@ -1275,6 +1369,10 @@ public class InstanceRequest implements
org.apache.thrift.TBase<InstanceRequest,
}
struct.setTableSegmentsInfoListIsSet(true);
}
+ if (incoming.get(6)) {
+ struct.ping = iprot.readBool();
+ struct.setPingIsSet(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 927cee215f5..45ab8ea1212 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
@@ -18,9 +18,20 @@
*/
package org.apache.pinot.common.failuredetector;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.function.Function;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.failuredetector.FailureDetector.ServerState;
import org.apache.pinot.common.metrics.BrokerGauge;
import org.apache.pinot.common.metrics.BrokerMetrics;
import org.apache.pinot.common.metrics.MetricValueUtils;
@@ -28,7 +39,7 @@ import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.metrics.NoopPinotMetricsRegistry;
import org.apache.pinot.spi.utils.CommonConstants.Broker;
import org.apache.pinot.util.TestUtils;
-import org.testng.annotations.AfterClass;
+import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@@ -40,6 +51,13 @@ public class ConnectionFailureDetectorTest {
private static final String INSTANCE_ID = "Server_localhost_1234";
private static final String HOST_NAME = "localhost";
+ private static final String OTHER_INSTANCE_ID = "Server_localhost_5678";
+
+ /// Long enough that the retry loop never runs during a test that is not
about it.
+ private static final long RETRIES_OFF_MS = 600_000L;
+
+ private final List<FailureDetector> _startedFailureDetectors = new
ArrayList<>();
+ private final List<String> _markedUnhealthy = new CopyOnWriteArrayList<>();
private BrokerMetrics _brokerMetrics;
private FailureDetector _failureDetector;
private UnhealthyServerRetrier _unhealthyServerRetrier;
@@ -48,6 +66,7 @@ public class ConnectionFailureDetectorTest {
@BeforeMethod
public void setUp() {
+ _markedUnhealthy.clear();
PinotConfiguration config = new PinotConfiguration();
config.setProperty(Broker.FailureDetector.CONFIG_OF_TYPE,
Broker.FailureDetector.Type.CONNECTION.name());
config.setProperty(Broker.FailureDetector.CONFIG_OF_RETRY_INITIAL_DELAY_MS,
100);
@@ -177,6 +196,263 @@ public class ConnectionFailureDetectorTest {
assertEquals(_unhealthyServerRetrier._retryUnhealthyServerCalled, 8);
}
+ /// A server that leaves a query unanswered and then fails the follow-up
check is marked unhealthy.
+ @Test
+ public void testServerFailingTheTimeoutCheckIsMarkedUnhealthy() {
+ FailureDetector failureDetector =
startFailureDetectorWithChecker(instanceId -> ServerState.UNHEALTHY);
+ failureDetector.notifyServerNotResponded(INSTANCE_ID, HOST_NAME);
+
+ TestUtils.waitForCondition(aVoid ->
failureDetector.getUnhealthyServers().contains(INSTANCE_ID), 5_000L,
+ "Server failing the check was not marked unhealthy");
+ assertEquals(_markedUnhealthy, List.of(INSTANCE_ID));
+ }
+
+ /// A server that leaves a query unanswered but passes the check -- busy,
not dead -- is left alone.
+ @Test
+ public void testServerPassingTheTimeoutCheckIsLeftAlone() {
+ AtomicInteger numChecks = new AtomicInteger();
+ FailureDetector failureDetector =
startFailureDetectorWithChecker(instanceId -> {
+ numChecks.incrementAndGet();
+ return ServerState.HEALTHY;
+ });
+ failureDetector.notifyServerNotResponded(INSTANCE_ID, HOST_NAME);
+
+ awaitCheckFinished(failureDetector, INSTANCE_ID, numChecks);
+ assertEquals(failureDetector.getUnhealthyServers(), Set.of());
+ assertEquals(_markedUnhealthy, List.of());
+ }
+
+ /// A check after a timeout runs the checkers only, never the retriers. A
retrier may report a live server unhealthy
+ /// -- the multi-stage one does for an idle gRPC channel -- which must not
take a server that answered out of routing.
+ @Test
+ public void testTimeoutCheckIgnoresRetriers() {
+ AtomicInteger numChecks = new AtomicInteger();
+ AtomicInteger numRetries = new AtomicInteger();
+ FailureDetector failureDetector = startFailureDetector(new
BrokerMetrics(new NoopPinotMetricsRegistry()),
+ RETRIES_OFF_MS, instanceId -> {
+ numRetries.incrementAndGet();
+ return ServerState.UNHEALTHY;
+ }, instanceId -> {
+ numChecks.incrementAndGet();
+ return ServerState.HEALTHY;
+ });
+ failureDetector.notifyServerNotResponded(INSTANCE_ID, HOST_NAME);
+
+ awaitCheckFinished(failureDetector, INSTANCE_ID, numChecks);
+ assertEquals(failureDetector.getUnhealthyServers(), Set.of());
+ assertEquals(numRetries.get(), 0);
+ }
+
+ /// A check that throws is inconclusive, and an inconclusive check must err
towards keeping the server.
+ @Test
+ public void testThrowingTimeoutCheckDoesNotMarkServerUnhealthy() {
+ AtomicInteger numChecks = new AtomicInteger();
+ FailureDetector failureDetector =
startFailureDetectorWithChecker(instanceId -> {
+ numChecks.incrementAndGet();
+ throw new RuntimeException("Check failed to run");
+ });
+ failureDetector.notifyServerNotResponded(INSTANCE_ID, HOST_NAME);
+
+ awaitCheckFinished(failureDetector, INSTANCE_ID, numChecks);
+ assertEquals(failureDetector.getUnhealthyServers(), Set.of());
+ }
+
+ /// A burst of timeouts against one server -- every query in flight to it,
say -- checks it once.
+ @Test
+ public void testBurstOfTimeoutsChecksServerOnce() {
+ CountDownLatch releaseCheck = new CountDownLatch(1);
+ AtomicInteger numChecks = new AtomicInteger();
+ FailureDetector failureDetector =
startFailureDetectorWithChecker(instanceId -> {
+ numChecks.incrementAndGet();
+ awaitQuietly(releaseCheck);
+ return ServerState.HEALTHY;
+ });
+ for (int i = 0; i < 100; i++) {
+ failureDetector.notifyServerNotResponded(INSTANCE_ID, HOST_NAME);
+ }
+ TestUtils.waitForCondition(aVoid -> numChecks.get() == 1, 5_000L, "Check
did not start");
+ for (int i = 0; i < 100; i++) {
+ failureDetector.notifyServerNotResponded(INSTANCE_ID, HOST_NAME);
+ }
+ releaseCheck.countDown();
+ assertEquals(numChecks.get(), 1);
+ }
+
+ /// A server already unhealthy is the retry loop's business; a timeout does
not check it again.
+ @Test
+ public void testUnhealthyServerIsNotCheckedOnTimeout() {
+ List<String> checkedServers = new CopyOnWriteArrayList<>();
+ FailureDetector failureDetector =
startFailureDetectorWithChecker(instanceId -> {
+ checkedServers.add(instanceId);
+ return ServerState.HEALTHY;
+ });
+ failureDetector.markServerUnhealthy(INSTANCE_ID, HOST_NAME);
+ failureDetector.notifyServerNotResponded(INSTANCE_ID, HOST_NAME);
+ // The skip happens on the caller's thread, so once another server's check
has run there is nothing left pending.
+ failureDetector.notifyServerNotResponded(OTHER_INSTANCE_ID, HOST_NAME);
+
+ TestUtils.waitForCondition(aVoid ->
checkedServers.contains(OTHER_INSTANCE_ID), 5_000L, "Check did not run");
+ assertEquals(checkedServers, List.of(OTHER_INSTANCE_ID));
+ }
+
+ /// With no checker registered -- pings disabled -- a timeout does nothing,
while connection failures are still
+ /// detected.
+ @Test
+ public void testTimeoutDoesNothingWithoutAChecker()
+ throws InterruptedException {
+ AtomicInteger numRetries = new AtomicInteger();
+ FailureDetector failureDetector =
+ startFailureDetector(new BrokerMetrics(new
NoopPinotMetricsRegistry()), RETRIES_OFF_MS, instanceId -> {
+ numRetries.incrementAndGet();
+ return ServerState.UNHEALTHY;
+ }, null);
+ failureDetector.notifyServerNotResponded(INSTANCE_ID, HOST_NAME);
+ // A check runs off this thread, within milliseconds; give one far longer
than that before concluding there is none.
+ Thread.sleep(500L);
+ assertEquals(numRetries.get(), 0);
+ assertEquals(failureDetector.getUnhealthyServers(), Set.of());
+
+ failureDetector.markServerUnhealthy(INSTANCE_ID, HOST_NAME);
+ assertEquals(failureDetector.getUnhealthyServers(), Set.of(INSTANCE_ID));
+ }
+
+ /// Several servers going silent at once are all checked side by side, and
all marked unhealthy. Each check waits
+ /// here until both have started: run one after the other, the first would
give up waiting and pass.
+ @Test
+ public void testTimeoutChecksRunInParallel() {
+ CountDownLatch bothStarted = new CountDownLatch(2);
+ FailureDetector failureDetector =
startFailureDetectorWithChecker(instanceId -> {
+ bothStarted.countDown();
+ return awaitQuietly(bothStarted) ? ServerState.UNHEALTHY :
ServerState.HEALTHY;
+ });
+ failureDetector.notifyServerNotResponded(INSTANCE_ID, HOST_NAME);
+ failureDetector.notifyServerNotResponded(OTHER_INSTANCE_ID, HOST_NAME);
+
+ TestUtils.waitForCondition(aVoid ->
failureDetector.getUnhealthyServers().size() == 2, 5_000L,
+ "Checks did not run in parallel");
+ assertEquals(failureDetector.getUnhealthyServers(), Set.of(INSTANCE_ID,
OTHER_INSTANCE_ID));
+ }
+
+ /// Retries run side by side too, so one dead server waiting out a ping
timeout cannot delay another's recovery.
+ @Test
+ public void testRetriesRunInParallel() {
+ CountDownLatch bothRetried = new CountDownLatch(2);
+ FailureDetector failureDetector =
startFailureDetectorWithRetrier(instanceId -> {
+ bothRetried.countDown();
+ return awaitQuietly(bothRetried) ? ServerState.HEALTHY :
ServerState.UNHEALTHY;
+ });
+ failureDetector.markServerUnhealthy(INSTANCE_ID, HOST_NAME);
+ failureDetector.markServerUnhealthy(OTHER_INSTANCE_ID, HOST_NAME);
+
+ TestUtils.waitForCondition(aVoid ->
failureDetector.getUnhealthyServers().isEmpty(), 5_000L,
+ "Retries did not run in parallel");
+ }
+
+ /// A retrier that throws counts as a failed retry. The server is retried
again later rather than stranded unhealthy
+ /// forever, which is what happened when the exception escaped the retry
loop.
+ @Test
+ public void testThrowingRetryDoesNotStrandServer() {
+ AtomicInteger numRetries = new AtomicInteger();
+ FailureDetector failureDetector =
startFailureDetectorWithRetrier(instanceId -> {
+ if (numRetries.incrementAndGet() == 1) {
+ throw new RuntimeException("Retry failed to run");
+ }
+ return ServerState.HEALTHY;
+ });
+ failureDetector.markServerUnhealthy(INSTANCE_ID, HOST_NAME);
+
+ TestUtils.waitForCondition(aVoid ->
failureDetector.getUnhealthyServers().isEmpty(), 5_000L,
+ "Server was stranded unhealthy");
+ assertEquals(numRetries.get(), 2);
+ }
+
+ /// The gauge stays exact when servers change state concurrently, which
parallel checks make routine. It used to be
+ /// adjusted by one inside each map update, where two concurrent updates
read the same size. Repeated, since a single
+ /// round only sometimes hits the race.
+ @Test
+ public void testUnhealthyServerGaugeStaysExactUnderConcurrentChanges()
+ throws Exception {
+ BrokerMetrics brokerMetrics = new BrokerMetrics(new
NoopPinotMetricsRegistry());
+ FailureDetector failureDetector = startFailureDetector(brokerMetrics,
RETRIES_OFF_MS, null, null);
+ int numServers = 64;
+ ExecutorService executorService = Executors.newFixedThreadPool(16);
+ try {
+ List<Callable<Void>> markUnhealthy = new ArrayList<>();
+ List<Callable<Void>> markHealthy = new ArrayList<>();
+ for (int i = 0; i < numServers; i++) {
+ String instanceId = "Server_host_" + i;
+ markUnhealthy.add(() -> {
+ failureDetector.markServerUnhealthy(instanceId, HOST_NAME);
+ return null;
+ });
+ markHealthy.add(() -> {
+ failureDetector.markServerHealthy(instanceId, HOST_NAME);
+ return null;
+ });
+ }
+ for (int round = 0; round < 20; round++) {
+ executorService.invokeAll(markUnhealthy);
+ assertEquals(MetricValueUtils.getGlobalGaugeValue(brokerMetrics,
BrokerGauge.UNHEALTHY_SERVERS), numServers);
+ executorService.invokeAll(markHealthy);
+ assertEquals(MetricValueUtils.getGlobalGaugeValue(brokerMetrics,
BrokerGauge.UNHEALTHY_SERVERS), 0);
+ }
+ } finally {
+ executorService.shutdownNow();
+ }
+ }
+
+ /// Starts a detector that checks servers after a timeout with `checker`,
and never retries during the test.
+ private FailureDetector startFailureDetectorWithChecker(Function<String,
ServerState> checker) {
+ return startFailureDetector(new BrokerMetrics(new
NoopPinotMetricsRegistry()), RETRIES_OFF_MS, null, checker);
+ }
+
+ /// Starts a detector that retries unhealthy servers with `retrier` after
100 ms, and has no timeout check.
+ private FailureDetector startFailureDetectorWithRetrier(Function<String,
ServerState> retrier) {
+ return startFailureDetector(new BrokerMetrics(new
NoopPinotMetricsRegistry()), 100L, retrier, null);
+ }
+
+ /// Starts a detector with the given retrier and timeout checker, each
optional, recording the servers it marks
+ /// unhealthy. Stopped after the test.
+ private FailureDetector startFailureDetector(BrokerMetrics brokerMetrics,
long retryInitialDelayMs,
+ @Nullable Function<String, ServerState> retrier, @Nullable
Function<String, ServerState> checker) {
+ PinotConfiguration config = new PinotConfiguration();
+ config.setProperty(Broker.FailureDetector.CONFIG_OF_TYPE,
Broker.FailureDetector.Type.CONNECTION.name());
+
config.setProperty(Broker.FailureDetector.CONFIG_OF_RETRY_INITIAL_DELAY_MS,
retryInitialDelayMs);
+ config.setProperty(Broker.FailureDetector.CONFIG_OF_RETRY_DELAY_FACTOR, 1);
+ FailureDetector failureDetector =
FailureDetectorFactory.getFailureDetector(config, brokerMetrics);
+ failureDetector.registerUnhealthyServerNotifier(_markedUnhealthy::add);
+ failureDetector.registerHealthyServerNotifier(instanceId -> {
+ });
+ if (retrier != null) {
+ failureDetector.registerUnhealthyServerRetrier(retrier);
+ }
+ if (checker != null) {
+ failureDetector.registerServerNotRespondedChecker(checker);
+ }
+ failureDetector.start();
+ _startedFailureDetectors.add(failureDetector);
+ return failureDetector;
+ }
+
+ /// Waits until the check of `instanceId` has finished, not merely started.
A notification only starts a new check
+ /// once the previous one has finished, so a second check proves the first
is done. Stops early if the server got
+ /// marked unhealthy, which ends its checks, and leaves that to the caller's
assertions.
+ private static void awaitCheckFinished(FailureDetector failureDetector,
String instanceId, AtomicInteger numChecks) {
+ TestUtils.waitForCondition(aVoid -> {
+ failureDetector.notifyServerNotResponded(instanceId, HOST_NAME);
+ return numChecks.get() >= 2 ||
failureDetector.getUnhealthyServers().contains(instanceId);
+ }, 5_000L, "Check did not finish");
+ }
+
+ private static boolean awaitQuietly(CountDownLatch latch) {
+ try {
+ return latch.await(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ }
+
private void verify(Set<String> expectedUnhealthyServers, int
expectedNotifyUnhealthyServerCalled,
int expectedNotifyHealthyServerCalled) {
assertEquals(_failureDetector.getUnhealthyServers(),
expectedUnhealthyServers);
@@ -186,9 +462,13 @@ public class ConnectionFailureDetectorTest {
assertEquals(_healthyServerNotifier._notifyHealthyServerCalled,
expectedNotifyHealthyServerCalled);
}
- @AfterClass
+ @AfterMethod
public void tearDown() {
_failureDetector.stop();
+ for (FailureDetector failureDetector : _startedFailureDetectors) {
+ failureDetector.stop();
+ }
+ _startedFailureDetectors.clear();
}
private static class HealthyServerNotifier implements Consumer<String> {
diff --git a/pinot-common/src/thrift/request.thrift
b/pinot-common/src/thrift/request.thrift
index c2134677444..ca69b1f9387 100644
--- a/pinot-common/src/thrift/request.thrift
+++ b/pinot-common/src/thrift/request.thrift
@@ -60,4 +60,8 @@ struct InstanceRequest {
6: optional list<string> optionalSegments;
7: optional string cid;
8: optional list<TableSegmentsInfo> tableSegmentsInfoList;
+ // Liveness check sent by the broker after a query times out. A server that
understands it replies straight away
+ // without running anything; an older server ignores the field and answers
it as an ordinary query, once the queries
+ // queued ahead of it have run.
+ 9: optional bool ping;
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/transport/InstanceRequestHandler.java
b/pinot-core/src/main/java/org/apache/pinot/core/transport/InstanceRequestHandler.java
index 0c4e239c626..1e6a1177374 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/transport/InstanceRequestHandler.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/transport/InstanceRequestHandler.java
@@ -133,6 +133,10 @@ public class InstanceRequestHandler extends
SimpleChannelInboundHandler<ByteBuf>
_serverMetrics.addMeteredGlobalValue(ServerMeter.REQUEST_DESERIALIZATION_EXCEPTIONS,
1);
return;
}
+ if (instanceRequest.isPing()) {
+ sendPingResponse(ctx, instanceRequest.getRequestId());
+ return;
+ }
ServerQueryRequest queryRequest;
try {
queryRequest = new ServerQueryRequest(instanceRequest, _serverMetrics,
queryArrivalTimeMs);
@@ -160,6 +164,21 @@ public class InstanceRequestHandler extends
SimpleChannelInboundHandler<ByteBuf>
}
}
+ /// Answers a broker's ping with an empty data table carrying its request
id, which is all the broker waits for.
+ ///
+ /// The reply is written straight from this event loop and never goes
through the query scheduler. The broker pings a
+ /// server after a query to it timed out, to ask whether the server is alive
at all; a server that is merely busy must
+ /// still answer promptly, so the reply cannot wait behind queued queries.
+ private void sendPingResponse(ChannelHandlerContext ctx, long requestId) {
+ try {
+ DataTable dataTable = DataTableBuilderFactory.getEmptyDataTable();
+ dataTable.getMetadata().put(MetadataKey.REQUEST_ID.getName(),
Long.toString(requestId));
+ ctx.writeAndFlush(Unpooled.wrappedBuffer(dataTable.toBytes()));
+ } catch (Exception e) {
+ LOGGER.warn("Failed to answer ping request: {}", requestId, e);
+ }
+ }
+
/// Submit query for execution and register callback for execution results.
/// If query cancellation is enabled, the query future is tracked as well.
@VisibleForTesting
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 7d58a2a3413..d0c9d3f1c5b 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
@@ -18,9 +18,15 @@
*/
package org.apache.pinot.core.transport;
+import com.google.common.annotations.VisibleForTesting;
+import io.netty.channel.Channel;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;
import org.apache.pinot.common.config.NettyConfig;
@@ -38,6 +44,9 @@ import org.apache.pinot.core.routing.TableRouteInfo;
import
org.apache.pinot.core.transport.server.routing.stats.ServerRoutingStatsManager;
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.builder.TableNameBuilder;
+import org.apache.pinot.sql.parsers.CalciteSqlCompiler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,6 +59,15 @@ import org.slf4j.LoggerFactory;
public class QueryRouter {
private static final Logger LOGGER =
LoggerFactory.getLogger(QueryRouter.class);
+ /// The query a ping carries. A server that understands pings never runs it.
An older server ignores the ping flag
+ /// and runs it like any other query, and naming a table no server hosts
keeps that cheap: the server answers with a
+ /// table-missing error, and any answer at all is what a ping is after. It
has to be a real query for that to work,
+ /// because a server that cannot parse a request drops it without replying.
The older server still queues it behind
+ /// the queries it is running, though, which is why pings stay off until
every server understands them.
+ private static final String PING_TABLE_NAME = "pinotBrokerPing";
+ private static final BrokerRequest PING_BROKER_REQUEST =
CalciteSqlCompiler.compileToBrokerRequest(
+ "SELECT COUNT(*) FROM " +
TableNameBuilder.OFFLINE.tableNameWithType(PING_TABLE_NAME));
+
private final String _brokerId;
private final ServerChannels _serverChannels;
private final ServerChannels _serverChannelsTls;
@@ -57,6 +75,12 @@ public class QueryRouter {
private final BrokerMetrics _brokerMetrics = BrokerMetrics.get();
private final ConcurrentHashMap<Long, AsyncQueryResponse>
_asyncQueryResponseMap = new ConcurrentHashMap<>();
+ /// 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();
+ /// Never initialized, which leaves stats collection disabled: pings must
not skew adaptive server selection.
+ private final ServerRoutingStatsManager _noOpRoutingStatsManager =
+ new ServerRoutingStatsManager(new PinotConfiguration(), _brokerMetrics);
/// Creates a query router with TLS config.
///
@@ -146,20 +170,17 @@ public class QueryRouter {
asyncQueryResponse.markQueryFailed(serverRoutingInstance, e);
}
+ /// Returns whether the broker has a query channel to the server, connected
or not. There is one channel per table
+ /// type, so this checks both. A server never queried over this path -- e.g.
one that only serves multi-stage queries
+ /// -- has none.
public boolean hasChannel(ServerInstance serverInstance) {
- if (_serverChannelsTls != null) {
- return _serverChannelsTls.hasChannel(
- serverInstance.toServerRoutingInstance(TableType.OFFLINE,
ServerInstance.RoutingType.NETTY_TLS));
- } else {
- return _serverChannels.hasChannel(
- serverInstance.toServerRoutingInstance(TableType.OFFLINE,
ServerInstance.RoutingType.NETTY));
- }
+ return !channelRoutingInstances(serverInstance).isEmpty();
}
/// Connects to the given server, returns `true` if the server is
successfully connected.
///
- /// Unchanged: this is the reachability probe the failure detector uses,
opening the OFFLINE channel
- /// with a TCP connect only. Startup pre-connect uses [#preConnect] instead.
+ /// 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.
public boolean connect(ServerInstance serverInstance) {
try {
if (_serverChannelsTls != null) {
@@ -176,6 +197,116 @@ public class QueryRouter {
}
}
+ /// Pings the server over every channel queries to it use, and returns
whether all of them answered within
+ /// `timeoutMs`. Returns `false` when there is no such channel, as there is
nothing to ping over.
+ ///
+ /// A server that is alive answers straight away even when it is busy,
because it replies from its network thread
+ /// instead of queueing behind queries (see `InstanceRequestHandler`). So a
channel that stays silent means the server
+ /// cannot answer anything -- its node is gone, its network path is
blackholed, or its JVM is frozen -- or that the
+ /// channel itself is half-open, left over from a server that went away.
Either way queries sent over it would time
+ /// out, which is why every channel has to answer, not just one.
+ ///
+ /// Each channel gets a ping and a response of its own. With a shared one, a
channel that fails at once would end the
+ /// wait for all of them, and a channel still due to answer would look
silent.
+ ///
+ /// The connection a ping went out on is closed when that ping stays silent,
so the next attempt connects afresh. It
+ /// is the one closed, rather than whichever connection the channel holds by
then: a query may have reconnected in
+ /// the meantime, and its new connection has nothing to do with the silent
one. Closing also ends whatever was still
+ /// in flight on the silent connection right away (see [DataTableHandler]),
rather than leaving those queries to wait
+ /// out their own timeouts. Pings record no routing stats, so they never
skew adaptive server selection.
+ public boolean ping(ServerInstance serverInstance, long timeoutMs) {
+ List<Long> requestIds = new ArrayList<>(2);
+ try {
+ List<ServerRoutingInstance> routingInstances =
channelRoutingInstances(serverInstance);
+ if (routingInstances.isEmpty()) {
+ return false;
+ }
+ ServerChannels serverChannels = channels();
+ List<SentPing> sentPings = new ArrayList<>(routingInstances.size());
+ boolean answered = true;
+ for (ServerRoutingInstance routingInstance : routingInstances) {
+ long requestId = _pingRequestIdGenerator.decrementAndGet();
+ // The timeout starts at this ping's own send, so a slow send over one
channel cannot cut short the wait for
+ // the next.
+ AsyncQueryResponse pingResponse =
+ new AsyncQueryResponse(this, requestId, Set.of(routingInstance),
System.currentTimeMillis(), timeoutMs,
+ _noOpRoutingStatsManager, false);
+ requestIds.add(requestId);
+ _asyncQueryResponseMap.put(requestId, pingResponse);
+ try {
+ Channel channel = serverChannels.sendRequest(PING_TABLE_NAME,
pingResponse, routingInstance,
+ newPingRequest(requestId, _brokerId), timeoutMs);
+ sentPings.add(new SentPing(pingResponse, channel));
+ } catch (Exception e) {
+ // No connection carried this ping, so there is none to close. Do
not spend another connect timeout on the
+ // remaining channels: the ping has failed already, and the channels
not reached stay as they are.
+ answered = false;
+ break;
+ }
+ }
+ for (SentPing sentPing : sentPings) {
+ if
(sentPing.response().getFinalResponses().values().iterator().next().getDataTable()
== null) {
+ answered = false;
+ sentPing.channel().close();
+ }
+ }
+ return answered;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ } catch (Exception e) {
+ // toServerRoutingInstance() rejects a server that has not advertised
the port, which happens mid-rollout.
+ LOGGER.debug("Failed to ping server: {}", serverInstance, e);
+ return false;
+ } finally {
+ // getFinalResponses() deregisters each response it waits for; this
covers the ones it never got to.
+ for (long requestId : requestIds) {
+ markQueryDone(requestId);
+ }
+ }
+ }
+
+ /// A ping that went out, and the connection it went out on.
+ private record SentPing(AsyncQueryResponse response, Channel channel) {
+ }
+
+ @VisibleForTesting
+ static InstanceRequest newPingRequest(long requestId, String brokerId) {
+ InstanceRequest pingRequest = new InstanceRequest(requestId,
PING_BROKER_REQUEST);
+ pingRequest.setPing(true);
+ pingRequest.setSearchSegments(List.of());
+ pingRequest.setBrokerId(brokerId);
+ return pingRequest;
+ }
+
+ /// The routing instances of the server's query channels, connected or not:
at most one per table type.
+ private List<ServerRoutingInstance> channelRoutingInstances(ServerInstance
serverInstance) {
+ ServerChannels serverChannels = channels();
+ List<ServerRoutingInstance> routingInstances = new ArrayList<>(2);
+ for (TableType tableType : TableType.values()) {
+ ServerRoutingInstance routingInstance = routingInstance(serverInstance,
tableType);
+ if (serverChannels.hasChannel(routingInstance)) {
+ routingInstances.add(routingInstance);
+ }
+ }
+ return routingInstances;
+ }
+
+ @VisibleForTesting
+ ServerChannels getServerChannels() {
+ return _serverChannels;
+ }
+
+ /// Queries go over TLS whenever the broker has TLS configured, so the
server's channels are all of one kind.
+ private ServerChannels channels() {
+ return _serverChannelsTls != null ? _serverChannelsTls : _serverChannels;
+ }
+
+ private ServerRoutingInstance routingInstance(ServerInstance serverInstance,
TableType tableType) {
+ return serverInstance.toServerRoutingInstance(tableType,
+ _serverChannelsTls != null ? ServerInstance.RoutingType.NETTY_TLS :
ServerInstance.RoutingType.NETTY);
+ }
+
/// Opens the channel used for the given table type ahead of query traffic,
including the TLS
/// handshake, bounded by `timeoutMs`. Returns `true` if it is connected.
///
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 d711c120337..6142662fe39 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
@@ -24,6 +24,7 @@ import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.PooledByteBufAllocatorMetric;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
@@ -139,14 +140,16 @@ public class ServerChannels {
_brokerMetrics.setOrUpdateGlobalGauge(BrokerGauge.NETTY_POOLED_CHUNK_SIZE,
metric::chunkSize);
}
- public void sendRequest(String rawTableName, AsyncQueryResponse
asyncQueryResponse,
+ /// Sends the request to the server, connecting first if there is no open
connection, and returns the connection the
+ /// request was written to.
+ public Channel sendRequest(String rawTableName, AsyncQueryResponse
asyncQueryResponse,
ServerRoutingInstance serverRoutingInstance, InstanceRequest
instanceRequest, long timeoutMs)
throws Exception {
byte[] requestBytes =
THREAD_LOCAL_T_SERIALIZER.get().serialize(instanceRequest);
ServerChannel serverChannel =
_serverToChannelMap.computeIfAbsent(serverRoutingInstance, ServerChannel::new);
// This server is now query-carrying, whoever opened the channel. See
[#hasChannel].
serverChannel._openedByQuery = true;
- serverChannel.sendRequest(rawTableName, asyncQueryResponse,
serverRoutingInstance, requestBytes, timeoutMs);
+ return serverChannel.sendRequest(rawTableName, asyncQueryResponse,
serverRoutingInstance, requestBytes, timeoutMs);
}
/// Whether this broker has sent, or tried to send, a single-stage query to
the given server.
@@ -249,13 +252,15 @@ public class ServerChannels {
}
}
- void sendRequest(String rawTableName, AsyncQueryResponse
asyncQueryResponse,
+ Channel sendRequest(String rawTableName, AsyncQueryResponse
asyncQueryResponse,
ServerRoutingInstance serverRoutingInstance, byte[] requestBytes, long
timeoutMs)
throws InterruptedException, TimeoutException {
if (_channelLock.tryLock(timeoutMs, TimeUnit.MILLISECONDS)) {
try {
- connectWithoutLocking();
+ connectWithoutLocking(timeoutMs);
+ Channel channel = _channel;
sendRequestWithoutLocking(rawTableName, asyncQueryResponse,
serverRoutingInstance, requestBytes);
+ return channel;
} finally {
_channelLock.unlock();
}
@@ -264,20 +269,29 @@ public class ServerChannels {
}
}
- /// 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.
- void connectWithoutLocking()
- throws InterruptedException {
+ /// Lazy query path: opens the TCP connection only, waiting at most
`timeoutMs`. 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.
+ ///
+ /// 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.
+ void connectWithoutLocking(long timeoutMs)
+ throws InterruptedException, TimeoutException {
if (_channel == null || !_channel.isActive()) {
long startTime = System.currentTimeMillis();
- _channel = _bootstrap.connect().sync().channel();
+ 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);
}
}
- /// Like [#connectWithoutLocking()] but additionally waits out the TLS
handshake, and bounds both
+ /// Like [#connectWithoutLocking(long)] but additionally waits out the TLS
handshake, and bounds both
/// waits by `timeoutMs`.
///
/// Used only by startup pre-connect ([#preConnect(long)]), never by the
lazy query path or by the
@@ -354,7 +368,9 @@ public class ServerChannels {
void sendRequestWithoutLocking(String rawTableName, AsyncQueryResponse
asyncQueryResponse,
ServerRoutingInstance serverRoutingInstance, byte[] requestBytes) {
long startTimeMs = System.currentTimeMillis();
-
_channel.writeAndFlush(Unpooled.wrappedBuffer(requestBytes)).addListener(f -> {
+ // The listener runs later, by when a reconnect may have replaced
_channel: on failure it closes this connection.
+ Channel channel = _channel;
+
channel.writeAndFlush(Unpooled.wrappedBuffer(requestBytes)).addListener(f -> {
if (f.isSuccess()) {
int requestSentLatencyMs = (int) (System.currentTimeMillis() -
startTimeMs);
_brokerMetrics.addTimedTableValue(rawTableName,
BrokerTimer.NETTY_CONNECTION_SEND_REQUEST_LATENCY,
@@ -367,7 +383,7 @@ public class ServerChannels {
new RuntimeException("Failed to send request to server: " +
serverRoutingInstance, f.cause()))) {
_brokerMetrics.addMeteredGlobalValue(BrokerMeter.SERVER_MARKED_DOWN_SKIPPED, 1);
}
- _channel.close();
+ channel.close();
}
});
_brokerMetrics.addMeteredGlobalValue(BrokerMeter.NETTY_CONNECTION_REQUESTS_SENT,
1);
@@ -383,7 +399,8 @@ public class ServerChannels {
throws InterruptedException, TimeoutException {
if (_channelLock.tryLock(TRY_CONNECT_CHANNEL_LOCK_TIMEOUT_MS,
TimeUnit.MILLISECONDS)) {
try {
- connectWithoutLocking();
+ // The connect gets as long as the lock wait.
+ connectWithoutLocking(TRY_CONNECT_CHANNEL_LOCK_TIMEOUT_MS);
} finally {
_channelLock.unlock();
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/transport/InstanceRequestHandlerTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/transport/InstanceRequestHandlerTest.java
index b83f19e6e80..ecf17e252f2 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/transport/InstanceRequestHandlerTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/transport/InstanceRequestHandlerTest.java
@@ -20,10 +20,12 @@ package org.apache.pinot.core.transport;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListenableFutureTask;
+import com.google.common.util.concurrent.SettableFuture;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.embedded.EmbeddedChannel;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import java.util.HashSet;
@@ -59,6 +61,7 @@ import org.testng.annotations.Test;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
@@ -75,6 +78,59 @@ public class InstanceRequestHandlerTest {
ServerMetrics.register(new ServerMetrics(registry));
}
+ /// A ping is answered straight from the network thread: the scheduler is
never touched, so a server whose queries
+ /// are queued, stuck or slow still answers it at once.
+ @Test
+ public void testPingIsAnsweredWithoutTouchingTheScheduler()
+ throws Exception {
+ QueryScheduler queryScheduler = mock(QueryScheduler.class);
+ EmbeddedChannel channel = new EmbeddedChannel(
+ new InstanceRequestHandler("server01", new PinotConfiguration(),
queryScheduler, mock(AccessControl.class),
+ ThreadAccountantUtils.getNoOpAccountant()));
+
+ long requestId = -42L;
+
channel.writeInbound(Unpooled.wrappedBuffer(serialize(QueryRouter.newPingRequest(requestId,
"broker01"))));
+
+ ByteBuf reply = channel.readOutbound();
+ byte[] replyBytes = new byte[reply.readableBytes()];
+ reply.readBytes(replyBytes);
+ reply.release();
+ DataTable dataTable = DataTableFactory.getDataTable(replyBytes);
+
assertEquals(dataTable.getMetadata().get(DataTable.MetadataKey.REQUEST_ID.getName()),
Long.toString(requestId),
+ "The broker matches the reply to its ping by request id");
+ verify(queryScheduler, never()).submit(any());
+ channel.finishAndReleaseAll();
+ }
+
+ /// A server that predates pings ignores the unknown field and sees an
ordinary query, which it answers once the
+ /// queries queued ahead of it have run. This is the step where such a
server would otherwise drop the request
+ /// without replying: if it cannot parse the request into a query, it sends
nothing back. (That queueing is why pings
+ /// are off by default: a busy old server can miss the ping timeout.)
+ @Test
+ public void testServerThatIgnoresThePingFlagSeesAnAnswerableQuery()
+ throws Exception {
+ InstanceRequest asSeenByOldServer = QueryRouter.newPingRequest(-42L,
"broker01");
+ asSeenByOldServer.unsetPing();
+
+ ServerQueryRequest queryRequest =
+ new ServerQueryRequest(asSeenByOldServer, ServerMetrics.get(),
System.currentTimeMillis());
+ assertEquals(queryRequest.getRequestId(), -42L);
+
+ QueryScheduler queryScheduler = mock(QueryScheduler.class);
+ when(queryScheduler.submit(any())).thenReturn(SettableFuture.create());
+ EmbeddedChannel channel = new EmbeddedChannel(
+ new InstanceRequestHandler("server01", new PinotConfiguration(),
queryScheduler, mock(AccessControl.class),
+ ThreadAccountantUtils.getNoOpAccountant()));
+ channel.writeInbound(Unpooled.wrappedBuffer(serialize(asSeenByOldServer)));
+ verify(queryScheduler).submit(any());
+ channel.finishAndReleaseAll();
+ }
+
+ private static byte[] serialize(InstanceRequest instanceRequest)
+ throws Exception {
+ return new TSerializer(new
TCompactProtocol.Factory()).serialize(instanceRequest);
+ }
+
@Test
public void testCancelQuery()
throws InterruptedException {
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 86a12592192..379326ca7fa 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
@@ -19,8 +19,23 @@
package org.apache.pinot.core.transport;
import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.SettableFuture;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.util.ReferenceCountUtil;
+import java.io.IOException;
+import java.io.InputStream;
import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.SocketTimeoutException;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -31,6 +46,7 @@ import org.apache.pinot.common.metrics.BrokerMeter;
import org.apache.pinot.common.metrics.BrokerMetrics;
import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.common.request.BrokerRequest;
+import org.apache.pinot.common.request.InstanceRequest;
import org.apache.pinot.common.utils.DataSchema;
import org.apache.pinot.core.common.datatable.DataTableBuilder;
import org.apache.pinot.core.common.datatable.DataTableBuilderFactory;
@@ -47,6 +63,8 @@ import org.apache.pinot.spi.metrics.PinotMetricsRegistry;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.sql.parsers.CalciteSqlCompiler;
import org.apache.pinot.util.TestUtils;
+import org.apache.thrift.TDeserializer;
+import org.apache.thrift.protocol.TCompactProtocol;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
@@ -236,6 +254,12 @@ public class QueryRoutingTest {
assertEquals(serverResponse.getDeserializationTimeMs(), 0);
// Query should time out
assertTrue(System.currentTimeMillis() - startTimeMs >= 1000);
+ // A request left unanswered until the deadline produces a timeout and
nothing else: the send succeeded and the
+ // channel was never torn down, so no failed server is reported. This is
why the broker's failure detector cannot
+ // rely on getFailedServer() alone -- see
FailureDetector#notifyServerNotResponded.
+ assertEquals(asyncQueryResponse.getStatus(),
QueryResponse.Status.TIMED_OUT);
+ assertNull(asyncQueryResponse.getException());
+ assertNull(asyncQueryResponse.getFailedServer());
_requestCount += 2;
waitForStatsUpdate(_requestCount);
assertEquals(_serverRoutingStatsManager.fetchNumInFlightRequestsForServer(serverId).intValue(),
0);
@@ -746,6 +770,252 @@ public class QueryRoutingTest {
PinotMetricUtils.makePinotMetricName(BrokerMetrics.class, fullName),
meter.getUnit(), TimeUnit.SECONDS).count();
}
+ /// A server that answers queries answers a ping over the channel they use.
Before any query there is no channel to
+ /// ping over, and the ping must not open one either: that would make a
server that only serves multi-stage queries
+ /// look queried over this path.
+ @Test
+ public void testPingIsAnsweredByLiveServer()
+ throws Exception {
+ long requestId = 123;
+ _queryServer = getQueryServer(0, dataTableBytesWithRequestId(requestId));
+ initializeTestFixtures(startAndGetPort(_queryServer));
+ QueryRouter queryRouter = newIsolatedQueryRouter();
+ try {
+ assertFalse(queryRouter.ping(_serverInstance, 5_000L));
+ assertFalse(queryRouter.hasChannel(_serverInstance));
+
+ assertNotNull(queryRouter.submitQuery(requestId, "testTable",
BROKER_REQUEST, _routingTable, null, null, 5_000L)
+
.getFinalResponses().get(_offlineServerRoutingInstance).getDataTable());
+ assertTrue(queryRouter.ping(_serverInstance, 5_000L));
+ } finally {
+ queryRouter.shutDown();
+ }
+ }
+
+ /// The property that keeps busy servers in routing: the ping is answered
from the server's network thread, so it
+ /// comes back at once even while a query to that server is stuck.
+ @Test
+ public void testPingIsAnsweredPromptlyWhileTheQueryEngineIsStuck() {
+ _queryServer = QueryServerTestUtils.newQueryServer(invocation ->
SettableFuture.<byte[]>create());
+ initializeTestFixtures(startAndGetPort(_queryServer));
+ QueryRouter queryRouter = newIsolatedQueryRouter();
+ try {
+ AsyncQueryResponse stuckQuery =
+ queryRouter.submitQuery(123, "testTable", BROKER_REQUEST,
_routingTable, null, null, 60_000L);
+
+ long startTimeMs = System.currentTimeMillis();
+ assertTrue(queryRouter.ping(_serverInstance, 5_000L));
+ assertTrue(System.currentTimeMillis() - startTimeMs < 2_000L,
+ "A server whose query engine is stuck must still answer a ping
promptly");
+ assertEquals(stuckQuery.getStatus(), QueryResponse.Status.IN_PROGRESS);
+ } finally {
+ queryRouter.shutDown();
+ }
+ }
+
+ /// A frozen JVM or a vanished node: the kernel (or nothing at all)
completes the TCP handshake, but nothing ever
+ /// reads the request. Modeled by a listening socket that is never accepted
from. The unanswered ping must also close
+ /// the channel, or every later attempt would write into the same possibly
half-open connection.
+ @Test
+ public void testPingGoesUnansweredByFrozenServer()
+ throws Exception {
+ long timeoutMs = 1_000L;
+ QueryRouter queryRouter = newIsolatedQueryRouter();
+ try (ServerSocket frozenServer = new ServerSocket(0)) {
+ initializeTestFixtures(frozenServer.getLocalPort());
+ // Opens the channel, and is never answered.
+ AsyncQueryResponse query =
+ queryRouter.submitQuery(123, "testTable", BROKER_REQUEST,
_routingTable, null, null, 60_000L);
+
+ long startTimeMs = System.currentTimeMillis();
+ assertFalse(queryRouter.ping(_serverInstance, timeoutMs));
+ long elapsedMs = System.currentTimeMillis() - startTimeMs;
+ assertTrue(elapsedMs >= timeoutMs && elapsedMs < timeoutMs + 2_000L,
+ "A ping to a frozen server must fail after the ping timeout, not
sooner or much later; took " + elapsedMs);
+ // Closing the channel also ends the query waiting on it, instead of
leaving it to its own timeout.
+ TestUtils.waitForCondition(aVoid -> query.getStatus() !=
QueryResponse.Status.IN_PROGRESS, 5_000L,
+ "The query on the closed channel did not end");
+
+ // Only now pick up the connection: past the query and the ping, the
broker must have closed its end.
+ frozenServer.setSoTimeout(5_000);
+ try (Socket connection = frozenServer.accept()) {
+ connection.setSoTimeout(5_000);
+ InputStream inputStream = connection.getInputStream();
+ while (inputStream.read() != -1) {
+ // Skip the requests.
+ }
+ } catch (SocketTimeoutException e) {
+ fail("The channel to a server that left a ping unanswered must be
closed");
+ }
+ } finally {
+ queryRouter.shutDown();
+ }
+ }
+
+ /// Every channel queries use has to answer. A server whose OFFLINE channel
answers while its REALTIME channel stays
+ /// silent -- left half-open, say, when the server came back under a new
address -- fails the ping, as REALTIME
+ /// queries would time out. Only the silent channel is closed: the one that
answered still carries queries.
+ @Test
+ public void testPingFailsWhenAnyChannelIsSilent()
+ throws Exception {
+ long requestId = 123;
+ _queryServer = getQueryServer(0, dataTableBytesWithRequestId(requestId));
+ initializeTestFixtures(startAndGetPort(_queryServer));
+ QueryRouter queryRouter = newIsolatedQueryRouter();
+ EmbeddedChannel silentChannel = new EmbeddedChannel();
+ try {
+ assertNotNull(queryRouter.submitQuery(requestId, "testTable",
BROKER_REQUEST, _routingTable, null, null, 5_000L)
+
.getFinalResponses().get(_offlineServerRoutingInstance).getDataTable());
+ ServerChannels serverChannels = queryRouter.getServerChannels();
+ ServerChannels.ServerChannel realtimeChannel =
+
serverChannels.getOrCreateServerChannel(_realtimeServerRoutingInstance);
+ realtimeChannel._openedByQuery = true;
+ realtimeChannel.setChannel(silentChannel);
+
+ assertFalse(queryRouter.ping(_serverInstance, 1_000L));
+ assertFalse(silentChannel.isOpen(), "The silent channel must be closed");
+ // A close runs on the channel's event loop, so give a wrong one time to
land.
+
assertFalse(serverChannels.getOrCreateServerChannel(_offlineServerRoutingInstance)._channel.closeFuture()
+ .await(500, TimeUnit.MILLISECONDS), "The channel that answered must
stay open");
+ } finally {
+ silentChannel.finishAndReleaseAll();
+ queryRouter.shutDown();
+ }
+ }
+
+ /// Each table type has its own channel to a server. A server queried only
for REALTIME tables -- as in the incident
+ /// this guards -- has no OFFLINE channel, and must still be found and
pinged over the channel queries use.
+ @Test
+ public void testRealtimeOnlyServerIsFoundAndPinged()
+ throws Exception {
+ long requestId = 123;
+ DataTable dataTable = DataTableBuilderFactory.getEmptyDataTable();
+ dataTable.getMetadata().put(MetadataKey.REQUEST_ID.getName(),
Long.toString(requestId));
+ _queryServer = getQueryServer(0, dataTable.toBytes());
+ initializeTestFixtures(startAndGetPort(_queryServer));
+ assertFalse(_queryRouter.hasChannel(_serverInstance));
+
+ _queryRouter.submitQuery(requestId, "testTable", null, null,
BROKER_REQUEST, _routingTable, 5_000L)
+ .getFinalResponses();
+ _requestCount += 2;
+ waitForStatsUpdate(_requestCount);
+
+ assertTrue(_queryRouter.hasChannel(_serverInstance),
+ "A REALTIME channel must count, or a REALTIME-only server is never
checked");
+ assertTrue(_queryRouter.ping(_serverInstance, 5_000L));
+ }
+
+ /// A channel that fails at once must not end the wait for another one that
is still due to answer, and get that one
+ /// closed as if it were silent -- along with the queries in flight on it.
+ @Test
+ public void testChannelFailingAtOnceDoesNotCloseAnotherStillDueToAnswer()
+ throws Exception {
+ _queryServer = new QueryServer(0, null, new SlowPingHandler(300L));
+ initializeTestFixtures(startAndGetPort(_queryServer));
+ QueryRouter queryRouter = newIsolatedQueryRouter();
+ // Fails every write at once, as a connection that was just reset does.
+ EmbeddedChannel failingChannel = new EmbeddedChannel(new
ChannelOutboundHandlerAdapter() {
+ @Override
+ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise
promise) {
+ ReferenceCountUtil.release(msg);
+ promise.setFailure(new IOException("Connection reset"));
+ }
+ });
+ try {
+ // Opens the OFFLINE channel, and stays in flight: the server never
answers queries.
+ AsyncQueryResponse query =
+ queryRouter.submitQuery(123, "testTable", BROKER_REQUEST,
_routingTable, null, null, 60_000L);
+ ServerChannels.ServerChannel realtimeChannel =
+
queryRouter.getServerChannels().getOrCreateServerChannel(_realtimeServerRoutingInstance);
+ realtimeChannel._openedByQuery = true;
+ realtimeChannel.setChannel(failingChannel);
+
+ assertFalse(queryRouter.ping(_serverInstance, 5_000L));
+ // A close runs on the channel's event loop, so give a wrong one time to
land.
+
assertFalse(queryRouter.getServerChannels().getOrCreateServerChannel(_offlineServerRoutingInstance)._channel
+ .closeFuture().await(500, TimeUnit.MILLISECONDS), "The channel that
answered late must stay open");
+ assertEquals(query.getStatus(), QueryResponse.Status.IN_PROGRESS);
+ } finally {
+ failingChannel.finishAndReleaseAll();
+ queryRouter.shutDown();
+ }
+ }
+
+ /// A ping that stays silent closes the connection it went out on, not
whichever connection its channel holds by
+ /// then. Here the ping's connection drops while the ping waits, and a query
reconnects and sends over a fresh
+ /// connection before the ping is done: the fresh connection, and the query
on it, must survive.
+ @Test
+ public void testFailedPingClosesOnlyItsOwnConnection()
+ throws Exception {
+ // Nothing listens on this port: every connection below is planted.
+ initializeTestFixtures(12345);
+ QueryRouter queryRouter = newIsolatedQueryRouter();
+ ServerChannels.ServerChannel offlineChannel =
+
queryRouter.getServerChannels().getOrCreateServerChannel(_offlineServerRoutingInstance);
+ offlineChannel._openedByQuery = true;
+ EmbeddedChannel freshConnection = new EmbeddedChannel(
+ new DataTableHandler(queryRouter,
ThreadAccountantUtils.getNoOpAccountant(), _offlineServerRoutingInstance));
+ List<AsyncQueryResponse> queriesOnFreshConnection = new ArrayList<>();
+ EmbeddedChannel pingConnection = new EmbeddedChannel(new
ChannelOutboundHandlerAdapter() {
+ @Override
+ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise
promise) {
+ ReferenceCountUtil.release(msg);
+ promise.setSuccess();
+ // The connection drops, and its inactive callback runs to the end,
failing the ping that waits on it. (Run here
+ // directly: Netty would defer the callback of a channel closed in the
middle of its own write.)
+ ctx.channel().close();
+ queryRouter.markServerUnavailable(_offlineServerRoutingInstance, new
IOException("Connection dropped"));
+ // Before the ping gets to its cleanup, a query reconnects and sends
over a fresh connection. This runs on the
+ // thread that holds the channel lock for the ping, so the query gets
the lock too.
+ offlineChannel.setChannel(freshConnection);
+ queriesOnFreshConnection.add(
+ queryRouter.submitQuery(123, "testTable", BROKER_REQUEST,
_routingTable, null, null, 60_000L));
+ }
+ });
+ offlineChannel.setChannel(pingConnection);
+ try {
+ assertFalse(queryRouter.ping(_serverInstance, 5_000L));
+ assertTrue(freshConnection.isOpen(), "The ping must not close the
connection that replaced its own");
+ assertEquals(queriesOnFreshConnection.size(), 1);
+ assertEquals(queriesOnFreshConnection.get(0).getStatus(),
QueryResponse.Status.IN_PROGRESS);
+ } finally {
+ pingConnection.finishAndReleaseAll();
+ freshConnection.finishAndReleaseAll();
+ queryRouter.shutDown();
+ }
+ }
+
+ /// Answers pings after a delay, and queries never.
+ @ChannelHandler.Sharable
+ private static class SlowPingHandler extends
SimpleChannelInboundHandler<ByteBuf> {
+ private final long _delayMs;
+
+ SlowPingHandler(long delayMs) {
+ _delayMs = delayMs;
+ }
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg)
+ throws Exception {
+ byte[] requestBytes = new byte[msg.readableBytes()];
+ msg.readBytes(requestBytes);
+ InstanceRequest instanceRequest = new InstanceRequest();
+ new TDeserializer(new
TCompactProtocol.Factory()).deserialize(instanceRequest, requestBytes);
+ if (instanceRequest.isPing()) {
+ byte[] responseBytes =
dataTableBytesWithRequestId(instanceRequest.getRequestId());
+ ctx.executor().schedule(() ->
ctx.writeAndFlush(Unpooled.wrappedBuffer(responseBytes)), _delayMs,
+ TimeUnit.MILLISECONDS);
+ }
+ }
+ }
+
+ private static byte[] dataTableBytesWithRequestId(long requestId)
+ throws IOException {
+ DataTable dataTable = DataTableBuilderFactory.getEmptyDataTable();
+ dataTable.getMetadata().put(MetadataKey.REQUEST_ID.getName(),
Long.toString(requestId));
+ return dataTable.toBytes();
+ }
+
private void waitForStatsUpdate(long taskCount) {
TestUtils.waitForCondition(aVoid -> {
return (_serverRoutingStatsManager.getCompletedTaskCount() == taskCount);
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/transport/QueryServerTestUtils.java
b/pinot-core/src/test/java/org/apache/pinot/core/transport/QueryServerTestUtils.java
new file mode 100644
index 00000000000..53cb513e8b6
--- /dev/null
+++
b/pinot-core/src/test/java/org/apache/pinot/core/transport/QueryServerTestUtils.java
@@ -0,0 +1,69 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.core.transport;
+
+import java.net.InetSocketAddress;
+import org.apache.pinot.core.query.scheduler.QueryScheduler;
+import org.apache.pinot.server.access.AccessControl;
+import org.apache.pinot.spi.accounting.ThreadAccountantUtils;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.mockito.stubbing.Answer;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+
+/// Helpers for standing up a real Netty [QueryServer] in a test.
+///
+/// Lives in `org.apache.pinot.core.transport` because
[QueryServer#getChannel] and the `host`/`port`
+/// [ServerInstance] constructor are package-private; pinot-broker consumes
this through the pinot-core test jar.
+public class QueryServerTestUtils {
+ private QueryServerTestUtils() {
+ }
+
+ /// Builds an unstarted query server whose scheduler answers every
submission with `submitAnswer`.
+ public static QueryServer newQueryServer(Answer<?> submitAnswer) {
+ return newQueryServer(0, submitAnswer);
+ }
+
+ /// Same as [#newQueryServer(Answer)], on a given port -- for a test that
brings a server back where it was.
+ public static QueryServer newQueryServer(int port, Answer<?> submitAnswer) {
+ QueryScheduler queryScheduler = mock(QueryScheduler.class);
+ when(queryScheduler.submit(any())).thenAnswer(submitAnswer);
+ InstanceRequestHandler handler =
+ new InstanceRequestHandler("testServer", new PinotConfiguration(),
queryScheduler, mock(AccessControl.class),
+ ThreadAccountantUtils.getNoOpAccountant());
+ return new QueryServer(port, null, handler);
+ }
+
+ /// Starts the server and returns the port it bound to.
+ ///
+ /// Binds on port 0 and reads the port back, rather than picking a free port
up front, which would leave a window
+ /// for another process on the machine to take it. [QueryServer#start]
blocks on `bind().sync()`, so the server is
+ /// accepting connections when this returns.
+ public static int startAndGetPort(QueryServer server) {
+ server.start();
+ return ((InetSocketAddress) server.getChannel().localAddress()).getPort();
+ }
+
+ public static ServerInstance serverInstance(String hostname, int port) {
+ return new ServerInstance(hostname, port);
+ }
+}
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 7fec4b33fed..bbbb0b7d971 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
@@ -25,6 +25,7 @@ import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.util.concurrent.GenericFutureListener;
import java.net.InetSocketAddress;
+import java.util.concurrent.TimeoutException;
import org.apache.pinot.common.config.NettyConfig;
import org.apache.pinot.common.metrics.BrokerMetrics;
import org.apache.pinot.common.request.BrokerRequest;
@@ -49,6 +50,7 @@ import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertSame;
import static org.testng.Assert.assertThrows;
import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
public class ServerChannelsTest {
@@ -82,7 +84,6 @@ public class ServerChannelsTest {
new ServerRoutingInstance("localhost",
dummyServer.getAddress().getPort(), TableType.REALTIME);
ServerChannels serverChannels =
new ServerChannels(queryRouter, nettyConfig, null,
ThreadAccountantUtils.getNoOpAccountant());
- serverChannels.connect(serverRoutingInstance);
final long requestId = System.currentTimeMillis();
@@ -98,6 +99,37 @@ public class ServerChannelsTest {
}
}
+ /// Connecting is bounded by the caller's timeout. Without that bound a
connect to a host that drops packets -- the
+ /// host the failure detector pings -- would block for Netty's 30 s default.
192.0.2.1 is reserved for
+ /// documentation (RFC 5737) and routes nowhere: depending on the network
the connect either hangs, which the timeout
+ /// must cut short, or fails at once.
+ @Test
+ public void testConnectIsBoundedByTheCallerTimeout()
+ throws Exception {
+ ServerChannels serverChannels =
+ new ServerChannels(mock(QueryRouter.class), null, null,
ThreadAccountantUtils.getNoOpAccountant());
+ try {
+ InstanceRequest instanceRequest = new InstanceRequest();
+ instanceRequest.setRequestId(1L);
+ instanceRequest.setQuery(new BrokerRequest());
+ long timeoutMs = 500L;
+ long startTimeMs = System.currentTimeMillis();
+ try {
+ serverChannels.sendRequest("dummy_table_name",
mock(AsyncQueryResponse.class),
+ new ServerRoutingInstance("192.0.2.1", 8098, TableType.OFFLINE),
instanceRequest, timeoutMs);
+ fail("Connecting to an unroutable address must fail");
+ } catch (Exception e) {
+ // A network that rejects the address fails the connect at once. One
that drops the packets must hit the bound.
+ if (System.currentTimeMillis() - startTimeMs >= timeoutMs) {
+ assertTrue(e instanceof TimeoutException, "A hanging connect must
end in the caller's timeout, got: " + e);
+ }
+ }
+ assertTrue(System.currentTimeMillis() - startTimeMs < 5_000L, "The
connect must be bounded by the timeout");
+ } finally {
+ serverChannels.shutDown();
+ }
+ }
+
@Test
public void testChannelsShareBufferAllocator() {
ServerChannels serverChannels =
@@ -149,6 +181,10 @@ public class ServerChannelsTest {
// Simulate write failure
when(mockFuture.isSuccess()).thenReturn(false);
when(mockFuture.cause()).thenReturn(new OutOfMemoryError("Direct buffer
memory"));
+ // By the time the listener runs, a reconnect may have replaced the
channel. Only the connection written to is
+ // closed.
+ Channel replacementChannel = mock(Channel.class);
+ serverChannel.setChannel(replacementChannel);
try {
listenerCaptor.getValue().operationComplete(mockFuture);
@@ -157,6 +193,7 @@ public class ServerChannelsTest {
}
verify(mockChannel).close();
+ verify(replacementChannel, never()).close();
verify(asyncQueryResponse).markServerUnavailable(any(ServerRoutingInstance.class),
any(Exception.class));
verify(asyncQueryResponse,
never()).markRequestSent(any(ServerRoutingInstance.class), any(Integer.class));
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 363490d95b6..e69bc0bd393 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
@@ -1352,7 +1352,8 @@ public class CommonConstants {
// Do not detect any failure
NO_OP,
- // Detect connection failures
+ // Detect connection failures, and optionally servers that leave a
query unanswered and then fail a ping (see
+ // CONFIG_OF_ENABLE_PING_ON_TIMEOUT)
CONNECTION,
// Use the custom failure detector of the configured class name
@@ -1371,6 +1372,23 @@ public class CommonConstants {
public static final double DEFAULT_RETRY_DELAY_FACTOR = 2.0;
public static final String CONFIG_OF_MAX_RETRIES =
"pinot.broker.failure.detector.max.retries";
public static final int DEFAULT_MAX_RETRIES = 10;
+
+ /// When a single-stage query times out on a server, ping the server and
mark it unhealthy if the ping goes
+ /// unanswered. Unhealthy servers are then retried with a ping too,
instead of a TCP connect. This is how a server
+ /// whose node is gone or whose JVM is frozen is detected, since such a
server keeps its connection open and
+ /// raises no connection failure. A server that is merely busy still
answers the ping, so it is left alone.
+ ///
+ /// Enable only once every server runs a version that answers pings. An
older server runs a ping like a query,
+ /// behind the queries already queued on it, so a busy one can miss the
ping timeout and be taken out of routing.
+ public static final String CONFIG_OF_ENABLE_PING_ON_TIMEOUT =
+ "pinot.broker.failure.detector.enable.ping.on.timeout";
+ public static final boolean DEFAULT_ENABLE_PING_ON_TIMEOUT = false;
+
+ /// With pings enabled, how long to wait for a server to answer one
before treating it as unhealthy, both when
+ /// checking a server after a query timeout and when retrying an
unhealthy one. Keep it above the longest
+ /// 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;
}
// Configs related to AdaptiveServerSelection.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
