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 d085f946779 Recover unhealthy MSE workers that are not servers in the
routing manager (#20005)
d085f946779 is described below
commit d085f946779eec3287398af31ea0effa3754aed7
Author: Yash Mayya <[email protected]>
AuthorDate: Mon Oct 26 11:28:59 2026 -0700
Recover unhealthy MSE workers that are not servers in the routing manager
(#20005)
---
.../requesthandler/GrpcBrokerRequestHandler.java | 4 +-
.../MultiStageBrokerRequestHandler.java | 24 ++++-
.../SingleConnectionBrokerRequestHandler.java | 4 +-
.../MultiStageBrokerRequestHandlerTest.java | 111 ++++++++++++++++++++-
.../query/service/dispatch/QueryDispatcher.java | 57 +++++++++--
.../service/dispatch/QueryDispatcherTest.java | 59 +++++++++++
6 files changed, 240 insertions(+), 19 deletions(-)
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/GrpcBrokerRequestHandler.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/GrpcBrokerRequestHandler.java
index 59a2bd4a584..dc9e85a616d 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/GrpcBrokerRequestHandler.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/GrpcBrokerRequestHandler.java
@@ -208,9 +208,11 @@ public class GrpcBrokerRequestHandler extends
BaseSingleStageBrokerRequestHandle
private FailureDetector.ServerState retryUnhealthyServer(String instanceId) {
LOGGER.info("Checking gRPC connection to unhealthy server: {}",
instanceId);
ServerInstance serverInstance =
_routingManager.getEnabledServerInstanceMap().get(instanceId);
+ // Not an enabled server of the routing manager, for example a disabled
server or a server of a remote cluster.
+ // Leave it to the other retriers.
if (serverInstance == null) {
LOGGER.info("Failed to find enabled server: {} in routing manager,
skipping the retry", instanceId);
- return FailureDetector.ServerState.UNHEALTHY;
+ return FailureDetector.ServerState.UNKNOWN;
}
String hostnamePort = String.format("%s_%d", serverInstance.getHostname(),
serverInstance.getGrpcPort());
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
index 6823d150e11..c68506ca14e 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
@@ -1262,13 +1262,27 @@ public class MultiStageBrokerRequestHandler extends
BaseBrokerRequestHandler {
/// Check if a server that was previously detected as unhealthy is now
healthy.
public FailureDetector.ServerState retryUnhealthyServer(String instanceId) {
LOGGER.info("Checking gRPC connection to unhealthy server: {}",
instanceId);
- ServerInstance serverInstance =
_routingManager.getEnabledServerInstanceMap().get(instanceId);
- if (serverInstance == null) {
- LOGGER.info("Failed to find enabled server: {} in routing manager,
skipping the retry", instanceId);
- return FailureDetector.ServerState.UNHEALTHY;
+ ServerInstance serverInstance = getEnabledServerInstance(instanceId);
+ if (serverInstance != null) {
+ return _queryDispatcher.checkConnectivityToInstance(serverInstance);
}
+ // Queries are also dispatched to workers that no routing manager knows,
for example workers that a custom worker
+ // manager picks. Check them at the address that a query was last
dispatched to.
+ return _queryDispatcher.checkConnectivityToInstance(instanceId);
+ }
- return _queryDispatcher.checkConnectivityToInstance(serverInstance);
+ /// Returns the enabled server with the given instance id: a server of this
cluster, or with multi-cluster routing, a
+ /// server of a remote cluster.
+ @Nullable
+ private ServerInstance getEnabledServerInstance(String instanceId) {
+ ServerInstance serverInstance =
_routingManager.getEnabledServerInstanceMap().get(instanceId);
+ if (serverInstance == null && _multiClusterRoutingContext != null) {
+ RoutingManager multiClusterRoutingManager =
_multiClusterRoutingContext.getMultiClusterRoutingManager();
+ if (multiClusterRoutingManager != null) {
+ serverInstance =
multiClusterRoutingManager.getEnabledServerInstanceMap().get(instanceId);
+ }
+ }
+ return serverInstance;
}
public static boolean isYellowError(QueryException e) {
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 cceced717cc..d927e675ef3 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
@@ -1048,9 +1048,11 @@ public class SingleConnectionBrokerRequestHandler
extends BaseSingleStageBrokerR
LOGGER.info("Retrying unhealthy server: {}", instanceId);
ServerInstance serverInstance =
_routingManager.getEnabledServerInstanceMap().get(instanceId);
+ // Not an enabled server of the routing manager, for example a disabled
server or a server of a remote cluster.
+ // Leave it to the other retriers.
if (serverInstance == null) {
LOGGER.info("Failed to find enabled server: {} in routing manager,
skipping the retry", instanceId);
- return FailureDetector.ServerState.UNHEALTHY;
+ return FailureDetector.ServerState.UNKNOWN;
}
// Could occur if the cluster is only serving multi-stage queries
diff --git
a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandlerTest.java
b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandlerTest.java
index 9cd5afcee77..149228f41d5 100644
---
a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandlerTest.java
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandlerTest.java
@@ -32,11 +32,15 @@ import org.apache.pinot.broker.queryquota.QueryQuotaManager;
import org.apache.pinot.common.config.provider.TableCache;
import org.apache.pinot.common.datatable.StatMap;
import org.apache.pinot.common.failuredetector.FailureDetector;
+import org.apache.pinot.common.failuredetector.FailureDetectorFactory;
import org.apache.pinot.common.metrics.BrokerMetrics;
import org.apache.pinot.common.response.BrokerResponse;
import org.apache.pinot.common.response.broker.BrokerResponseNativeV2;
import org.apache.pinot.common.response.broker.ResultTable;
+import org.apache.pinot.core.routing.MultiClusterRoutingContext;
import org.apache.pinot.core.routing.RoutingManager;
+import org.apache.pinot.core.transport.ServerInstance;
+import
org.apache.pinot.core.transport.server.routing.stats.ServerRoutingStatsManager;
import org.apache.pinot.query.QueryEnvironmentTestBase;
import org.apache.pinot.query.mailbox.MailboxService;
import org.apache.pinot.query.planner.physical.DispatchablePlanFragment;
@@ -67,6 +71,8 @@ import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.NetUtils;
import org.apache.pinot.sql.parsers.CalciteSqlParser;
import org.apache.pinot.sql.parsers.SqlNodeAndOptions;
+import org.apache.pinot.util.TestUtils;
+import org.mockito.MockedConstruction;
import org.mockito.Mockito;
import org.testng.Assert;
import org.testng.annotations.Test;
@@ -75,7 +81,11 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
@@ -383,6 +393,12 @@ public class MultiStageBrokerRequestHandlerTest extends
QueryEnvironmentTestBase
private static MultiStageBrokerRequestHandler newHandler(PinotConfiguration
config)
throws Exception {
+ return newHandler(config, mock(RoutingManager.class),
mock(FailureDetector.class), null);
+ }
+
+ private static MultiStageBrokerRequestHandler newHandler(PinotConfiguration
config, RoutingManager routingManager,
+ FailureDetector failureDetector, @Nullable MultiClusterRoutingContext
multiClusterRoutingContext)
+ throws Exception {
config.setProperty(MultiStageQueryRunner.KEY_OF_QUERY_RUNNER_HOSTNAME,
"localhost");
config.setProperty(MultiStageQueryRunner.KEY_OF_QUERY_RUNNER_PORT,
Integer.toString(NetUtils.findOpenPort()));
BrokerQueryEventListenerFactory.init(config);
@@ -394,9 +410,10 @@ public class MultiStageBrokerRequestHandlerTest extends
QueryEnvironmentTestBase
when(queryQuotaManager.acquireApplication(anyString())).thenReturn(true);
return new MultiStageBrokerRequestHandler(config, "testBrokerId", new
BrokerRequestIdGenerator(),
- mock(RoutingManager.class), new AllowAllAccessControlFactory(),
queryQuotaManager,
- mock(TableCache.class), mock(MultiStageQueryThrottler.class),
mock(FailureDetector.class),
- ThreadAccountantUtils.getNoOpAccountant(), null,
mock(WorkerManager.class), mock(WorkerManager.class)) {
+ routingManager, new AllowAllAccessControlFactory(), queryQuotaManager,
+ mock(TableCache.class), mock(MultiStageQueryThrottler.class),
failureDetector,
+ ThreadAccountantUtils.getNoOpAccountant(), multiClusterRoutingContext,
mock(WorkerManager.class),
+ mock(WorkerManager.class)) {
@Override
public void start() {
}
@@ -408,6 +425,94 @@ public class MultiStageBrokerRequestHandlerTest extends
QueryEnvironmentTestBase
};
}
+ /// The failure detector keeps an instance unhealthy while any retrier
returns UNHEALTHY for it. A multi-stage worker
+ /// that is not an enabled server of the routing manager, for example a
server of a remote cluster, must recover
+ /// once the dispatch channel to it is READY, with the retriers of all
request handlers registered.
+ @Test
+ public void testFailureDetectorRecoversWorkerOutsideRoutingManager()
+ throws Exception {
+ String workerId = "Server_remote_8098";
+ PinotConfiguration failureDetectorConfig = new PinotConfiguration();
+
failureDetectorConfig.setProperty(CommonConstants.Broker.FailureDetector.CONFIG_OF_TYPE,
+ CommonConstants.Broker.FailureDetector.Type.CONNECTION.name());
+
failureDetectorConfig.setProperty(CommonConstants.Broker.FailureDetector.CONFIG_OF_RETRY_INITIAL_DELAY_MS,
100);
+
failureDetectorConfig.setProperty(CommonConstants.Broker.FailureDetector.CONFIG_OF_RETRY_DELAY_FACTOR,
1);
+ // Recover only through the retriers, not because the retries ran out
+
failureDetectorConfig.setProperty(CommonConstants.Broker.FailureDetector.CONFIG_OF_MAX_RETRIES,
1_000);
+ FailureDetector failureDetector =
+ FailureDetectorFactory.getFailureDetector(failureDetectorConfig,
BrokerMetrics.noop());
+ failureDetector.registerHealthyServerNotifier(instanceId -> {
+ });
+ failureDetector.registerUnhealthyServerNotifier(instanceId -> {
+ });
+ failureDetector.start();
+ RoutingManager routingManager = mock(RoutingManager.class);
+ when(routingManager.getEnabledServerInstanceMap()).thenReturn(Map.of());
+ BrokerQueryEventListenerFactory.init(new PinotConfiguration());
+ // Like the broker starter, register the retriers of the single-stage
handlers first
+ SingleConnectionBrokerRequestHandler singleConnectionHandler =
+ new SingleConnectionBrokerRequestHandler(new PinotConfiguration(),
"testBrokerId",
+ new BrokerRequestIdGenerator(), routingManager, new
AllowAllAccessControlFactory(), null, null, null, null,
+ mock(ServerRoutingStatsManager.class), failureDetector,
ThreadAccountantUtils.getNoOpAccountant(), null,
+ null);
+ GrpcBrokerRequestHandler grpcHandler =
+ new GrpcBrokerRequestHandler(new PinotConfiguration(), "testBrokerId",
new BrokerRequestIdGenerator(),
+ routingManager, new AllowAllAccessControlFactory(), null, null,
failureDetector,
+ ThreadAccountantUtils.getNoOpAccountant(), null);
+ AtomicReference<FailureDetector.ServerState> workerState =
+ new AtomicReference<>(FailureDetector.ServerState.UNHEALTHY);
+ try (MockedConstruction<QueryDispatcher> dispatchers =
mockConstruction(QueryDispatcher.class,
+ (dispatcher, context) ->
when(dispatcher.checkConnectivityToInstance(workerId)).thenAnswer(
+ invocation -> workerState.get()))) {
+ newHandler(new PinotConfiguration(), routingManager, failureDetector,
null);
+ QueryDispatcher dispatcher = dispatchers.constructed().get(0);
+ failureDetector.markServerUnhealthy(workerId, "remote");
+
+ // The dispatch channel to the worker is not READY
+ verify(dispatcher,
timeout(5_000L).atLeast(3)).checkConnectivityToInstance(workerId);
+ assertEquals(failureDetector.getUnhealthyServers(), Set.of(workerId));
+
+ workerState.set(FailureDetector.ServerState.HEALTHY);
+ TestUtils.waitForCondition(aVoid ->
failureDetector.getUnhealthyServers().isEmpty(), 5_000L,
+ "The worker did not recover");
+ } finally {
+ failureDetector.stop();
+ singleConnectionHandler.shutDown();
+ grpcHandler.shutDown();
+ }
+ }
+
+ /// The retrier checks a server at its current address, in this cluster or
with multi-cluster routing in a remote
+ /// cluster. It checks other instances at the address that a query was last
dispatched to.
+ @Test
+ public void testRetryUnhealthyServerChecksCurrentAddress()
+ throws Exception {
+ ServerInstance localServer = mock(ServerInstance.class);
+ ServerInstance remoteServer = mock(ServerInstance.class);
+ RoutingManager routingManager = mock(RoutingManager.class);
+
when(routingManager.getEnabledServerInstanceMap()).thenReturn(Map.of("Server_local_8098",
localServer));
+ RoutingManager multiClusterRoutingManager = mock(RoutingManager.class);
+ when(multiClusterRoutingManager.getEnabledServerInstanceMap()).thenReturn(
+ Map.of("Server_local_8098", localServer, "Server_remote_8098",
remoteServer));
+ MultiClusterRoutingContext multiClusterRoutingContext =
+ new MultiClusterRoutingContext(Map.of(), routingManager,
multiClusterRoutingManager, Set.of());
+ try (MockedConstruction<QueryDispatcher> dispatchers =
mockConstruction(QueryDispatcher.class)) {
+ MultiStageBrokerRequestHandler handler = newHandler(new
PinotConfiguration(), routingManager,
+ mock(FailureDetector.class), multiClusterRoutingContext);
+ QueryDispatcher dispatcher = dispatchers.constructed().get(0);
+
when(dispatcher.checkConnectivityToInstance(localServer)).thenReturn(FailureDetector.ServerState.HEALTHY);
+
when(dispatcher.checkConnectivityToInstance(remoteServer)).thenReturn(FailureDetector.ServerState.UNHEALTHY);
+
when(dispatcher.checkConnectivityToInstance("Server_worker_8098")).thenReturn(
+ FailureDetector.ServerState.HEALTHY);
+
+ assertEquals(handler.retryUnhealthyServer("Server_local_8098"),
FailureDetector.ServerState.HEALTHY);
+ assertEquals(handler.retryUnhealthyServer("Server_remote_8098"),
FailureDetector.ServerState.UNHEALTHY);
+ assertEquals(handler.retryUnhealthyServer("Server_worker_8098"),
FailureDetector.ServerState.HEALTHY);
+ verify(dispatcher,
never()).checkConnectivityToInstance("Server_local_8098");
+ verify(dispatcher,
never()).checkConnectivityToInstance("Server_remote_8098");
+ }
+ }
+
// Timeout guards against deadlock: if rewriteReduceStageForEmptyLeaves
fails to inline
// all MailboxReceiveNodes, the reducer will block forever polling a mailbox
with no sender.
@Test(timeOut = 10_000)
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/QueryDispatcher.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/QueryDispatcher.java
index d2897d6ef0f..318a4d3484a 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/QueryDispatcher.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/QueryDispatcher.java
@@ -124,8 +124,13 @@ public class QueryDispatcher implements
PinotClusterConfigChangeListener {
private final MailboxService _mailboxService;
private final ExecutorService _executorService;
private final Map<String, DispatchClient> _dispatchClientMap = new
ConcurrentHashMap<>();
- /// Last known dispatch address of each server, by instance id, kept by
[#onServerEnabled] to retire the dispatch
- /// client of an address a server has left.
+ /// Last known dispatch address of each instance, by instance id.
[#onServerEnabled] sets it for the servers of the
+ /// routing manager, and uses it to retire the dispatch client of an address
a server has left. Each dispatch sets it
+ /// too, so that [#checkConnectivityToInstance(String)] can check instances
that no routing manager knows.
+ ///
+ /// Entries are never removed. There is one for each instance id that a
query was dispatched to, or that
+ /// [#onServerEnabled] saw, so the number of instances bounds the size as
long as instance ids are stable, as Helix
+ /// instance ids are. [#onServerEnabled] scans the values, but it runs only
when a server is enabled.
private final Map<String, String> _dispatchAddressByInstanceId = new
ConcurrentHashMap<>();
@Nullable
private final TlsConfig _tlsConfig;
@@ -411,6 +416,7 @@ public class QueryDispatcher implements
PinotClusterConfigChangeListener {
expectedForServer += workerIds.size();
}
}
+ recordDispatchAddress(server);
DispatchClient client = getOrCreateDispatchClient(server);
try {
client.submitWithStream(request, server, deadline, session,
expectedForServer,
@@ -634,24 +640,42 @@ public class QueryDispatcher implements
PinotClusterConfigChangeListener {
}
}
+ /// Checks the dispatch channel to a server, at its current address in a
routing manager.
public FailureDetector.ServerState
checkConnectivityToInstance(ServerInstance serverInstance) {
- String hostname = serverInstance.getHostname();
- int port = serverInstance.getQueryServicePort();
+ return checkConnectivity(serverInstance.getInstanceId(),
+ toHostnamePortKey(serverInstance.getHostname(),
serverInstance.getQueryServicePort()));
+ }
- DispatchClient client = _dispatchClientMap.get(toHostnamePortKey(hostname,
port));
+ /// Checks the dispatch channel to an instance, at the address where a query
was last dispatched to it, or where
+ /// [#onServerEnabled] last found it. The failure detector uses this for
instances that no routing manager knows, for
+ /// example workers that a custom worker manager picks. Returns
[FailureDetector.ServerState#UNKNOWN] if there is no
+ /// dispatch channel to that address, for example because no query was
dispatched to the instance.
+ ///
+ /// Use [#checkConnectivityToInstance(ServerInstance)] when the current
address of the instance is known. This address
+ /// goes stale when the instance moves to another port, until the next
dispatch or [#onServerEnabled] call for it.
+ public FailureDetector.ServerState checkConnectivityToInstance(String
instanceId) {
+ String hostnamePortKey = _dispatchAddressByInstanceId.get(instanceId);
+ if (hostnamePortKey == null) {
+ LOGGER.debug("No dispatch address found for instance: {}", instanceId);
+ return FailureDetector.ServerState.UNKNOWN;
+ }
+ return checkConnectivity(instanceId, hostnamePortKey);
+ }
+
+ private FailureDetector.ServerState checkConnectivity(String instanceId,
String hostnamePortKey) {
+ DispatchClient client = _dispatchClientMap.get(hostnamePortKey);
// Could occur if the cluster is only serving single-stage queries
if (client == null) {
- LOGGER.debug("No DispatchClient found for server with instanceId: {}",
serverInstance.getInstanceId());
+ LOGGER.debug("No DispatchClient found for server with instanceId: {}",
instanceId);
return FailureDetector.ServerState.UNKNOWN;
}
ConnectivityState connectivityState = client.getChannel().getState(true);
if (connectivityState == ConnectivityState.READY) {
- LOGGER.info("Successfully connected to server: {}",
serverInstance.getInstanceId());
+ LOGGER.info("Successfully connected to server: {}", instanceId);
return FailureDetector.ServerState.HEALTHY;
} else {
- LOGGER.info("Still can't connect to server: {}, current state: {}",
serverInstance.getInstanceId(),
- connectivityState);
+ LOGGER.info("Still can't connect to server: {}, current state: {}",
instanceId, connectivityState);
return FailureDetector.ServerState.UNHEALTHY;
}
}
@@ -694,6 +718,7 @@ public class QueryDispatcher implements
PinotClusterConfigChangeListener {
}
};
R request = requestBuilder.apply(serverInstance);
+ recordDispatchAddress(serverInstance);
DispatchClient dispatchClient =
getOrCreateDispatchClient(serverInstance);
try {
@@ -899,6 +924,20 @@ public class QueryDispatcher implements
PinotClusterConfigChangeListener {
_dispatchMaxInboundMessageSizeBytes));
}
+ /// Records the address of an instance that a query is dispatched to, as its
last known address, see
+ /// [#checkConnectivityToInstance(String)]. Only dispatches record it: a
cancel can still go to an address that the
+ /// instance has left.
+ @VisibleForTesting
+ void recordDispatchAddress(QueryServerInstance queryServerInstance) {
+ String instanceId = queryServerInstance.getInstanceId();
+ String hostnamePortKey =
+ toHostnamePortKey(queryServerInstance.getHostname(),
queryServerInstance.getQueryServicePort());
+ // Read first, so that a dispatch to an instance that did not move does
not write the map
+ if (!hostnamePortKey.equals(_dispatchAddressByInstanceId.get(instanceId)))
{
+ _dispatchAddressByInstanceId.put(instanceId, hostnamePortKey);
+ }
+ }
+
/// Called when a server becomes routable, which it does when it starts and
when it is enabled again after it was
/// disabled or excluded. The server may have restarted at a new address, so:
///
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/service/dispatch/QueryDispatcherTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/service/dispatch/QueryDispatcherTest.java
index ee225479c1d..384efb83279 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/service/dispatch/QueryDispatcherTest.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/service/dispatch/QueryDispatcherTest.java
@@ -209,6 +209,65 @@ public class QueryDispatcherTest extends QueryTestSet {
}
}
+ /// The failure detector checks an instance that the routing manager does
not know, for example a server of a remote
+ /// cluster, at the address where a query was last dispatched to it.
+ @Test
+ public void testCheckConnectivityToInstanceAtLastDispatchAddress() {
+ QueryDispatcher dispatcher =
+ new QueryDispatcher(Mockito.mock(MailboxService.class),
Mockito.mock(FailureDetector.class), null, false,
+ Duration.ofSeconds(1));
+ String instanceId = "Server_remote_8098";
+ // Nothing listens there until the server starts
+ int port = QueryTestUtils.getAvailablePort();
+ QueryServerInstance instance = new QueryServerInstance(instanceId,
"localhost", port, 8101);
+ QueryServer queryServer = new QueryServer(port,
Mockito.mock(QueryRunner.class));
+ try {
+ Assert.assertEquals(dispatcher.checkConnectivityToInstance(instanceId),
FailureDetector.ServerState.UNKNOWN);
+
+ // What a dispatch does
+ dispatcher.recordDispatchAddress(instance);
+ ManagedChannel channel =
dispatcher.getOrCreateDispatchClient(instance).getChannel();
+ Assert.assertEquals(dispatcher.checkConnectivityToInstance(instanceId),
FailureDetector.ServerState.UNHEALTHY);
+ TestUtils.waitForCondition(aVoid -> channel.getState(false) ==
ConnectivityState.TRANSIENT_FAILURE, 10_000L,
+ "Connecting did not fail");
+ Assert.assertEquals(dispatcher.checkConnectivityToInstance(instanceId),
FailureDetector.ServerState.UNHEALTHY);
+
+ queryServer.start();
+ TestUtils.waitForCondition(
+ aVoid -> dispatcher.checkConnectivityToInstance(instanceId) ==
FailureDetector.ServerState.HEALTHY, 10_000L,
+ "The instance did not become healthy after its server started");
+
+ // A request other than a dispatch, for example a cancel, does not
change the address to check
+ QueryServerInstance movedInstance =
+ new QueryServerInstance(instanceId, "localhost",
QueryTestUtils.getAvailablePort(), 8101);
+ dispatcher.getOrCreateDispatchClient(movedInstance);
+ Assert.assertEquals(dispatcher.checkConnectivityToInstance(instanceId),
FailureDetector.ServerState.HEALTHY);
+
+ // A dispatch does: the instance moved to another port, where nothing
listens
+ dispatcher.recordDispatchAddress(movedInstance);
+ Assert.assertEquals(dispatcher.checkConnectivityToInstance(instanceId),
FailureDetector.ServerState.UNHEALTHY);
+ } finally {
+ dispatcher.shutdown();
+ queryServer.shutdown();
+ }
+ }
+
+ /// Each dispatch records the address of the instances it reaches, for the
failure detector.
+ @Test
+ public void testDispatchRecordsInstanceAddress()
+ throws Exception {
+ Set<QueryServerInstance> servers = new HashSet<>();
+ DispatchableSubPlan dispatchableSubPlan =
_queryEnvironment.planQuery("SELECT * FROM a");
+ try (QueryThreadContext ignore = QueryThreadContext.openForMseTest()) {
+ _queryDispatcher.submit(REQUEST_ID_GEN.getAndIncrement(),
dispatchableSubPlan, 10_000L, servers, Map.of());
+ }
+ Assert.assertFalse(servers.isEmpty());
+ for (QueryServerInstance server : servers) {
+
Assert.assertEquals(_queryDispatcher.checkConnectivityToInstance(server.getInstanceId()),
+ FailureDetector.ServerState.HEALTHY);
+ }
+ }
+
private static ServerInstance serverInstance(String instanceId, int
queryServerPort) {
InstanceConfig instanceConfig = new InstanceConfig(instanceId);
instanceConfig.setHostName("localhost");
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
