This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new bfdb305e148 Skip mailbox config scans for non-server membership
changes (#20107)
bfdb305e148 is described below
commit bfdb305e148ff4ccb1ac0bee3062aedb55ee0dcc
Author: Xiang Fu <[email protected]>
AuthorDate: Fri Oct 30 21:30:53 2026 -0700
Skip mailbox config scans for non-server membership changes (#20107)
* Skip mailbox config scans for non-server membership changes
Cache participant names so minion membership changes do not reread all
server configs. Preserve initialization, server membership scans, and retries
after failed reads.
Reproduction: initialize with a server, then add and remove a minion. The
regression test previously observed three server-config reads instead of one.
* Preserve mailbox recovery when skipping membership scans
Check a coherent parent generation and child version before skipping
non-server membership changes. Invalidate the snapshot after failed reads and
retain known shutdown state while retrying.
Reproduction: coalesce a same-name server recreation with a minion join, or
fail a readiness config read before a minion membership callback. Both cases
previously missed the channel refresh.
* Simplify mailbox membership snapshot handling
Use an immutable cache entry and keep the skip decision and snapshot
publication together. Preserve all recovery and reconciliation safeguards.
How to reproduce:
Run ServerMailboxChannelRefreshHandlerTest. Server add/remove snapshots now
advance cversion from 10 to 11 so they exercise the server-type guard.
---
.../helix/ServerMailboxChannelRefreshHandler.java | 81 +++++++-
.../ServerMailboxChannelRefreshHandlerTest.java | 215 +++++++++++++++++++--
2 files changed, 275 insertions(+), 21 deletions(-)
diff --git
a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/ServerMailboxChannelRefreshHandler.java
b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/ServerMailboxChannelRefreshHandler.java
index 331d88e12d9..578bd5476bc 100644
---
a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/ServerMailboxChannelRefreshHandler.java
+++
b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/ServerMailboxChannelRefreshHandler.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.server.starter.helix;
+import com.google.common.collect.Sets;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -25,16 +26,21 @@ import java.util.Map;
import java.util.Set;
import javax.annotation.Nullable;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.helix.AccessOption;
+import org.apache.helix.BaseDataAccessor;
import org.apache.helix.HelixAdmin;
+import org.apache.helix.HelixDataAccessor;
import org.apache.helix.NotificationContext;
import org.apache.helix.api.listeners.BatchMode;
import org.apache.helix.api.listeners.InstanceConfigChangeListener;
import org.apache.helix.api.listeners.PreFetch;
import org.apache.helix.model.InstanceConfig;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.pinot.core.transport.ServerInstance;
import org.apache.pinot.query.mailbox.MailboxService;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.InstanceTypeUtils;
+import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -68,6 +74,17 @@ public class ServerMailboxChannelRefreshHandler implements
InstanceConfigChangeL
private final String _selfInstanceId;
private final MailboxService _mailboxService;
+ // A coherent participant snapshot, invalidated whenever a membership or
instance config read fails.
+ @Nullable
+ private MembershipSnapshot _membershipSnapshot;
+
+ /// Immutable participant names and parent version from a coherent
membership read.
+ private record MembershipSnapshot(Set<String> instances, long parentCzxid,
int parentCversion) {
+ private MembershipSnapshot {
+ instances = Set.copyOf(instances);
+ }
+ }
+
// 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
refresh its channel.
private final Set<String> _shuttingDownServers = new HashSet<>();
@@ -90,10 +107,10 @@ public class ServerMailboxChannelRefreshHandler implements
InstanceConfigChangeL
// INIT: first callback when the listener is registered (full cluster
snapshot).
// isChildChange: an instance ZK node was added or removed under
/CONFIGS/PARTICIPANT.
- // Both require a full scan to rebuild _shuttingDownServers from the
current cluster state.
+ // INIT always scans; child changes scan only if they could affect servers.
NotificationContext.Type type = context.getType();
if (type == NotificationContext.Type.INIT || context.getIsChildChange()) {
- handleFullScan();
+ handleMembershipChange(context);
} else if (type == NotificationContext.Type.CALLBACK) {
// An existing instance's config changed (e.g. IS_SHUTDOWN_IN_PROGRESS
toggled).
// pathChanged is the ZK path of the specific instance that changed.
@@ -103,16 +120,57 @@ public class ServerMailboxChannelRefreshHandler
implements InstanceConfigChangeL
}
}
- /// Full cluster scan: used on INIT (first callback) and when instances are
added/removed.
+ /// Scans on INIT and child changes, skipping membership changes confined to
non-server instances.
+ private void handleMembershipChange(NotificationContext context) {
+ MembershipSnapshot previousSnapshot = _membershipSnapshot;
+ _membershipSnapshot = null;
+ HelixDataAccessor dataAccessor =
context.getManager().getHelixDataAccessor();
+ BaseDataAccessor<ZNRecord> baseDataAccessor =
dataAccessor.getBaseDataAccessor();
+ String parentPath = dataAccessor.keyBuilder().instanceConfigs().getPath();
+ Set<String> instances;
+ Stat before;
+ Stat after;
+ try {
+ // Helix has no atomic child-names/Stat read. A changing parent makes
this snapshot unsafe to cache.
+ before = baseDataAccessor.getStat(parentPath, AccessOption.PERSISTENT);
+ instances = new HashSet<>(baseDataAccessor.getChildNames(parentPath,
AccessOption.PERSISTENT));
+ after = baseDataAccessor.getStat(parentPath, AccessOption.PERSISTENT);
+ } catch (Exception e) {
+ LOGGER.warn("Failed to read participant membership version, scanning
server configs", e);
+ before = null;
+ after = null;
+ instances = new
HashSet<>(_helixAdmin.getInstancesInCluster(_clusterName));
+ }
+ boolean stableSnapshot = before != null && after != null &&
before.getCzxid() == after.getCzxid()
+ && before.getCversion() == after.getCversion() && before.getPzxid() ==
after.getPzxid();
+ if (context.getType() != NotificationContext.Type.INIT && previousSnapshot
!= null && stableSnapshot) {
+ Set<String> changedInstances =
Sets.symmetricDifference(previousSnapshot.instances(), instances);
+ // cversion counts all child creates/deletes, including same-name server
recreation hidden by the net diff.
+ // Long subtraction conservatively rejects counter wrap/reset; czxid
identifies parent recreation.
+ if (after.getCzxid() == previousSnapshot.parentCzxid()
+ && (long) after.getCversion() - previousSnapshot.parentCversion() ==
changedInstances.size()
+ && !changedInstances.isEmpty()
+ && changedInstances.stream().noneMatch(InstanceTypeUtils::isServer))
{
+ _membershipSnapshot = new MembershipSnapshot(instances,
after.getCzxid(), after.getCversion());
+ return;
+ }
+ }
+ if (scanServerConfigs(instances) && stableSnapshot) {
+ _membershipSnapshot = new MembershipSnapshot(instances,
after.getCzxid(), after.getCversion());
+ }
+ }
+
+ /// Reconciles server configs, returning whether every config was read
successfully.
///
/// 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() {
+ private boolean scanServerConfigs(Set<String> instances) {
+ boolean allConfigsRead = true;
Set<String> currentShuttingDown = new HashSet<>();
Set<String> currentServers = new HashSet<>();
Map<String, Pair<String, Integer>> previousAddresses = new HashMap<>();
- for (String instance : _helixAdmin.getInstancesInCluster(_clusterName)) {
+ for (String instance : instances) {
if (!InstanceTypeUtils.isServer(instance) ||
instance.equals(_selfInstanceId)) {
continue;
}
@@ -120,6 +178,10 @@ public class ServerMailboxChannelRefreshHandler implements
InstanceConfigChangeL
currentServers.add(instance);
InstanceConfig config = getInstanceConfig(instance);
if (config == null) {
+ allConfigsRead = false;
+ if (_shuttingDownServers.contains(instance)) {
+ currentShuttingDown.add(instance);
+ }
continue;
}
Pair<String, Integer> previousAddress = recordMailboxAddress(instance,
config);
@@ -145,6 +207,7 @@ public class ServerMailboxChannelRefreshHandler implements
InstanceConfigChangeL
}
_shuttingDownServers.clear();
_shuttingDownServers.addAll(currentShuttingDown);
+ return allConfigsRead;
}
/// Handles a single instance config change by reading only the changed
instance's config
@@ -168,11 +231,15 @@ public class ServerMailboxChannelRefreshHandler
implements InstanceConfigChangeL
@Nullable
private InstanceConfig getInstanceConfig(String instance) {
try {
- return _helixAdmin.getInstanceConfig(_clusterName, instance);
+ InstanceConfig config = _helixAdmin.getInstanceConfig(_clusterName,
instance);
+ if (config != null) {
+ return config;
+ }
} catch (Exception e) {
LOGGER.warn("Failed to get instance config for: {}, skipping", instance,
e);
- return null;
}
+ _membershipSnapshot = null;
+ return null;
}
private static boolean isShutdownInProgress(InstanceConfig config) {
diff --git
a/pinot-server/src/test/java/org/apache/pinot/server/starter/helix/ServerMailboxChannelRefreshHandlerTest.java
b/pinot-server/src/test/java/org/apache/pinot/server/starter/helix/ServerMailboxChannelRefreshHandlerTest.java
index 3fddb06e72b..36ae0337396 100644
---
a/pinot-server/src/test/java/org/apache/pinot/server/starter/helix/ServerMailboxChannelRefreshHandlerTest.java
+++
b/pinot-server/src/test/java/org/apache/pinot/server/starter/helix/ServerMailboxChannelRefreshHandlerTest.java
@@ -18,24 +18,31 @@
*/
package org.apache.pinot.server.starter.helix;
-import java.util.Arrays;
import java.util.List;
+import org.apache.helix.AccessOption;
+import org.apache.helix.BaseDataAccessor;
import org.apache.helix.HelixAdmin;
+import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixManager;
import org.apache.helix.NotificationContext;
+import org.apache.helix.PropertyKey;
import org.apache.helix.model.InstanceConfig;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.pinot.query.mailbox.MailboxService;
import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.zookeeper.data.Stat;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
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.any;
import static org.mockito.Mockito.anyInt;
+import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -46,6 +53,8 @@ public class ServerMailboxChannelRefreshHandlerTest {
private static final String OTHER_SERVER_ID = "Server_otherhost_8000";
private static final String OTHER_SERVER_HOST = "otherhost";
private static final int OTHER_SERVER_MAILBOX_PORT = 9000;
+ private static final String INSTANCE_CONFIGS_PATH =
+ new PropertyKey.Builder(CLUSTER_NAME).instanceConfigs().getPath();
private AutoCloseable _mocks;
@@ -55,6 +64,12 @@ public class ServerMailboxChannelRefreshHandlerTest {
@Mock
private HelixAdmin _helixAdmin;
+ @Mock
+ private HelixDataAccessor _helixDataAccessor;
+
+ @Mock
+ private BaseDataAccessor<ZNRecord> _baseDataAccessor;
+
@Mock
private MailboxService _mailboxService;
@@ -63,6 +78,9 @@ public class ServerMailboxChannelRefreshHandlerTest {
@BeforeMethod
public void setUp() {
_mocks = MockitoAnnotations.openMocks(this);
+ when(_helixManager.getHelixDataAccessor()).thenReturn(_helixDataAccessor);
+ when(_helixDataAccessor.keyBuilder()).thenReturn(new
PropertyKey.Builder(CLUSTER_NAME));
+
when(_helixDataAccessor.getBaseDataAccessor()).thenReturn(_baseDataAccessor);
_handler = new ServerMailboxChannelRefreshHandler(_helixAdmin,
CLUSTER_NAME, SELF_INSTANCE_ID, _mailboxService);
}
@@ -190,8 +208,7 @@ public class ServerMailboxChannelRefreshHandlerTest {
@Test
public void testInitCallbackDoesFullScanAndTracksState() {
- when(_helixAdmin.getInstancesInCluster(CLUSTER_NAME))
- .thenReturn(Arrays.asList(SELF_INSTANCE_ID, OTHER_SERVER_ID));
+ stubParticipantSnapshot(List.of(SELF_INSTANCE_ID, OTHER_SERVER_ID));
// INIT callback: server is shutting down, triggers full cluster scan
when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID))
@@ -206,12 +223,163 @@ public class ServerMailboxChannelRefreshHandlerTest {
verify(_mailboxService).refreshChannel(OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT);
}
+ @Test
+ public void testMinionMembershipChangesDoNotReadServerConfigs() {
+ String minionId = "Minion_minionhost_8000";
+ stubParticipantSnapshot(List.of(SELF_INSTANCE_ID, OTHER_SERVER_ID), 10, 1);
+ 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());
+
+ stubParticipantSnapshot(List.of(minionId, OTHER_SERVER_ID,
SELF_INSTANCE_ID), 11, 1);
+ _handler.onInstanceConfigChange(List.of(), createChildChangeContext());
+ stubParticipantSnapshot(List.of(SELF_INSTANCE_ID, OTHER_SERVER_ID), 12, 1);
+ _handler.onInstanceConfigChange(List.of(), createChildChangeContext());
+ verify(_helixAdmin, times(1)).getInstanceConfig(CLUSTER_NAME,
OTHER_SERVER_ID);
+ verify(_mailboxService, never()).refreshChannel(any(), anyInt());
+
+ 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).refreshChannel(OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT);
+ }
+
+ @Test
+ public void testServerMembershipChangesStillScanConfigs() {
+ stubParticipantSnapshot(List.of(SELF_INSTANCE_ID), 10, 1);
+ 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());
+ stubParticipantSnapshot(List.of(SELF_INSTANCE_ID, OTHER_SERVER_ID), 11, 1);
+ _handler.onInstanceConfigChange(List.of(), createChildChangeContext());
+ verify(_helixAdmin).getInstanceConfig(CLUSTER_NAME, 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());
+ verify(_mailboxService).refreshChannel(OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT);
+ }
+
+ @DataProvider(name = "ambiguousMembershipChanges")
+ public static Object[][] ambiguousMembershipChanges() {
+ return new Object[][]{
+ // A server delete/recreate and a minion join are coalesced into one
callback.
+ {10, 13, 1L, 1L, true},
+ // A recreated parent has an unrelated cversion, even if its delta
matches the minion join.
+ {10, 11, 1L, 2L, true},
+ // Integer wrap must not look like an unambiguous single child change.
+ {Integer.MAX_VALUE, Integer.MIN_VALUE, 1L, 1L, true},
+ // The parent changed while its child names were being read.
+ {10, 11, 1L, 1L, false}
+ };
+ }
+
+ @Test(dataProvider = "ambiguousMembershipChanges")
+ public void testAmbiguousMinionMembershipChangeStillTracksServerStartup(int
previousCversion, int cversion,
+ long previousCzxid, long czxid, boolean stableSnapshot) {
+ stubParticipantSnapshot(List.of(SELF_INSTANCE_ID, OTHER_SERVER_ID),
previousCversion, previousCzxid);
+ 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());
+
+ stubParticipantSnapshot(List.of(SELF_INSTANCE_ID, OTHER_SERVER_ID,
"Minion_minionhost_8000"), cversion, czxid);
+ if (!stableSnapshot) {
+ when(_baseDataAccessor.getStat(INSTANCE_CONFIGS_PATH,
AccessOption.PERSISTENT))
+ .thenReturn(participantStat(previousCversion, czxid, 2),
participantStat(cversion, czxid, 3));
+ }
+ 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(), createChildChangeContext());
+ verify(_mailboxService, never()).refreshChannel(any(), anyInt());
+
+ 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).refreshChannel(OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT);
+ }
+
+ @DataProvider(name = "retryScanFailure")
+ public static Object[][] retryScanFailure() {
+ return new Object[][]{{false}, {true}};
+ }
+
+ @Test(dataProvider = "retryScanFailure")
+ public void testFailedReadRetriesOnMinionMembershipChange(boolean
failRetryScan) {
+ stubParticipantSnapshot(List.of(SELF_INSTANCE_ID, OTHER_SERVER_ID), 10, 1);
+ 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());
+
+ when(_helixAdmin.getInstanceConfig(CLUSTER_NAME,
OTHER_SERVER_ID)).thenThrow(new RuntimeException("ZK hiccup"));
+ _handler.onInstanceConfigChange(List.of(),
createCallbackContextForInstance(OTHER_SERVER_ID));
+ verify(_mailboxService, never()).refreshChannel(any(), anyInt());
+
+ stubParticipantSnapshot(List.of(SELF_INSTANCE_ID, OTHER_SERVER_ID,
"Minion_minionhost_8000"), 11, 1);
+ if (failRetryScan) {
+ _handler.onInstanceConfigChange(List.of(), createChildChangeContext());
+ stubParticipantSnapshot(List.of(SELF_INSTANCE_ID, OTHER_SERVER_ID,
+ "Minion_minionhost_8000", "Minion_anotherhost_8000"), 12, 1);
+ }
+ doReturn(createServerConfig(OTHER_SERVER_ID, OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT, false))
+ .when(_helixAdmin).getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID);
+ _handler.onInstanceConfigChange(List.of(), createChildChangeContext());
+ verify(_mailboxService).refreshChannel(OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT);
+ }
+
+ @Test
+ public void testFailedInitReadRetriesOnMinionMembershipChange() {
+ stubParticipantSnapshot(List.of(SELF_INSTANCE_ID, OTHER_SERVER_ID), 10, 1);
+ when(_helixAdmin.getInstanceConfig(CLUSTER_NAME,
OTHER_SERVER_ID)).thenThrow(new RuntimeException("ZK hiccup"));
+ _handler.onInstanceConfigChange(List.of(), createInitContext());
+
+ stubParticipantSnapshot(List.of(SELF_INSTANCE_ID, OTHER_SERVER_ID,
"Minion_minionhost_8000"), 11, 1);
+ doReturn(createServerConfig(OTHER_SERVER_ID, OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT, true))
+ .when(_helixAdmin).getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID);
+ _handler.onInstanceConfigChange(List.of(), createChildChangeContext());
+ verify(_mailboxService, never()).refreshChannel(any(), anyInt());
+
+ 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).refreshChannel(OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT);
+ }
+
+ @DataProvider(name = "metadataReadFailure")
+ public static Object[][] metadataReadFailure() {
+ return new Object[][]{{false}, {true}};
+ }
+
+ @Test(dataProvider = "metadataReadFailure")
+ public void
testMembershipMetadataReadFailureStillTracksServerStartup(boolean
throwOnStatRead) {
+ stubParticipantSnapshot(List.of(SELF_INSTANCE_ID, OTHER_SERVER_ID), 10, 1);
+ 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());
+
+ stubParticipantSnapshot(List.of(SELF_INSTANCE_ID, OTHER_SERVER_ID,
"Minion_minionhost_8000"), 11, 1);
+ // Missing metadata cannot prove that the minion join was the only change.
+ if (throwOnStatRead) {
+ when(_baseDataAccessor.getStat(INSTANCE_CONFIGS_PATH,
AccessOption.PERSISTENT))
+ .thenThrow(new RuntimeException("ZK hiccup"));
+ } else {
+ when(_baseDataAccessor.getStat(INSTANCE_CONFIGS_PATH,
AccessOption.PERSISTENT)).thenReturn(null);
+ }
+ 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(), createChildChangeContext());
+ verify(_mailboxService, never()).refreshChannel(any(), anyInt());
+
+ 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).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));
+ stubParticipantSnapshot(List.of(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());
@@ -229,13 +397,12 @@ public class ServerMailboxChannelRefreshHandlerTest {
@Test
public void testRetiresChannelOfServerThatLeftTheCluster() {
- when(_helixAdmin.getInstancesInCluster(CLUSTER_NAME))
- .thenReturn(Arrays.asList(SELF_INSTANCE_ID, OTHER_SERVER_ID));
+ stubParticipantSnapshot(List.of(SELF_INSTANCE_ID, OTHER_SERVER_ID), 10, 1);
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));
+ stubParticipantSnapshot(List.of(SELF_INSTANCE_ID), 11, 1);
_handler.onInstanceConfigChange(List.of(), createChildChangeContext());
verify(_mailboxService).retireChannel(OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT);
}
@@ -243,8 +410,7 @@ public class ServerMailboxChannelRefreshHandlerTest {
/// 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));
+ stubParticipantSnapshot(List.of(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());
@@ -258,8 +424,7 @@ public class ServerMailboxChannelRefreshHandlerTest {
@Test
public void testKeepsChannelToAddressTakenOverByAnotherServer() {
String thirdServerId = "Server_thirdhost_8000";
- when(_helixAdmin.getInstancesInCluster(CLUSTER_NAME))
- .thenReturn(Arrays.asList(SELF_INSTANCE_ID, OTHER_SERVER_ID,
thirdServerId));
+ stubParticipantSnapshot(List.of(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))
@@ -283,8 +448,7 @@ public class ServerMailboxChannelRefreshHandlerTest {
@Test
public void testFullScanKeepsChannelToAddressTakenOverInTheSameScan() {
String thirdServerId = "Server_thirdhost_8000";
- when(_helixAdmin.getInstancesInCluster(CLUSTER_NAME))
- .thenReturn(Arrays.asList(SELF_INSTANCE_ID, OTHER_SERVER_ID,
thirdServerId));
+ stubParticipantSnapshot(List.of(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))
@@ -302,6 +466,27 @@ public class ServerMailboxChannelRefreshHandlerTest {
verify(_mailboxService, never()).retireChannel(OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT);
}
+ private void stubParticipantSnapshot(List<String> instances) {
+ stubParticipantSnapshot(instances, 0, 1);
+ }
+
+ private void stubParticipantSnapshot(List<String> instances, int cversion,
long czxid) {
+ // The Admin name list is the fallback when the versioned membership read
throws.
+
when(_helixAdmin.getInstancesInCluster(CLUSTER_NAME)).thenReturn(instances);
+ when(_baseDataAccessor.getChildNames(INSTANCE_CONFIGS_PATH,
AccessOption.PERSISTENT)).thenReturn(instances);
+ when(_baseDataAccessor.getStat(INSTANCE_CONFIGS_PATH,
AccessOption.PERSISTENT))
+ .thenReturn(participantStat(cversion, czxid, instances.size()));
+ }
+
+ private static Stat participantStat(int cversion, long czxid, int
numChildren) {
+ Stat stat = new Stat();
+ stat.setCversion(cversion);
+ stat.setCzxid(czxid);
+ stat.setPzxid(100L + Integer.toUnsignedLong(cversion));
+ stat.setNumChildren(numChildren);
+ return stat;
+ }
+
private InstanceConfig createServerConfig(String instanceId, String
hostname, int mailboxPort,
boolean shutdownInProgress) {
ZNRecord record = new ZNRecord(instanceId);
@@ -332,6 +517,8 @@ public class ServerMailboxChannelRefreshHandlerTest {
verify(_mailboxService, never()).refreshChannel(any(), anyInt());
verify(_helixAdmin, never()).getInstancesInCluster(any());
+ verify(_baseDataAccessor, never()).getStat(any(), anyInt());
+ verify(_baseDataAccessor, never()).getChildNames(any(), anyInt());
}
private NotificationContext createFinalizeContextWithNullPath() {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
