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 c5ac29be0e6 Refresh MSE gRPC channels when a server restarts (#19875)
c5ac29be0e6 is described below
commit c5ac29be0e6a3d5c5fae5c0ed79ece50f87847d9
Author: Yash Mayya <[email protected]>
AuthorDate: Sat Oct 24 15:24:56 2026 -0700
Refresh MSE gRPC channels when a server restarts (#19875)
---
.../broker/broker/helix/BaseBrokerStarter.java | 4 +-
.../routing/manager/BaseBrokerRoutingManager.java | 23 +--
.../routing/manager/BrokerRoutingManagerTest.java | 33 +++--
.../apache/pinot/query/mailbox/MailboxService.java | 18 ++-
.../query/mailbox/channel/ChannelManager.java | 73 ++++++++--
.../query/service/dispatch/DispatchClient.java | 21 +++
.../query/service/dispatch/QueryDispatcher.java | 36 +++--
.../query/mailbox/channel/ChannelManagerTest.java | 111 +++++++--------
.../service/dispatch/QueryDispatcherTest.java | 83 +++++++++++
.../server/starter/helix/BaseServerStarter.java | 10 +-
...ava => ServerMailboxChannelRefreshHandler.java} | 107 +++++++++++---
...=> ServerMailboxChannelRefreshHandlerTest.java} | 155 +++++++++++++++++----
12 files changed, 516 insertions(+), 158 deletions(-)
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java
index 878d08f2db8..a9f4cf4a721 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java
@@ -585,8 +585,8 @@ public abstract class BaseBrokerStarter implements
ServiceStartable {
_threadAccountant, multiClusterRoutingContext, workerManager,
multiClusterWorkerManager,
_serverRoutingStatsManager);
MultiStageBrokerRequestHandler finalHandler =
multiStageBrokerRequestHandler;
- _routingManager.setServerReenableCallback(
- serverInstance ->
finalHandler.getQueryDispatcher().resetClientConnectionBackoff(serverInstance));
+ _routingManager.setServerEnabledCallback(
+ serverInstance ->
finalHandler.getQueryDispatcher().onServerEnabled(serverInstance));
// Lets an operator turn the proto segment list encoding on and off
through cluster config, without a restart.
_clusterConfigChangeHandler.registerClusterConfigChangeListener(
multiStageBrokerRequestHandler.getQueryDispatcher());
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java
index 6503606f600..85f699db3ce 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java
@@ -145,7 +145,7 @@ public abstract class BaseBrokerRoutingManager
private final long _newSegmentExpirationMs;
private final ExecutorService _executorService;
@Nullable
- private Consumer<ServerInstance> _serverReenableCallback;
+ private Consumer<ServerInstance> _serverEnabledCallback;
// Providers of extra per-table segment ZK metadata fetch listeners,
registered alongside the built-in segment
// pruners on every table's routing entry. Each provider is invoked once per
table (with the table name with type)
@@ -235,10 +235,11 @@ public abstract class BaseBrokerRoutingManager
return _partitionPruningPreparationThreshold;
}
- /// Sets a callback to be invoked when a server is re-enabled after being
excluded.
- /// This is useful for resetting gRPC channel state to avoid exponential
backoff delays.
- public void setServerReenableCallback(Consumer<ServerInstance> callback) {
- _serverReenableCallback = callback;
+ /// Sets a callback to be invoked when a server becomes routable: when it
starts, and when it is enabled again after
+ /// it was disabled (for example while it restarted) or excluded. This is
useful for resetting gRPC channel state to
+ /// avoid exponential backoff delays and connections to a stale address.
+ public void setServerEnabledCallback(Consumer<ServerInstance> callback) {
+ _serverEnabledCallback = callback;
}
/// Registers a provider of an extra [SegmentZkMetadataFetchListener] that
is attached to every table's routing
@@ -466,12 +467,12 @@ public abstract class BaseBrokerRoutingManager
// NOTE: Remove new enabled server from excluded servers because
the server is likely being restarted
if (_excludedServers.remove(instanceId)) {
LOGGER.info("Got excluded server: {} re-enabled, including it
into the routing", instanceId);
- // We clear any GRPC channel reconnection backoff in this
callback when a server is re-enabled. Otherwise,
- // MSE queries to this server may fail fast until the next
backoff retry succeeds.
- if (_serverReenableCallback != null) {
- LOGGER.info("Calling server re-enable callback for server:
{}", instanceId);
- _serverReenableCallback.accept(serverInstance);
- }
+ }
+ // The server likely restarted, possibly at a new address. The
callback refreshes the GRPC channel to it:
+ // it clears any reconnection backoff, without which MSE queries
to this server may fail fast until the
+ // next backoff retry succeeds, and it makes the channel resolve
the hostname again.
+ if (_serverEnabledCallback != null) {
+ _serverEnabledCallback.accept(serverInstance);
}
}
}
diff --git
a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/manager/BrokerRoutingManagerTest.java
b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/manager/BrokerRoutingManagerTest.java
index 4bbd9d88535..e56065a649e 100644
---
a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/manager/BrokerRoutingManagerTest.java
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/manager/BrokerRoutingManagerTest.java
@@ -124,7 +124,7 @@ public class BrokerRoutingManagerTest {
private PropertyKey _instanceConfigsKey;
@Mock
- private Consumer<ServerInstance> _serverReenableCallback;
+ private Consumer<ServerInstance> _serverEnabledCallback;
private BrokerRoutingManager _routingManager;
@@ -284,9 +284,9 @@ public class BrokerRoutingManagerTest {
}
@Test
- public void testServerReenableCallbackInvokedWhenExcludedServerReenabled() {
+ public void testServerEnabledCallbackInvokedWhenExcludedServerReenabled() {
// Set up callback
- _routingManager.setServerReenableCallback(_serverReenableCallback);
+ _routingManager.setServerEnabledCallback(_serverEnabledCallback);
// First, enable the server by processing instance config change
List<ZNRecord> instanceConfigs =
List.of(createEnabledServerZNRecord(SERVER_INSTANCE_ID));
@@ -312,29 +312,40 @@ public class BrokerRoutingManagerTest {
anyInt(), anyInt())).thenReturn(instanceConfigs);
_routingManager.processClusterChange(ChangeType.INSTANCE_CONFIG);
- // Verify callback was invoked with correct ServerInstance
+ // Verify callback was invoked with correct ServerInstance: when the
server was first enabled, and when it was
+ // enabled again
ArgumentCaptor<ServerInstance> captor =
ArgumentCaptor.forClass(ServerInstance.class);
- verify(_serverReenableCallback).accept(captor.capture());
+ verify(_serverEnabledCallback, times(2)).accept(captor.capture());
ServerInstance capturedInstance = captor.getValue();
assertEquals(capturedInstance.getHostname(), SERVER_HOST);
assertEquals(capturedInstance.getPort(), SERVER_PORT);
}
+ /// A server that restarts without having been excluded is disabled while it
shuts down and enabled again once it
+ /// is ready. It may be back at a new address, so the callback must run for
it too, to refresh the channel to it.
@Test
- public void testServerReenableCallbackNotInvokedForNewServer() {
- // Set up callback
- _routingManager.setServerReenableCallback(_serverReenableCallback);
+ public void testServerEnabledCallbackInvokedWhenRestartedServerReenabled() {
+ _routingManager.setServerEnabledCallback(_serverEnabledCallback);
- // Enable a new server (never excluded)
List<ZNRecord> instanceConfigs =
List.of(createEnabledServerZNRecord(SERVER_INSTANCE_ID));
when(_zkDataAccessor.getChildren(eq(INSTANCE_CONFIGS_PATH), any(),
eq(AccessOption.PERSISTENT),
anyInt(), anyInt())).thenReturn(instanceConfigs);
+ _routingManager.processClusterChange(ChangeType.INSTANCE_CONFIG);
+ verify(_serverEnabledCallback, times(1)).accept(any());
+ // A change that leaves the server enabled does not call it again
_routingManager.processClusterChange(ChangeType.INSTANCE_CONFIG);
+ verify(_serverEnabledCallback, times(1)).accept(any());
- // Verify callback was NOT invoked (server was never excluded)
- verify(_serverReenableCallback, never()).accept(any());
+ // Disabled while it restarts, then enabled again
+ when(_zkDataAccessor.getChildren(eq(INSTANCE_CONFIGS_PATH), any(),
eq(AccessOption.PERSISTENT),
+ anyInt(), anyInt())).thenReturn(List.of());
+ _routingManager.processClusterChange(ChangeType.INSTANCE_CONFIG);
+ when(_zkDataAccessor.getChildren(eq(INSTANCE_CONFIGS_PATH), any(),
eq(AccessOption.PERSISTENT),
+ anyInt(), anyInt())).thenReturn(instanceConfigs);
+ _routingManager.processClusterChange(ChangeType.INSTANCE_CONFIG);
+ verify(_serverEnabledCallback, times(2)).accept(any());
}
@Test
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/MailboxService.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/MailboxService.java
index 0a130b6692f..8309e84d42c 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/MailboxService.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/MailboxService.java
@@ -193,6 +193,7 @@ public class MailboxService {
public void shutdown() {
LOGGER.info("Shutting down GrpcMailboxServer");
_grpcMailboxServer.shutdown();
+ _channelManager.shutdown();
}
public String getHostname() {
@@ -232,11 +233,18 @@ public class MailboxService {
}
}
- /// Resets the GRPC connection backoff for the channel to the given server.
- /// @see ChannelManager#resetConnectBackoff(String, int)
- /// @return true if the channel was in TRANSIENT_FAILURE and backoff was
reset
- public boolean resetConnectBackoff(String hostname, int port) {
- return _channelManager.resetConnectBackoff(hostname, port);
+ /// Makes the channel to the given server resolve its hostname again before
the next send.
+ /// @see ChannelManager#refreshChannel(String, int)
+ /// @return false if there is no channel to the server, or it was just
refreshed
+ public boolean refreshChannel(String hostname, int port) {
+ return _channelManager.refreshChannel(hostname, port);
+ }
+
+ /// Shuts down the channel to an address that is no longer used.
+ /// @see ChannelManager#retireChannel(String, int)
+ /// @return false if there is no channel to the address
+ public boolean retireChannel(String hostname, int port) {
+ return _channelManager.retireChannel(hostname, port);
}
/// Current value of [BrokerGauge#MAILBOX_CLIENT_USED_DIRECT_MEMORY] /
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/channel/ChannelManager.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/channel/ChannelManager.java
index b9b06816935..c238422c102 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/channel/ChannelManager.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/channel/ChannelManager.java
@@ -29,6 +29,7 @@ import io.grpc.netty.shaded.io.netty.handler.ssl.SslContext;
import java.time.Duration;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
import javax.annotation.Nullable;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.pinot.query.grpc.GrpcKeepAliveConfig;
@@ -43,8 +44,13 @@ import org.slf4j.LoggerFactory;
public class ChannelManager {
private static final Logger LOGGER =
LoggerFactory.getLogger(ChannelManager.class);
+ /// A channel is refreshed at most once in this interval, see
[#refreshChannel].
+ private static final long MIN_REFRESH_INTERVAL_NS =
TimeUnit.SECONDS.toNanos(1);
+
/// Map from (hostname, port) to the ManagedChannel with all known channels
private final ConcurrentHashMap<Pair<String, Integer>, ManagedChannel>
_channelMap = new ConcurrentHashMap<>();
+ /// When each channel was last refreshed, from [System#nanoTime()], by
(hostname, port), see [#refreshChannel].
+ private final ConcurrentHashMap<Pair<String, Integer>, AtomicLong>
_lastRefreshTimeNs = new ConcurrentHashMap<>();
/// The idle timeout for the channel, which cannot be disabled in gRPC.
///
/// In general we want to prevent the channel from going idle, so that we
don't have to re-establish the connection
@@ -140,7 +146,7 @@ public class ChannelManager {
return;
}
ConnectivityState next = channel.getState(false);
- if (state == ConnectivityState.READY && next != ConnectivityState.READY)
{
+ if (state == ConnectivityState.READY && next != ConnectivityState.READY
&& !isRecentlyRefreshed(hostname, port)) {
LOGGER.warn("Mailbox channel to {}:{} left READY for {}; sends to that
peer will fail until it "
+ "reconnects. If keep-alive reported it, the peer stopped
answering about one keep-alive "
+ "interval ago.", hostname, port, next);
@@ -150,18 +156,63 @@ public class ChannelManager {
return channel;
}
- /// Resets the connection backoff for the channel to the given server if the
channel is in
- /// TRANSIENT_FAILURE state. Returns true if a reset was performed, false
otherwise.
+ /// Makes the channel to the given address resolve the hostname again and
open a new connection before its next
+ /// stream, dropping the connection it has, and any reconnect backoff.
Streams in flight finish on the old
+ /// connection. Returns false if there is no channel to the address, or if
it was refreshed less than a second ago.
+ ///
+ /// A channel resolves the hostname only when it starts, and again after a
connection fails or is lost. When the
+ /// server at that address goes away, gRPC resolves once, right then, and
keeps that answer until the next failure.
+ /// If DNS still had the old IP address at that moment (a terminating pod
stays in DNS for a while), the channel
+ /// later reconnects to the old address, and when another server took that
address over and listens on the same
+ /// port, the connection succeeds and is never questioned again. Call this
when the address can be stale, for example
+ /// when the server at the address has restarted.
///
- /// @return true if the channel was in TRANSIENT_FAILURE and backoff was
reset
- public boolean resetConnectBackoff(String hostname, int port) {
- ManagedChannel channel = _channelMap.get(Pair.of(hostname, port));
- if (channel != null && channel.getState(false) ==
ConnectivityState.TRANSIENT_FAILURE) {
- LOGGER.info("Resetting mailbox channel backoff for server: {}:{}",
hostname, port);
- channel.resetConnectBackoff();
- return true;
+ /// Moving the channel to IDLE, rather than replacing it, keeps the channel
object that concurrent senders may
+ /// already hold valid. The refreshes are limited to one a second, because
each refresh drops the connection that the
+ /// previous one opened.
+ public boolean refreshChannel(String hostname, int port) {
+ Pair<String, Integer> key = Pair.of(hostname, port);
+ ManagedChannel channel = _channelMap.get(key);
+ if (channel == null) {
+ return false;
+ }
+ long nowNs = System.nanoTime();
+ AtomicLong lastRefreshTimeNs =
+ _lastRefreshTimeNs.computeIfAbsent(key, k -> new AtomicLong(nowNs -
MIN_REFRESH_INTERVAL_NS));
+ long lastNs = lastRefreshTimeNs.get();
+ if (nowNs - lastNs < MIN_REFRESH_INTERVAL_NS ||
!lastRefreshTimeNs.compareAndSet(lastNs, nowNs)) {
+ return false;
+ }
+ channel.enterIdle();
+ return true;
+ }
+
+ private boolean isRecentlyRefreshed(String hostname, int port) {
+ AtomicLong lastRefreshTimeNs = _lastRefreshTimeNs.get(Pair.of(hostname,
port));
+ return lastRefreshTimeNs != null && System.nanoTime() -
lastRefreshTimeNs.get() < MIN_REFRESH_INTERVAL_NS;
+ }
+
+ /// Removes the channel to the given address and shuts it down, letting
streams in flight finish. Use it for an
+ /// address that is no longer used, such as the old mailbox port of a server
that restarted on another port, so
+ /// that channels do not pile up. Returns false if there is no channel to
the address.
+ public boolean retireChannel(String hostname, int port) {
+ Pair<String, Integer> key = Pair.of(hostname, port);
+ ManagedChannel channel = _channelMap.remove(key);
+ if (channel == null) {
+ return false;
+ }
+ _lastRefreshTimeNs.remove(key);
+ channel.shutdown();
+ return true;
+ }
+
+ /// Shuts down all channels, letting streams in flight finish.
+ public void shutdown() {
+ for (ManagedChannel channel : _channelMap.values()) {
+ channel.shutdown();
}
- return false;
+ _channelMap.clear();
+ _lastRefreshTimeNs.clear();
}
private NettyChannelBuilder decorate(NettyChannelBuilder builder) {
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/DispatchClient.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/DispatchClient.java
index 05c5f0905b0..2c1ce9e6c5a 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/DispatchClient.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/DispatchClient.java
@@ -26,6 +26,8 @@ import io.grpc.netty.shaded.io.netty.channel.ChannelOption;
import io.grpc.netty.shaded.io.netty.handler.ssl.SslContext;
import io.grpc.stub.StreamObserver;
import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import javax.annotation.Nullable;
@@ -50,8 +52,13 @@ class DispatchClient {
/// Using a static allocator allows for better memory pooling across all
DispatchClient instances.
private static final PooledByteBufAllocator BUF_ALLOCATOR = new
PooledByteBufAllocator(true);
+ /// The channel is refreshed at most once in this interval, see
[#refreshChannel].
+ private static final long MIN_REFRESH_INTERVAL_NS =
TimeUnit.SECONDS.toNanos(1);
+
private final ManagedChannel _channel;
private final PinotQueryWorkerGrpc.PinotQueryWorkerStub _dispatchStub;
+ /// When the channel was last refreshed, from [System#nanoTime()], see
[#refreshChannel].
+ private final AtomicLong _lastRefreshTimeNs = new
AtomicLong(System.nanoTime() - MIN_REFRESH_INTERVAL_NS);
public DispatchClient(String host, int port, @Nullable TlsConfig tlsConfig) {
this(host, port, tlsConfig, null, GrpcKeepAliveConfig.DISABLED);
@@ -99,6 +106,20 @@ class DispatchClient {
_dispatchStub.withDeadline(deadline).submit(request, new
LastValueDispatchObserver<>(virtualServer, callback));
}
+ /// Makes the channel resolve the server's hostname again and open a new
connection, without reconnect backoff, for
+ /// the next request. Requests in flight finish on the old connection. Does
nothing, and returns false, if the
+ /// channel was refreshed less than a second ago: each refresh drops the
connection that the previous one opened.
+ public boolean refreshChannel() {
+ long nowNs = System.nanoTime();
+ long lastRefreshTimeNs = _lastRefreshTimeNs.get();
+ if (nowNs - lastRefreshTimeNs < MIN_REFRESH_INTERVAL_NS
+ || !_lastRefreshTimeNs.compareAndSet(lastRefreshTimeNs, nowNs)) {
+ return false;
+ }
+ _channel.enterIdle();
+ return true;
+ }
+
/// Opens a `SubmitWithStream` bidi RPC for one server, sends the initial
`submit`, and registers the
/// resulting [StreamingDispatchObserver] with `session` for cancel fan-out
and `OpChainComplete`
/// accumulation.
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 b3a1e605da6..d2897d6ef0f 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,6 +124,9 @@ 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.
+ private final Map<String, String> _dispatchAddressByInstanceId = new
ConcurrentHashMap<>();
@Nullable
private final TlsConfig _tlsConfig;
@Nullable
@@ -896,16 +899,31 @@ public class QueryDispatcher implements
PinotClusterConfigChangeListener {
_dispatchMaxInboundMessageSizeBytes));
}
- /// Reset the connection backoff for a server. When the GRPC channel enters
a TRANSIENT_FAILURE state from
- /// connection failures, it will fast fail requests and reconnect with
exponential backoff. This method
- /// resets the backoff so servers that have recovered can be reconnected to
immediately.
- public void resetClientConnectionBackoff(ServerInstance serverInstance) {
- String hostname = serverInstance.getHostname();
- int port = serverInstance.getQueryServicePort();
- DispatchClient dispatchClient =
_dispatchClientMap.get(toHostnamePortKey(hostname, port));
+ /// 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:
+ ///
+ /// - the channel to its address resolves the hostname again before the next
request, and connects without
+ /// reconnect backoff. A channel resolves only when a connection fails or
is lost, and the last time was likely
+ /// when the server went away, while DNS could still have the old IP
address. Reconnecting to that address could
+ /// reach another server that took it over.
+ /// - the channel to its previous address, if the server moved to another
port, is shut down, since nothing uses it
+ /// again.
+ public void onServerEnabled(ServerInstance serverInstance) {
+ String hostnamePortKey = toHostnamePortKey(serverInstance.getHostname(),
serverInstance.getQueryServicePort());
+ String previousKey =
_dispatchAddressByInstanceId.put(serverInstance.getInstanceId(),
hostnamePortKey);
+ if (previousKey != null && !previousKey.equals(hostnamePortKey)
+ && !_dispatchAddressByInstanceId.containsValue(previousKey)) {
+ DispatchClient previousClient = _dispatchClientMap.remove(previousKey);
+ if (previousClient != null) {
+ LOGGER.info("Server: {} moved from {} to {}, shutting down the
dispatch channel to the old address",
+ serverInstance.getInstanceId(), previousKey, hostnamePortKey);
+ previousClient.getChannel().shutdown();
+ }
+ }
+ DispatchClient dispatchClient = _dispatchClientMap.get(hostnamePortKey);
if (dispatchClient != null) {
- LOGGER.info("Resetting connection backoff for server: {}",
serverInstance.getInstanceId());
- dispatchClient.getChannel().resetConnectBackoff();
+ LOGGER.info("Server: {} is enabled, refreshing the dispatch channel to
it", serverInstance.getInstanceId());
+ dispatchClient.refreshChannel();
}
}
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/channel/ChannelManagerTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/channel/ChannelManagerTest.java
index 0eab4e2bde2..1f0bb9dcd6e 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/channel/ChannelManagerTest.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/channel/ChannelManagerTest.java
@@ -20,76 +20,20 @@ package org.apache.pinot.query.mailbox.channel;
import io.grpc.ConnectivityState;
import io.grpc.ManagedChannel;
-import java.lang.reflect.Field;
import java.time.Duration;
-import java.util.concurrent.ConcurrentHashMap;
-import org.apache.commons.lang3.tuple.Pair;
import org.apache.pinot.query.grpc.GrpcKeepAliveConfig;
import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.util.TestUtils;
import org.testng.annotations.Test;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotSame;
import static org.testng.Assert.assertSame;
import static org.testng.Assert.assertTrue;
public class ChannelManagerTest {
- @Test
- public void testResetConnectBackoffNoOpForUnknownChannel() {
- ChannelManager channelManager = new ChannelManager(null, 4_000_000,
Duration.ofDays(365),
-
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_HIGH_WATER_MARK_BYTES,
-
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_LOW_WATER_MARK_BYTES,
- GrpcKeepAliveConfig.DISABLED);
- // Should return false and not throw when no channel exists for the given
host/port
- assertFalse(channelManager.resetConnectBackoff("unknown-host", 12345));
- }
-
- @Test
- public void testResetConnectBackoffNoOpWhenNotInTransientFailure() {
- ChannelManager channelManager = new ChannelManager(null, 4_000_000,
Duration.ofDays(365),
-
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_HIGH_WATER_MARK_BYTES,
-
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_LOW_WATER_MARK_BYTES,
- GrpcKeepAliveConfig.DISABLED);
- // Create a channel by calling getChannel
- ManagedChannel channel = channelManager.getChannel("localhost", 12345);
- try {
- // Reset backoff should return false since channel is not in
TRANSIENT_FAILURE
- assertFalse(channelManager.resetConnectBackoff("localhost", 12345));
- // Channel should still be the same cached instance
- ManagedChannel sameChannel = channelManager.getChannel("localhost",
12345);
- assertSame(sameChannel, channel);
- } finally {
- channel.shutdownNow();
- }
- }
-
- @Test
- @SuppressWarnings("unchecked")
- public void testResetConnectBackoffResetsWhenInTransientFailure()
- throws Exception {
- ChannelManager channelManager = new ChannelManager(null, 4_000_000,
Duration.ofDays(365),
-
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_HIGH_WATER_MARK_BYTES,
-
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_LOW_WATER_MARK_BYTES,
- GrpcKeepAliveConfig.DISABLED);
-
- ManagedChannel mockChannel = mock(ManagedChannel.class);
-
when(mockChannel.getState(false)).thenReturn(ConnectivityState.TRANSIENT_FAILURE);
-
- // Inject the mock channel into _channelMap via reflection
- Field channelMapField =
ChannelManager.class.getDeclaredField("_channelMap");
- channelMapField.setAccessible(true);
- ConcurrentHashMap<Pair<String, Integer>, ManagedChannel> channelMap =
- (ConcurrentHashMap<Pair<String, Integer>, ManagedChannel>)
channelMapField.get(channelManager);
- channelMap.put(Pair.of("failing-host", 9999), mockChannel);
-
- assertTrue(channelManager.resetConnectBackoff("failing-host", 9999));
- verify(mockChannel).resetConnectBackoff();
- }
-
/// Pins the fail-fast gate added in commit 1d29438dc0 ("Fail fast on
invalid gRPC mailbox transport
/// configuration"): a non-positive `writeBufferLowWaterMarkBytes` must
throw at startup rather than
/// surfacing later as a Netty `WriteBufferWaterMark` constructor failure on
the first send.
@@ -135,6 +79,57 @@ public class ChannelManagerTest {
}
}
+ /// Refreshing keeps the channel object, which senders may hold, and only
drops what it resolved and connected to.
+ @Test
+ public void testRefreshChannel() {
+ ChannelManager channelManager = newChannelManager();
+ assertFalse(channelManager.refreshChannel("localhost", 12348));
+ ManagedChannel channel = channelManager.getChannel("localhost", 12348);
+ try {
+ // Nothing listens there: the connect attempt fails, but takes the
channel out of IDLE
+ channel.getState(true);
+ assertTrue(channelManager.refreshChannel("localhost", 12348));
+ // A stale connection fails many streams at once: only the first one
drops the connection
+ assertFalse(channelManager.refreshChannel("localhost", 12348));
+ TestUtils.waitForCondition(aVoid -> channel.getState(false) ==
ConnectivityState.IDLE, 10_000L,
+ "The channel did not go back to IDLE");
+ assertFalse(channel.isShutdown());
+ assertSame(channelManager.getChannel("localhost", 12348), channel);
+
+ // Once a second has passed, the channel can be refreshed again
+ channel.getState(true);
+ TestUtils.waitForCondition(aVoid ->
channelManager.refreshChannel("localhost", 12348), 10_000L,
+ "The channel was not refreshed again");
+ TestUtils.waitForCondition(aVoid -> channel.getState(false) ==
ConnectivityState.IDLE, 10_000L,
+ "The channel did not go back to IDLE");
+ } finally {
+ channelManager.shutdown();
+ }
+ }
+
+ @Test
+ public void testRetireChannelAndShutdown() {
+ ChannelManager channelManager = newChannelManager();
+ assertFalse(channelManager.retireChannel("localhost", 12349));
+ ManagedChannel retired = channelManager.getChannel("localhost", 12349);
+ assertTrue(channelManager.retireChannel("localhost", 12349));
+ assertTrue(retired.isShutdown());
+ ManagedChannel replacement = channelManager.getChannel("localhost", 12349);
+ assertNotSame(replacement, retired);
+
+ ManagedChannel other = channelManager.getChannel("localhost", 12350);
+ channelManager.shutdown();
+ assertTrue(replacement.isShutdown());
+ assertTrue(other.isShutdown());
+ }
+
+ private static ChannelManager newChannelManager() {
+ return new ChannelManager(null, 4_000_000, Duration.ofDays(365),
+
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_HIGH_WATER_MARK_BYTES,
+
CommonConstants.MultiStageQueryRunner.DEFAULT_GRPC_WRITE_BUFFER_LOW_WATER_MARK_BYTES,
+ GrpcKeepAliveConfig.DISABLED);
+ }
+
/// A disabled policy must leave the builder alone rather than passing `-1`
to gRPC, which throws.
@Test
public void testDisabledKeepAliveStillBuildsChannel() {
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 d0b4caae606..ee225479c1d 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
@@ -18,6 +18,8 @@
*/
package org.apache.pinot.query.service.dispatch;
+import io.grpc.ConnectivityState;
+import io.grpc.ManagedChannel;
import io.grpc.stub.StreamObserver;
import java.time.Duration;
import java.util.ArrayList;
@@ -29,9 +31,11 @@ import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;
+import org.apache.helix.model.InstanceConfig;
import org.apache.pinot.common.failuredetector.FailureDetector;
import org.apache.pinot.common.metrics.BrokerMetrics;
import org.apache.pinot.common.proto.Worker;
+import org.apache.pinot.core.transport.ServerInstance;
import
org.apache.pinot.core.transport.server.routing.stats.ServerRoutingStatsManager;
import org.apache.pinot.query.QueryEnvironment;
import org.apache.pinot.query.QueryEnvironmentTestBase;
@@ -137,6 +141,85 @@ public class QueryDispatcherTest extends QueryTestSet {
}
}
+ /// When a server becomes routable, the dispatch channel to it resolves the
hostname again, and the channel to an
+ /// address it left is shut down.
+ @Test
+ public void testServerEnabledRefreshesAndRetiresChannels() {
+ QueryDispatcher dispatcher =
+ new QueryDispatcher(Mockito.mock(MailboxService.class),
Mockito.mock(FailureDetector.class), null, false,
+ Duration.ofSeconds(1));
+ try {
+ // Nothing listens there
+ int port = QueryTestUtils.getAvailablePort();
+ QueryServerInstance server = new
QueryServerInstance("Server_localhost_8098", "localhost", port, 8101);
+ DispatchClient client = dispatcher.getOrCreateDispatchClient(server);
+ ManagedChannel channel = client.getChannel();
+ // Connecting fails, but takes the channel out of IDLE
+ channel.getState(true);
+ dispatcher.onServerEnabled(serverInstance("Server_localhost_8098",
port));
+ Assert.assertSame(dispatcher.getOrCreateDispatchClient(server), client);
+ TestUtils.waitForCondition(aVoid -> channel.getState(false) ==
ConnectivityState.IDLE, 10_000L,
+ "The channel did not go back to IDLE");
+ Assert.assertFalse(channel.isShutdown());
+
+ // The server restarted on another port
+ dispatcher.onServerEnabled(serverInstance("Server_localhost_8098", port
+ 1));
+ Assert.assertTrue(channel.isShutdown());
+ Assert.assertNotSame(dispatcher.getOrCreateDispatchClient(server),
client);
+ } finally {
+ dispatcher.shutdown();
+ }
+ }
+
+ /// The dispatch channel to an address stays when the server that had the
address moves away, if another server has
+ /// taken the address over.
+ @Test
+ public void testServerEnabledKeepsChannelToAddressTakenOver() {
+ QueryDispatcher dispatcher =
+ new QueryDispatcher(Mockito.mock(MailboxService.class),
Mockito.mock(FailureDetector.class), null, false,
+ Duration.ofSeconds(1));
+ try {
+ int port = QueryTestUtils.getAvailablePort();
+ dispatcher.onServerEnabled(serverInstance("Server_localhost_8098",
port));
+ DispatchClient client = dispatcher.getOrCreateDispatchClient(
+ new QueryServerInstance("Server_localhost_8098", "localhost", port,
8101));
+
+ // Another server takes over the address, then the server that had it
comes back on another port
+ dispatcher.onServerEnabled(serverInstance("Server_localhost_8099",
port));
+ dispatcher.onServerEnabled(serverInstance("Server_localhost_8098", port
+ 1));
+
+ Assert.assertFalse(client.getChannel().isShutdown());
+ Assert.assertSame(dispatcher.getOrCreateDispatchClient(
+ new QueryServerInstance("Server_localhost_8099", "localhost", port,
8101)), client);
+ } finally {
+ dispatcher.shutdown();
+ }
+ }
+
+ /// Each refresh drops the connection that the previous one opened, so a
channel is refreshed at most once a second.
+ @Test
+ public void testDispatchChannelRefreshIsDebounced() {
+ DispatchClient client = new DispatchClient("localhost",
QueryTestUtils.getAvailablePort(), null);
+ try {
+ Assert.assertTrue(client.refreshChannel());
+ Assert.assertFalse(client.refreshChannel());
+ TestUtils.waitForCondition(aVoid -> client.refreshChannel(), 10_000L,
"The channel was not refreshed again");
+ } finally {
+ client.getChannel().shutdownNow();
+ }
+ }
+
+ private static ServerInstance serverInstance(String instanceId, int
queryServerPort) {
+ InstanceConfig instanceConfig = new InstanceConfig(instanceId);
+ instanceConfig.setHostName("localhost");
+ instanceConfig.setPort(instanceId.substring(instanceId.lastIndexOf('_') +
1));
+ instanceConfig.getRecord()
+
.setIntField(CommonConstants.Helix.Instance.MULTI_STAGE_QUERY_ENGINE_SERVICE_PORT_KEY,
queryServerPort);
+ instanceConfig.getRecord()
+
.setIntField(CommonConstants.Helix.Instance.MULTI_STAGE_QUERY_ENGINE_MAILBOX_PORT_KEY,
8101);
+ return new ServerInstance(instanceConfig);
+ }
+
@AfterClass
public void tearDown() {
_queryDispatcher.shutdown();
diff --git
a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java
b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java
index 7d90df3ac06..248d1d96210 100644
---
a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java
+++
b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java
@@ -853,19 +853,19 @@ public abstract class BaseServerStarter implements
ServiceStartable {
}
}
- // Register handler to reset GRPC mailbox channel backoff when servers
come online
+ // Register handler to refresh the GRPC mailbox channels to servers when
they come online
WorkerQueryServer workerQueryServer =
_serverInstance.getWorkerQueryServer();
if (workerQueryServer != null) {
MailboxService mailboxService =
workerQueryServer.getQueryRunner().getMailboxService();
- LOGGER.info("Registering ServerGrpcChannelBackoffResetHandler");
+ LOGGER.info("Registering ServerMailboxChannelRefreshHandler");
try {
_helixManager.addInstanceConfigChangeListener(
- new ServerGrpcChannelBackoffResetHandler(_helixAdmin,
_helixClusterName, _instanceId, mailboxService));
+ new ServerMailboxChannelRefreshHandler(_helixAdmin,
_helixClusterName, _instanceId, mailboxService));
} catch (Exception e) {
- LOGGER.error("Failed to register
ServerGrpcChannelBackoffResetHandler", e);
+ LOGGER.error("Failed to register ServerMailboxChannelRefreshHandler",
e);
}
} else {
- LOGGER.info("Multi-stage query engine not enabled, skipping
ServerGrpcChannelBackoffResetHandler registration");
+ LOGGER.info("Multi-stage query engine not enabled, skipping
ServerMailboxChannelRefreshHandler registration");
}
// Start restlet server for admin API endpoint
diff --git
a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/ServerGrpcChannelBackoffResetHandler.java
b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/ServerMailboxChannelRefreshHandler.java
similarity index 53%
rename from
pinot-server/src/main/java/org/apache/pinot/server/starter/helix/ServerGrpcChannelBackoffResetHandler.java
rename to
pinot-server/src/main/java/org/apache/pinot/server/starter/helix/ServerMailboxChannelRefreshHandler.java
index 28f5f173f92..331d88e12d9 100644
---
a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/ServerGrpcChannelBackoffResetHandler.java
+++
b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/ServerMailboxChannelRefreshHandler.java
@@ -18,10 +18,13 @@
*/
package org.apache.pinot.server.starter.helix;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import javax.annotation.Nullable;
+import org.apache.commons.lang3.tuple.Pair;
import org.apache.helix.HelixAdmin;
import org.apache.helix.NotificationContext;
import org.apache.helix.api.listeners.BatchMode;
@@ -36,8 +39,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/// Listens for instance config changes and resets GRPC mailbox channel
backoff when a server
-/// completes startup (IS_SHUTDOWN_IN_PROGRESS transitions from true to false).
+/// Listens for instance config changes and refreshes the GRPC mailbox channel
to a server when it completes startup
+/// (IS_SHUTDOWN_IN_PROGRESS transitions from true to false).
///
/// When a server restarts (gracefully or after a crash), the startup sequence
is:
///
@@ -46,13 +49,19 @@ import org.slf4j.LoggerFactory;
/// 3. Query/gRPC server starts listening
/// 4. IS_SHUTDOWN_IN_PROGRESS set to false (readiness signal)
///
-/// This handler detects the transition in step 4 and resets the GRPC channel
backoff for
-/// that specific server, but only if the channel is currently in
TRANSIENT_FAILURE state.
-/// Channels in other states (READY, IDLE, CONNECTING) are left untouched.
+/// This handler detects the transition in step 4 and refreshes the channel to
that server: the channel resolves the
+/// server's hostname again before the next send, and connects without waiting
for its reconnect backoff. Resolving
+/// again matters because the restarted server may have a new IP address. The
channel last resolved the hostname when
+/// the old connection was lost, while DNS could still return the old address,
and it would reconnect there. When
+/// another server took over that address and listens on the same port, the
connection succeeds and the data goes to
+/// the wrong server.
+///
+/// It also shuts down the channel to the old mailbox address of a server that
moved to another address (for example a
+/// new port after a restart) or left the cluster, since nothing will use it
again.
@BatchMode(enabled = false)
@PreFetch(enabled = false)
-public class ServerGrpcChannelBackoffResetHandler implements
InstanceConfigChangeListener {
- private static final Logger LOGGER =
LoggerFactory.getLogger(ServerGrpcChannelBackoffResetHandler.class);
+public class ServerMailboxChannelRefreshHandler implements
InstanceConfigChangeListener {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(ServerMailboxChannelRefreshHandler.class);
private final HelixAdmin _helixAdmin;
private final String _clusterName;
@@ -60,10 +69,12 @@ public class ServerGrpcChannelBackoffResetHandler
implements InstanceConfigChang
private final MailboxService _mailboxService;
// Tracks servers that have IS_SHUTDOWN_IN_PROGRESS=true. When a server
transitions
- // from being in this set to having IS_SHUTDOWN_IN_PROGRESS=false, we reset
its channel backoff.
+ // from being in this set to having IS_SHUTDOWN_IN_PROGRESS=false, we
refresh its channel.
private final Set<String> _shuttingDownServers = new HashSet<>();
+ // Last known mailbox address of each server, to retire the channel of an
address a server has left
+ private final Map<String, Pair<String, Integer>> _mailboxAddresses = new
HashMap<>();
- public ServerGrpcChannelBackoffResetHandler(HelixAdmin helixAdmin, String
clusterName, String selfInstanceId,
+ public ServerMailboxChannelRefreshHandler(HelixAdmin helixAdmin, String
clusterName, String selfInstanceId,
MailboxService mailboxService) {
_helixAdmin = helixAdmin;
_clusterName = clusterName;
@@ -93,20 +104,43 @@ public class ServerGrpcChannelBackoffResetHandler
implements InstanceConfigChang
}
/// Full cluster scan: used on INIT (first callback) and when instances are
added/removed.
+ ///
+ /// It records the new address of every server before it retires any old
one. Otherwise, when one server left an
+ /// address and a server read later in the scan took it over, the scan would
shut down the channel the second server
+ /// uses.
private void handleFullScan() {
Set<String> currentShuttingDown = new HashSet<>();
+ Set<String> currentServers = new HashSet<>();
+ Map<String, Pair<String, Integer>> previousAddresses = new HashMap<>();
for (String instance : _helixAdmin.getInstancesInCluster(_clusterName)) {
if (!InstanceTypeUtils.isServer(instance) ||
instance.equals(_selfInstanceId)) {
continue;
}
+ // Before reading the config: a server whose config cannot be read now
has not left the cluster
+ currentServers.add(instance);
InstanceConfig config = getInstanceConfig(instance);
if (config == null) {
continue;
}
+ Pair<String, Integer> previousAddress = recordMailboxAddress(instance,
config);
+ if (previousAddress != null) {
+ previousAddresses.put(instance, previousAddress);
+ }
if (isShutdownInProgress(config)) {
currentShuttingDown.add(instance);
} else if (_shuttingDownServers.contains(instance)) {
- resetBackoffForServer(instance, config);
+ refreshChannelToServer(instance, config);
+ }
+ }
+ // Servers that left the cluster
+ for (String instance : new HashSet<>(_mailboxAddresses.keySet())) {
+ if (!currentServers.contains(instance)) {
+ previousAddresses.put(instance, _mailboxAddresses.remove(instance));
+ }
+ }
+ for (Map.Entry<String, Pair<String, Integer>> entry :
previousAddresses.entrySet()) {
+ if (!entry.getValue().equals(_mailboxAddresses.get(entry.getKey()))) {
+ retireChannel(entry.getKey(), entry.getValue());
}
}
_shuttingDownServers.clear();
@@ -123,10 +157,11 @@ public class ServerGrpcChannelBackoffResetHandler
implements InstanceConfigChang
if (config == null) {
return;
}
+ updateMailboxAddress(instance, config);
if (isShutdownInProgress(config)) {
_shuttingDownServers.add(instance);
} else if (_shuttingDownServers.remove(instance)) {
- resetBackoffForServer(instance, config);
+ refreshChannelToServer(instance, config);
}
}
@@ -145,24 +180,56 @@ public class ServerGrpcChannelBackoffResetHandler
implements InstanceConfigChang
config.getRecord().getStringField(CommonConstants.Helix.IS_SHUTDOWN_IN_PROGRESS,
"false"));
}
+ /// Records the server's current mailbox address, and retires the channel to
its previous one if it moved.
+ private void updateMailboxAddress(String instanceId, InstanceConfig config) {
+ Pair<String, Integer> previousAddress = recordMailboxAddress(instanceId,
config);
+ if (previousAddress != null &&
!previousAddress.equals(_mailboxAddresses.get(instanceId))) {
+ retireChannel(instanceId, previousAddress);
+ }
+ }
+
+ /// Records the server's current mailbox address, and returns its previous
one.
+ @Nullable
+ private Pair<String, Integer> recordMailboxAddress(String instanceId,
InstanceConfig config) {
+ Pair<String, Integer> address = getMailboxAddress(instanceId, config);
+ return address != null ? _mailboxAddresses.put(instanceId, address) :
_mailboxAddresses.remove(instanceId);
+ }
- private void resetBackoffForServer(String instanceId, InstanceConfig config)
{
+ private void retireChannel(String instanceId, Pair<String, Integer> address)
{
+ // Another server may have taken over the address
+ if (_mailboxAddresses.containsValue(address)) {
+ return;
+ }
+ if (_mailboxService.retireChannel(address.getLeft(), address.getRight())) {
+ LOGGER.info("Server {} left mailbox address {}:{}, shut down the channel
to it", instanceId, address.getLeft(),
+ address.getRight());
+ }
+ }
+
+ @Nullable
+ private static Pair<String, Integer> getMailboxAddress(String instanceId,
InstanceConfig config) {
int mailboxPort = config.getRecord().getIntField(
CommonConstants.Helix.Instance.MULTI_STAGE_QUERY_ENGINE_MAILBOX_PORT_KEY, -1);
if (mailboxPort <= 0) {
- LOGGER.debug("Server {} has no mailbox port configured, skipping backoff
reset", instanceId);
- return;
+ return null;
}
- String hostname;
try {
- hostname = ServerInstance.extractHostnameFromConfig(config);
+ return Pair.of(ServerInstance.extractHostnameFromConfig(config),
mailboxPort);
} catch (Exception e) {
- LOGGER.warn("Could not extract hostname for server: {}, skipping backoff
reset", instanceId, e);
+ LOGGER.warn("Could not extract hostname for server: {}", instanceId, e);
+ return null;
+ }
+ }
+
+ private void refreshChannelToServer(String instanceId, InstanceConfig
config) {
+ Pair<String, Integer> address = getMailboxAddress(instanceId, config);
+ if (address == null) {
+ LOGGER.debug("Server {} has no mailbox address, skipping channel
refresh", instanceId);
return;
}
- if (_mailboxService.resetConnectBackoff(hostname, mailboxPort)) {
- LOGGER.info("Server {} completed startup, reset mailbox channel backoff
to {}:{}", instanceId, hostname,
- mailboxPort);
+ if (_mailboxService.refreshChannel(address.getLeft(), address.getRight()))
{
+ LOGGER.info("Server {} completed startup, refreshed mailbox channel to
{}:{}", instanceId, address.getLeft(),
+ address.getRight());
}
}
}
diff --git
a/pinot-server/src/test/java/org/apache/pinot/server/starter/helix/ServerGrpcChannelBackoffResetHandlerTest.java
b/pinot-server/src/test/java/org/apache/pinot/server/starter/helix/ServerMailboxChannelRefreshHandlerTest.java
similarity index 55%
rename from
pinot-server/src/test/java/org/apache/pinot/server/starter/helix/ServerGrpcChannelBackoffResetHandlerTest.java
rename to
pinot-server/src/test/java/org/apache/pinot/server/starter/helix/ServerMailboxChannelRefreshHandlerTest.java
index 36ae9b6b0c8..3fddb06e72b 100644
---
a/pinot-server/src/test/java/org/apache/pinot/server/starter/helix/ServerGrpcChannelBackoffResetHandlerTest.java
+++
b/pinot-server/src/test/java/org/apache/pinot/server/starter/helix/ServerMailboxChannelRefreshHandlerTest.java
@@ -40,7 +40,7 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
-public class ServerGrpcChannelBackoffResetHandlerTest {
+public class ServerMailboxChannelRefreshHandlerTest {
private static final String CLUSTER_NAME = "testCluster";
private static final String SELF_INSTANCE_ID = "Server_selfhost_8000";
private static final String OTHER_SERVER_ID = "Server_otherhost_8000";
@@ -58,12 +58,12 @@ public class ServerGrpcChannelBackoffResetHandlerTest {
@Mock
private MailboxService _mailboxService;
- private ServerGrpcChannelBackoffResetHandler _handler;
+ private ServerMailboxChannelRefreshHandler _handler;
@BeforeMethod
public void setUp() {
_mocks = MockitoAnnotations.openMocks(this);
- _handler = new ServerGrpcChannelBackoffResetHandler(_helixAdmin,
CLUSTER_NAME, SELF_INSTANCE_ID, _mailboxService);
+ _handler = new ServerMailboxChannelRefreshHandler(_helixAdmin,
CLUSTER_NAME, SELF_INSTANCE_ID, _mailboxService);
}
@AfterMethod
@@ -73,22 +73,22 @@ public class ServerGrpcChannelBackoffResetHandlerTest {
}
@Test
- public void testResetsBackoffWhenServerCompletesStartup() {
+ public void testRefreshesChannelWhenServerCompletesStartup() {
// First callback: server has IS_SHUTDOWN_IN_PROGRESS=true (starting up)
when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID))
.thenReturn(createServerConfig(OTHER_SERVER_ID, OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT, true));
_handler.onInstanceConfigChange(List.of(),
createCallbackContextForInstance(OTHER_SERVER_ID));
- verify(_mailboxService, never()).resetConnectBackoff(any(), anyInt());
+ verify(_mailboxService, never()).refreshChannel(any(), anyInt());
// Second callback: server has IS_SHUTDOWN_IN_PROGRESS=false (startup
complete)
when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID))
.thenReturn(createServerConfig(OTHER_SERVER_ID, OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT, false));
_handler.onInstanceConfigChange(List.of(),
createCallbackContextForInstance(OTHER_SERVER_ID));
- verify(_mailboxService).resetConnectBackoff(OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT);
+ verify(_mailboxService).refreshChannel(OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT);
}
@Test
- public void testDoesNotResetOnFirstCallbackForHealthyServer() {
+ public void testDoesNotRefreshOnFirstCallbackForHealthyServer() {
// First callback: server is healthy (IS_SHUTDOWN_IN_PROGRESS=false)
// This should NOT trigger a reset because we haven't seen it in
shutting-down state
when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID))
@@ -96,11 +96,11 @@ public class ServerGrpcChannelBackoffResetHandlerTest {
_handler.onInstanceConfigChange(List.of(),
createCallbackContextForInstance(OTHER_SERVER_ID));
- verify(_mailboxService, never()).resetConnectBackoff(any(), anyInt());
+ verify(_mailboxService, never()).refreshChannel(any(), anyInt());
}
@Test
- public void testDoesNotResetForServerStillShuttingDown() {
+ public void testDoesNotRefreshForServerStillShuttingDown() {
// First callback: server is shutting down
when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID))
.thenReturn(createServerConfig(OTHER_SERVER_ID, OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT, true));
@@ -109,28 +109,28 @@ public class ServerGrpcChannelBackoffResetHandlerTest {
// Second callback: server still shutting down
_handler.onInstanceConfigChange(List.of(),
createCallbackContextForInstance(OTHER_SERVER_ID));
- verify(_mailboxService, never()).resetConnectBackoff(any(), anyInt());
+ verify(_mailboxService, never()).refreshChannel(any(), anyInt());
}
@Test
- public void testDoesNotResetForSelfInstance() {
+ public void testDoesNotRefreshForSelfInstance() {
_handler.onInstanceConfigChange(List.of(),
createCallbackContextForInstance(SELF_INSTANCE_ID));
verify(_helixAdmin, never()).getInstanceConfig(any(), any());
- verify(_mailboxService, never()).resetConnectBackoff(any(), anyInt());
+ verify(_mailboxService, never()).refreshChannel(any(), anyInt());
}
@Test
- public void testDoesNotResetForBrokerInstance() {
+ public void testDoesNotRefreshForBrokerInstance() {
String brokerId = "Broker_brokerhost_8000";
_handler.onInstanceConfigChange(List.of(),
createCallbackContextForInstance(brokerId));
verify(_helixAdmin, never()).getInstanceConfig(any(), any());
- verify(_mailboxService, never()).resetConnectBackoff(any(), anyInt());
+ verify(_mailboxService, never()).refreshChannel(any(), anyInt());
}
@Test
- public void testDoesNotResetForServerWithoutMailboxPort() {
+ public void testDoesNotRefreshForServerWithoutMailboxPort() {
// First callback: server shutting down (no mailbox port)
when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID))
.thenReturn(createServerConfig(OTHER_SERVER_ID, OTHER_SERVER_HOST, -1,
true));
@@ -141,32 +141,32 @@ public class ServerGrpcChannelBackoffResetHandlerTest {
.thenReturn(createServerConfig(OTHER_SERVER_ID, OTHER_SERVER_HOST, -1,
false));
_handler.onInstanceConfigChange(List.of(),
createCallbackContextForInstance(OTHER_SERVER_ID));
- verify(_mailboxService, never()).resetConnectBackoff(any(), anyInt());
+ verify(_mailboxService, never()).refreshChannel(any(), anyInt());
}
@Test
- public void testResetsBackoffOnCrashRecovery() {
+ public void testRefreshesChannelOnCrashRecovery() {
// First callback: server is healthy (IS_SHUTDOWN_IN_PROGRESS=false)
when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID))
.thenReturn(createServerConfig(OTHER_SERVER_ID, OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT, false));
_handler.onInstanceConfigChange(List.of(),
createCallbackContextForInstance(OTHER_SERVER_ID));
- verify(_mailboxService, never()).resetConnectBackoff(any(), anyInt());
+ verify(_mailboxService, never()).refreshChannel(any(), anyInt());
// Second callback: server crashed and is restarting
(IS_SHUTDOWN_IN_PROGRESS=true)
when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID))
.thenReturn(createServerConfig(OTHER_SERVER_ID, OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT, true));
_handler.onInstanceConfigChange(List.of(),
createCallbackContextForInstance(OTHER_SERVER_ID));
- verify(_mailboxService, never()).resetConnectBackoff(any(), anyInt());
+ verify(_mailboxService, never()).refreshChannel(any(), anyInt());
// Third callback: server completed startup (IS_SHUTDOWN_IN_PROGRESS=false)
when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID))
.thenReturn(createServerConfig(OTHER_SERVER_ID, OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT, false));
_handler.onInstanceConfigChange(List.of(),
createCallbackContextForInstance(OTHER_SERVER_ID));
- verify(_mailboxService).resetConnectBackoff(OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT);
+ verify(_mailboxService).refreshChannel(OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT);
}
@Test
- public void testResetsBackoffOnGracefulRestart() {
+ public void testRefreshesChannelOnGracefulRestart() {
// First callback: server is healthy
when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID))
.thenReturn(createServerConfig(OTHER_SERVER_ID, OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT, false));
@@ -179,13 +179,13 @@ public class ServerGrpcChannelBackoffResetHandlerTest {
// Third callback: still shutting down / restarting
(IS_SHUTDOWN_IN_PROGRESS=true)
_handler.onInstanceConfigChange(List.of(),
createCallbackContextForInstance(OTHER_SERVER_ID));
- verify(_mailboxService, never()).resetConnectBackoff(any(), anyInt());
+ verify(_mailboxService, never()).refreshChannel(any(), anyInt());
// Fourth callback: startup complete (IS_SHUTDOWN_IN_PROGRESS=false)
when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID))
.thenReturn(createServerConfig(OTHER_SERVER_ID, OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT, false));
_handler.onInstanceConfigChange(List.of(),
createCallbackContextForInstance(OTHER_SERVER_ID));
- verify(_mailboxService).resetConnectBackoff(OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT);
+ verify(_mailboxService).refreshChannel(OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT);
}
@Test
@@ -197,13 +197,109 @@ public class ServerGrpcChannelBackoffResetHandlerTest {
when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID))
.thenReturn(createServerConfig(OTHER_SERVER_ID, OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT, true));
_handler.onInstanceConfigChange(List.of(), createInitContext());
- verify(_mailboxService, never()).resetConnectBackoff(any(), anyInt());
+ verify(_mailboxService, never()).refreshChannel(any(), anyInt());
// Subsequent individual callback: server completes startup
when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID))
.thenReturn(createServerConfig(OTHER_SERVER_ID, OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT, false));
_handler.onInstanceConfigChange(List.of(),
createCallbackContextForInstance(OTHER_SERVER_ID));
- verify(_mailboxService).resetConnectBackoff(OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT);
+ verify(_mailboxService).refreshChannel(OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT);
+ }
+
+ /// A server that comes back on another mailbox port (the default when ports
are not pinned) leaves a channel to its
+ /// old address that nothing will use again.
+ @Test
+ public void testRetiresChannelToOldAddressWhenServerMoves() {
+ when(_helixAdmin.getInstancesInCluster(CLUSTER_NAME))
+ .thenReturn(Arrays.asList(SELF_INSTANCE_ID, OTHER_SERVER_ID));
+ when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID))
+ .thenReturn(createServerConfig(OTHER_SERVER_ID, OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT, false));
+ _handler.onInstanceConfigChange(List.of(), createInitContext());
+
+ // A config change that keeps the address retires nothing
+ _handler.onInstanceConfigChange(List.of(),
createCallbackContextForInstance(OTHER_SERVER_ID));
+ verify(_mailboxService, never()).retireChannel(any(), anyInt());
+
+ when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID))
+ .thenReturn(createServerConfig(OTHER_SERVER_ID, OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT + 1, true));
+ _handler.onInstanceConfigChange(List.of(),
createCallbackContextForInstance(OTHER_SERVER_ID));
+ verify(_mailboxService).retireChannel(OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT);
+ verify(_mailboxService, never()).retireChannel(OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT + 1);
+ }
+
+ @Test
+ public void testRetiresChannelOfServerThatLeftTheCluster() {
+ when(_helixAdmin.getInstancesInCluster(CLUSTER_NAME))
+ .thenReturn(Arrays.asList(SELF_INSTANCE_ID, OTHER_SERVER_ID));
+ when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID))
+ .thenReturn(createServerConfig(OTHER_SERVER_ID, OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT, false));
+ _handler.onInstanceConfigChange(List.of(), createInitContext());
+
+
when(_helixAdmin.getInstancesInCluster(CLUSTER_NAME)).thenReturn(List.of(SELF_INSTANCE_ID));
+ _handler.onInstanceConfigChange(List.of(), createChildChangeContext());
+ verify(_mailboxService).retireChannel(OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT);
+ }
+
+ /// A config that cannot be read now does not mean that the server left: its
channel stays.
+ @Test
+ public void testDoesNotRetireChannelWhenInstanceConfigReadFails() {
+ when(_helixAdmin.getInstancesInCluster(CLUSTER_NAME))
+ .thenReturn(Arrays.asList(SELF_INSTANCE_ID, OTHER_SERVER_ID));
+ when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID))
+ .thenReturn(createServerConfig(OTHER_SERVER_ID, OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT, false));
+ _handler.onInstanceConfigChange(List.of(), createInitContext());
+
+ when(_helixAdmin.getInstanceConfig(CLUSTER_NAME,
OTHER_SERVER_ID)).thenThrow(new RuntimeException("ZK hiccup"));
+ _handler.onInstanceConfigChange(List.of(), createChildChangeContext());
+ verify(_mailboxService, never()).retireChannel(any(), anyInt());
+ }
+
+ /// The channel to an address must stay when another server now has that
address.
+ @Test
+ public void testKeepsChannelToAddressTakenOverByAnotherServer() {
+ String thirdServerId = "Server_thirdhost_8000";
+ when(_helixAdmin.getInstancesInCluster(CLUSTER_NAME))
+ .thenReturn(Arrays.asList(SELF_INSTANCE_ID, OTHER_SERVER_ID,
thirdServerId));
+ when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID))
+ .thenReturn(createServerConfig(OTHER_SERVER_ID, OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT, false));
+ when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, thirdServerId))
+ .thenReturn(createServerConfig(thirdServerId, "thirdhost",
OTHER_SERVER_MAILBOX_PORT, false));
+ _handler.onInstanceConfigChange(List.of(), createInitContext());
+
+ // The third server takes over the address of the other server, which
moves away
+ when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, thirdServerId))
+ .thenReturn(createServerConfig(thirdServerId, OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT, false));
+ _handler.onInstanceConfigChange(List.of(),
createCallbackContextForInstance(thirdServerId));
+ when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID))
+ .thenReturn(createServerConfig(OTHER_SERVER_ID, "newhost",
OTHER_SERVER_MAILBOX_PORT, false));
+ _handler.onInstanceConfigChange(List.of(),
createCallbackContextForInstance(OTHER_SERVER_ID));
+
+ verify(_mailboxService).retireChannel("thirdhost",
OTHER_SERVER_MAILBOX_PORT);
+ verify(_mailboxService, never()).retireChannel(OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT);
+ }
+
+ /// A full scan applies every server's new address before it retires any, so
it keeps the channel to an address that
+ /// one server left and another one took over in the same scan, whichever of
the two it reads first.
+ @Test
+ public void testFullScanKeepsChannelToAddressTakenOverInTheSameScan() {
+ String thirdServerId = "Server_thirdhost_8000";
+ when(_helixAdmin.getInstancesInCluster(CLUSTER_NAME))
+ .thenReturn(Arrays.asList(SELF_INSTANCE_ID, OTHER_SERVER_ID,
thirdServerId));
+ when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID))
+ .thenReturn(createServerConfig(OTHER_SERVER_ID, OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT, false));
+ when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, thirdServerId))
+ .thenReturn(createServerConfig(thirdServerId, "thirdhost",
OTHER_SERVER_MAILBOX_PORT, false));
+ _handler.onInstanceConfigChange(List.of(), createInitContext());
+
+ // One scan reads the other server first: it moved away, and the third
server took over its address
+ when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID))
+ .thenReturn(createServerConfig(OTHER_SERVER_ID, "newhost",
OTHER_SERVER_MAILBOX_PORT, false));
+ when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, thirdServerId))
+ .thenReturn(createServerConfig(thirdServerId, OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT, false));
+ _handler.onInstanceConfigChange(List.of(), createChildChangeContext());
+
+ verify(_mailboxService).retireChannel("thirdhost",
OTHER_SERVER_MAILBOX_PORT);
+ verify(_mailboxService, never()).retireChannel(OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT);
}
private InstanceConfig createServerConfig(String instanceId, String
hostname, int mailboxPort,
@@ -234,7 +330,7 @@ public class ServerGrpcChannelBackoffResetHandlerTest {
// Simulate a FINALIZE notification where pathChanged is null. The
listener should ignore this.
_handler.onInstanceConfigChange(List.of(),
createFinalizeContextWithNullPath());
- verify(_mailboxService, never()).resetConnectBackoff(any(), anyInt());
+ verify(_mailboxService, never()).refreshChannel(any(), anyInt());
verify(_helixAdmin, never()).getInstancesInCluster(any());
}
@@ -245,6 +341,13 @@ public class ServerGrpcChannelBackoffResetHandlerTest {
return context;
}
+ private NotificationContext createChildChangeContext() {
+ NotificationContext context = new NotificationContext(_helixManager);
+ context.setType(NotificationContext.Type.CALLBACK);
+ context.setIsChildChange(true);
+ return context;
+ }
+
private NotificationContext createInitContext() {
NotificationContext context = new NotificationContext(_helixManager);
context.setType(NotificationContext.Type.INIT);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
