This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
The following commit(s) were added to refs/heads/master by this push:
new 08c6870bf Fix 2 reopened flaky CI tests (#3146, #3178)
08c6870bf is described below
commit 08c6870bf6d8adec1d2dff584ec52e50218ed4e7
Author: Junkai Xue <[email protected]>
AuthorDate: Wed Nov 4 14:06:50 2026 -0800
Fix 2 reopened flaky CI tests (#3146, #3178)
TestInstanceOperation.testNodeSwapBasic (#3146): after
completeSwapIfPossible
the test demanded an exact in-place copy of the pre-swap external view
(swap-in substituted for swap-out, everything else identical). CI showed
leaders and even membership moving on partitions unrelated to the pair once
the swap completed. Once the swap-in is ENABLE and the swap-out is UNKNOWN
the controller no longer pairs the two instances; the final assignment is an
ordinary rebalance over the new node set, so the mirror is only guaranteed
while the SWAP_IN is in flight (still asserted). Every post-completion check
of this kind (testNodeSwapBasic,
testSwapNotAffectedByDisableReenableOfSwapOut,
testSwapInCanBeSetAfterInstanceAdded, testSwapInDuringMaintenanceMode,
testDisabledPartitionsBeforeSwap, testDisabledPartitionsAfterSwap) now uses
validateReplacementAssignment, the check already adopted for the sibling
swap/evacuate tests in the previous round: swap-out holds nothing, every
partition fully replicated with exactly one top-state replica, swap-in hosts
load. testNodeSwapBasic re-enables the carried-over disabled partition
before
the final check, using the resource/partition overload (the deprecated 2-arg
overload it used before edits a legacy list field the controller does not
read, so that re-enable was a no-op). The two one-shot post-completion
validations now poll. waitForSwapProcessed now waits for the actual
post-completion signal (swap-out holds no replica, swap-in holds one)
instead
of insisting the swap-in land on exactly the swap-out's partitions, which
made
it spin for its full 30s on the same shift. Top-state lookups in
validateReplacementAssignment are cached since it runs inside 50ms polls.
TestConsecutiveZkSessionExpiry.testDistributedController (#3178): the double
session expiry left the cluster non-converged for the whole verifier window
(30s in Sept, 120s in Nov), so timeouts are not the issue. It did not
reproduce locally in 18 runs including a CPU-stressed loop, and the CI log
level is ERROR-only so nothing about the stuck state was captured. The test
now (1) waits for the expired manager to actually re-create its LIVEINSTANCE
under the new session instead of only for the ZK client's session id to
change, which lets the test start the convergence check from a known-good
point and separates "manager never re-joined" from "cluster never
converged", and (2) dumps leader, live instances, per-session current
states, pending messages and external views to stdout when either wait
fails, so the next CI failure is diagnosable. getSessionId() is guarded in
the poll and in the dump (it throws when the manager is disconnected) and
the dump tolerates znodes disappearing underneath it.
No production code changed.
---
.../manager/TestConsecutiveZkSessionExpiry.java | 166 +++++++++++++++++----
.../rebalancer/TestInstanceOperation.java | 138 ++++++++---------
2 files changed, 197 insertions(+), 107 deletions(-)
diff --git
a/helix-core/src/test/java/org/apache/helix/integration/manager/TestConsecutiveZkSessionExpiry.java
b/helix-core/src/test/java/org/apache/helix/integration/manager/TestConsecutiveZkSessionExpiry.java
index 0254073b7..57469a671 100644
---
a/helix-core/src/test/java/org/apache/helix/integration/manager/TestConsecutiveZkSessionExpiry.java
+++
b/helix-core/src/test/java/org/apache/helix/integration/manager/TestConsecutiveZkSessionExpiry.java
@@ -21,9 +21,11 @@ package org.apache.helix.integration.manager;
import java.util.Date;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.CountDownLatch;
import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixManager;
import org.apache.helix.PreConnectCallback;
import org.apache.helix.PropertyKey;
import org.apache.helix.TestHelper;
@@ -34,7 +36,10 @@ import org.apache.helix.manager.zk.CallbackHandler;
import org.apache.helix.manager.zk.ZKHelixDataAccessor;
import org.apache.helix.manager.zk.ZkBaseDataAccessor;
import org.apache.helix.mock.participant.MockMSModelFactory;
+import org.apache.helix.model.CurrentState;
+import org.apache.helix.model.ExternalView;
import org.apache.helix.model.LiveInstance;
+import org.apache.helix.model.Message;
import org.apache.helix.tools.ClusterStateVerifier;
import
org.apache.helix.tools.ClusterStateVerifier.BestPossAndExtViewZkVerifier;
import org.slf4j.Logger;
@@ -146,23 +151,12 @@ public class TestConsecutiveZkSessionExpiry extends
ZkUnitTestBase {
endCountdown.countDown();
- // Wait for the participant to fully reconnect with a new session before
verifying cluster
- // state. Recovering from consecutive session expiries (re-joining the
cluster and carrying
- // over the current states) can take a while on overloaded machines, and
the default 30s
- // verifier window was not enough.
- final MockParticipantManager expiredParticipant = participants[0];
- final String expiredSessionId = oldSessionId;
- Assert.assertTrue(TestHelper.verify(() -> {
- String currentSessionId = expiredParticipant.getSessionId();
- return expiredParticipant.isConnected() && currentSessionId != null
- && !currentSessionId.equals(expiredSessionId);
- }, TestHelper.WAIT_DURATION),
- "Participant should reconnect with a new session after consecutive
expiries");
-
- result =
- ClusterStateVerifier.verifyByPolling(new
ClusterStateVerifier.BestPossAndExtViewZkVerifier(
- ZK_ADDR, clusterName), 2 * TestHelper.WAIT_DURATION);
- Assert.assertTrue(result);
+ // Wait for the participant to fully re-join the cluster under its new
session before
+ // verifying cluster state. Recovering from consecutive session expiries
(re-joining the
+ // cluster and carrying over the current states) can take a while on
overloaded machines.
+ waitForRejoinWithNewSession(participants[0], clusterName, oldSessionId);
+
+ verifyClusterConverged(clusterName);
// clean up
controller.syncStop();
@@ -238,22 +232,12 @@ public class TestConsecutiveZkSessionExpiry extends
ZkUnitTestBase {
endCountdown.countDown();
- // Wait for the distributed controller to fully reconnect with a new
session before
- // verifying cluster state. Recovering from consecutive session expiries
can be slow
+ // Wait for the distributed controller to fully re-join the cluster under
its new session
+ // before verifying cluster state. Recovering from consecutive session
expiries can be slow
// on overloaded machines.
- final ClusterDistributedController expiredController =
distributedControllers[0];
- final String expiredSessionId = oldSessionId;
- Assert.assertTrue(TestHelper.verify(() -> {
- String currentSessionId = expiredController.getSessionId();
- return expiredController.isConnected() && currentSessionId != null
- && !currentSessionId.equals(expiredSessionId);
- }, TestHelper.WAIT_DURATION),
- "Distributed controller should reconnect with a new session after
consecutive expiries");
-
- result =
- ClusterStateVerifier.verifyByPolling(new
ClusterStateVerifier.BestPossAndExtViewZkVerifier(
- ZK_ADDR, clusterName), 2 * TestHelper.WAIT_DURATION);
- Assert.assertTrue(result);
+ waitForRejoinWithNewSession(distributedControllers[0], clusterName,
oldSessionId);
+
+ verifyClusterConverged(clusterName);
// verify leader changes to localhost_12919
HelixDataAccessor accessor =
@@ -301,4 +285,122 @@ public class TestConsecutiveZkSessionExpiry extends
ZkUnitTestBase {
System.out.println("END " + clusterName + " at " + new
Date(System.currentTimeMillis()));
}
+ /**
+ * Waits until the manager has re-joined the cluster under a session other
than the expired one:
+ * the ZK client is connected with a new session id AND the LIVEINSTANCE
znode has been
+ * re-created by that session. The second condition is what matters for the
cluster state
+ * verification that follows: it means HelixManager#handleNewSession() for
the surviving session
+ * got past the live instance creation (the point where the handler for the
intermediate,
+ * expired-while-being-handled session fails by design in this test). Only
checking the client's
+ * session id would let the test proceed while the manager is still
(re)initializing.
+ */
+ private void waitForRejoinWithNewSession(final HelixManager manager, String
clusterName,
+ final String expiredSessionId) throws Exception {
+ final HelixDataAccessor accessor = accessor(clusterName);
+ final PropertyKey liveInstanceKey =
+ accessor.keyBuilder().liveInstance(manager.getInstanceName());
+ boolean rejoined = TestHelper.verify(() -> {
+ // getSessionId() throws when the manager is (or becomes) disconnected;
treat that as
+ // "not re-joined yet" and keep polling instead of aborting the wait.
+ String currentSessionId = currentSessionIdOrNull(manager);
+ if (currentSessionId == null ||
currentSessionId.equals(expiredSessionId)) {
+ return false;
+ }
+ LiveInstance liveInstance = accessor.getProperty(liveInstanceKey);
+ return liveInstance != null &&
currentSessionId.equals(liveInstance.getEphemeralOwner());
+ }, TestHelper.WAIT_DURATION);
+ if (!rejoined) {
+ dumpClusterState(clusterName, "instance " + manager.getInstanceName()
+ + " did not re-join with a new session (expired session: " +
expiredSessionId
+ + ", current session: " + currentSessionIdOrNull(manager) + ",
connected: "
+ + manager.isConnected() + ")");
+ }
+ Assert.assertTrue(rejoined, manager.getInstanceName()
+ + " should re-create its live instance with a new session after
consecutive expiries");
+ }
+
+ /**
+ * Verifies that the external views converge to the best possible state. On
failure, dumps the
+ * relevant cluster state to stdout (the CI log level is ERROR only, so test
logging is not
+ * available there) before failing, so that a non-converging cluster can be
diagnosed from the
+ * CI output instead of just "expected:<true> but was:<false>".
+ */
+ private void verifyClusterConverged(String clusterName) {
+ boolean result =
+ ClusterStateVerifier.verifyByPolling(new
ClusterStateVerifier.BestPossAndExtViewZkVerifier(
+ ZK_ADDR, clusterName), 2 * TestHelper.WAIT_DURATION);
+ if (!result) {
+ dumpClusterState(clusterName, "cluster did not converge after
consecutive session expiries");
+ }
+ Assert.assertTrue(result,
+ "Cluster should converge to the best possible state after consecutive
session expiries");
+ }
+
+ private HelixDataAccessor accessor(String clusterName) {
+ return new ZKHelixDataAccessor(clusterName, new
ZkBaseDataAccessor<ZNRecord>(_gZkClient));
+ }
+
+ private static String currentSessionIdOrNull(HelixManager manager) {
+ try {
+ return manager.isConnected() ? manager.getSessionId() : null;
+ } catch (Exception e) {
+ return null;
+ }
+ }
+
+ /**
+ * Best effort: every read below tolerates znodes disappearing underneath it
(which is exactly
+ * what happens around a session expiry) so that one failed section does not
hide the others.
+ */
+ private void dumpClusterState(String clusterName, String reason) {
+ StringBuilder sb = new StringBuilder();
+ sb.append("==== ").append(clusterName).append(":
").append(reason).append(" ====\n");
+ try {
+ HelixDataAccessor accessor = accessor(clusterName);
+ PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+
+ LiveInstance leader =
accessor.getProperty(keyBuilder.controllerLeader());
+ sb.append("LEADER: ").append(leader == null ? "none"
+ : leader.getInstanceName() + " (session " +
leader.getEphemeralOwner() + ")")
+ .append('\n');
+
+ Map<String, LiveInstance> liveInstances =
+ accessor.getChildValuesMap(keyBuilder.liveInstances(), false);
+ for (Map.Entry<String, LiveInstance> entry : liveInstances.entrySet()) {
+ sb.append("LIVEINSTANCE ").append(entry.getKey()).append(": session ")
+ .append(entry.getValue().getEphemeralOwner()).append('\n');
+ }
+
+ for (String instance : accessor.getChildNames(keyBuilder.instances())) {
+ for (String session :
accessor.getChildNames(keyBuilder.sessions(instance))) {
+ List<CurrentState> currentStates =
+ accessor.getChildValues(keyBuilder.currentStates(instance,
session), false);
+ for (CurrentState currentState : currentStates) {
+ sb.append("CURRENTSTATE
").append(instance).append('/').append(session).append('/')
+ .append(currentState.getResourceName()).append(": ")
+ .append(currentState.getPartitionStateMap()).append('\n');
+ }
+ }
+ List<Message> messages =
accessor.getChildValues(keyBuilder.messages(instance), false);
+ for (Message message : messages) {
+ sb.append("MESSAGE
").append(instance).append('/').append(message.getMsgId())
+ .append(": ").append(message.getMsgType()).append(' ')
+
.append(message.getResourceName()).append('/').append(message.getPartitionName())
+ .append('
').append(message.getFromState()).append("->").append(message.getToState())
+ .append("
tgtSession=").append(message.getTgtSessionId()).append(" state=")
+ .append(message.getMsgState()).append('\n');
+ }
+ }
+
+ List<ExternalView> externalViews =
+ accessor.getChildValues(keyBuilder.externalViews(), false);
+ for (ExternalView externalView : externalViews) {
+ sb.append("EXTERNALVIEW
").append(externalView.getResourceName()).append(": ")
+ .append(externalView.getRecord().getMapFields()).append('\n');
+ }
+ } catch (Exception e) {
+ sb.append("failed to dump cluster state: ").append(e).append('\n');
+ }
+ System.out.println(sb);
+ }
}
diff --git
a/helix-core/src/test/java/org/apache/helix/integration/rebalancer/TestInstanceOperation.java
b/helix-core/src/test/java/org/apache/helix/integration/rebalancer/TestInstanceOperation.java
index 35f874958..84cc3117d 100644
---
a/helix-core/src/test/java/org/apache/helix/integration/rebalancer/TestInstanceOperation.java
+++
b/helix-core/src/test/java/org/apache/helix/integration/rebalancer/TestInstanceOperation.java
@@ -123,6 +123,8 @@ public class TestInstanceOperation extends ZkTestBase {
// Test state
protected ClusterControllerManager _controller;
+ // Top state per resource, looked up once; validateReplacementAssignment
runs inside 50ms polls.
+ private final Map<String, String> _topStateByResource = new HashMap<>();
private HelixManager _spectator;
private RoutingTableProvider _routingTableProviderDefault;
private RoutingTableProvider _routingTableProviderEV;
@@ -556,12 +558,12 @@ public class TestInstanceOperation extends ZkTestBase {
// Disable one partition on swap-out for testing
String resourceToDisable = _allDBs.iterator().next();
- getPartitionsAndStatesOnInstance(getEVs(), swapOutName).entrySet().stream()
- .filter(entry -> entry.getKey().startsWith(resourceToDisable))
- .findFirst()
- .ifPresent(entry -> {
- swapOutConfig.setInstanceEnabledForPartition(resourceToDisable,
entry.getKey(), false);
- });
+ String partitionToDisable = getPartitionsAndStatesOnInstance(getEVs(),
swapOutName).keySet()
+ .stream().filter(partition ->
partition.startsWith(resourceToDisable)).findFirst()
+ .orElse(null);
+ Assert.assertNotNull(partitionToDisable,
+ "Swap-out " + swapOutName + " should host a partition of " +
resourceToDisable);
+ swapOutConfig.setInstanceEnabledForPartition(resourceToDisable,
partitionToDisable, false);
_gSetupTool.getClusterManagementTool()
.setInstanceConfig(CLUSTER_NAME, swapOutName, swapOutConfig);
@@ -606,7 +608,7 @@ public class TestInstanceOperation extends ZkTestBase {
// Wait for swap to be fully processed and EV to stabilize
// After completeSwapIfPossible, the controller needs time to update the EV
- waitForSwapProcessed(originalEVs, swapOutName, swapInName);
+ waitForSwapProcessed(swapOutName, swapInName);
// Verify swap-in is now in routing tables
validateRoutingTablesInstance(swapInName, true);
@@ -624,17 +626,24 @@ public class TestInstanceOperation extends ZkTestBase {
Assert.assertTrue(configListener.isThrottlesEnabled(),
"Throttles should be re-enabled after swap");
- // Verify final EV is correct
- verifier(() -> validateEVsCorrect(getEVs(), originalEVs, swapOutToSwapIn,
- Collections.emptySet(), ImmutableSet.of(swapInName)),
VERIFICATION_TIMEOUT);
-
- // Re-enable the disabled partition on swap-in
+ // The disabled partition was carried over from the swap-out's config.
Re-enable it on the
+ // swap-in (same resource/partition overload as the disable above; the
deprecated 2-arg
+ // overload edits a legacy list field the controller does not read) so the
final assignment
+ // can be validated as fully healthy.
InstanceConfig swapInConfig = _gSetupTool.getClusterManagementTool()
.getInstanceConfig(CLUSTER_NAME, swapInName);
- swapInConfig.setInstanceEnabledForPartition(resourceToDisable, true);
+ swapInConfig.setInstanceEnabledForPartition(resourceToDisable,
partitionToDisable, true);
_gSetupTool.getClusterManagementTool()
.setInstanceConfig(CLUSTER_NAME, swapInName, swapInConfig);
+ // Verify final EV. Once the swap is completed the swap-in is a regular
ENABLE instance and
+ // the swap-out is UNKNOWN, so the controller no longer pairs them and the
final assignment is
+ // an ordinary rebalance over the new node set, not an in-place copy of
the pre-swap EV (see
+ // validateReplacementAssignment). The exact mirror is asserted above
while the swap is in
+ // flight, which is the only phase where the product guarantees it.
+ verifier(() -> validateReplacementAssignment(getEVs(), originalEVs,
swapOutName, swapInName),
+ VERIFICATION_TIMEOUT * 2);
+
LOG.info("END testNodeSwapBasic");
}
@@ -701,16 +710,18 @@ public class TestInstanceOperation extends ZkTestBase {
"Swap should complete");
// Wait for swap to be fully processed and EV to stabilize
- waitForSwapProcessed(originalEVs, swapOutName, swapInName);
+ waitForSwapProcessed(swapOutName, swapInName);
// Verify swap-out is disabled
Assert.assertFalse(_gSetupTool.getClusterManagementTool()
.getInstanceConfig(CLUSTER_NAME, swapOutName).getInstanceEnabled(),
"Swap-out should be disabled");
- // Verify final EV
- verifier(() -> validateEVsCorrect(getEVs(), originalEVs, swapOutToSwapIn,
- Collections.emptySet(), ImmutableSet.of(swapInName)),
VERIFICATION_TIMEOUT);
+ // Verify final EV. The exact pre-swap mirror is only guaranteed while the
swap is in flight
+ // (asserted above); after completion the pair is no longer tracked and
the assignment is
+ // recomputed over the new node set, so check the product's guarantees
instead.
+ verifier(() -> validateReplacementAssignment(getEVs(), originalEVs,
swapOutName, swapInName),
+ VERIFICATION_TIMEOUT * 2);
LOG.info("END testSwapNotAffectedByDisableReenableOfSwapOut");
}
@@ -756,9 +767,10 @@ public class TestInstanceOperation extends ZkTestBase {
Assert.assertTrue(_clusterVerifier.verifyByPolling());
- // Verify final state
- verifier(() -> validateEVsCorrect(getEVs(), originalEVs, swapOutToSwapIn,
- Collections.emptySet(), ImmutableSet.of(swapInName)),
VERIFICATION_TIMEOUT);
+ // Verify final state. See testNodeSwapBasic: after completion the exact
pre-swap mirror is
+ // not guaranteed, only a healthy replacement assignment is.
+ verifier(() -> validateReplacementAssignment(getEVs(), originalEVs,
swapOutName, swapInName),
+ VERIFICATION_TIMEOUT * 2);
LOG.info("END testSwapInCanBeSetAfterInstanceAdded");
}
@@ -878,9 +890,10 @@ public class TestInstanceOperation extends ZkTestBase {
// Verify swap-in in routing tables
validateRoutingTablesInstance(swapInName, true);
- // Verify final EV
- verifier(() -> validateEVsCorrect(getEVs(), originalEVs, swapOutToSwapIn,
- Collections.emptySet(), ImmutableSet.of(swapInName)),
VERIFICATION_TIMEOUT);
+ // Verify final EV. See testNodeSwapBasic: after completion the exact
pre-swap mirror is
+ // not guaranteed, only a healthy replacement assignment is.
+ verifier(() -> validateReplacementAssignment(getEVs(), originalEVs,
swapOutName, swapInName),
+ VERIFICATION_TIMEOUT * 2);
LOG.info("END testSwapInDuringMaintenanceMode");
}
@@ -1289,8 +1302,11 @@ public class TestInstanceOperation extends ZkTestBase {
"Swap should complete");
Assert.assertTrue(_clusterVerifier.verifyByPolling());
- validateEVsCorrect(getEVs(), beforeEVs, swapOutToSwapIn,
- Collections.emptySet(), ImmutableSet.of(swapInName));
+ // See testNodeSwapBasic: after completion the exact pre-swap mirror is
not guaranteed, only a
+ // healthy replacement assignment is. The one-shot read is also replaced
by polling because
+ // the EV is rewritten asynchronously after completeSwapIfPossible.
+ verifier(() -> validateReplacementAssignment(getEVs(), beforeEVs,
swapOutName, swapInName),
+ VERIFICATION_TIMEOUT * 2);
// Cleanup
_participants.get(_participants.size() - 1).syncStop();
@@ -1375,8 +1391,10 @@ public class TestInstanceOperation extends ZkTestBase {
"Swap should complete");
Assert.assertTrue(_clusterVerifier.verifyByPolling());
- validateEVsCorrect(getEVs(), beforeSwapDisableEVs, swapOutToSwapIn,
- Collections.emptySet(), ImmutableSet.of(swapInName));
+ // See testNodeSwapBasic: after completion the exact pre-swap mirror is
not guaranteed, only a
+ // healthy replacement assignment is. Polled because the EV is rewritten
asynchronously.
+ verifier(() -> validateReplacementAssignment(getEVs(),
beforeSwapDisableEVs, swapOutName,
+ swapInName), VERIFICATION_TIMEOUT * 2);
// Cleanup
_participants.get(_participants.size() - 1).syncStop();
@@ -2059,56 +2077,25 @@ public class TestInstanceOperation extends ZkTestBase {
/**
* Waits (up to 30s) for a completed swap to be reflected in the external
views: the swap-out
- * instance no longer holds LEADER where it originally did, and the swap-in
instance has taken
- * over every partition the swap-out instance originally hosted. Only logs a
warning on timeout;
- * callers are expected to run their own definitive validations afterwards.
+ * instance holds no replica anymore and the swap-in instance hosts at least
one. Only logs a
+ * warning on timeout; callers are expected to run their own definitive
validations afterwards.
*/
- private void waitForSwapProcessed(Map<String, ExternalView> originalEVs,
String swapOutName,
- String swapInName) throws InterruptedException {
+ private void waitForSwapProcessed(String swapOutName, String swapInName)
+ throws InterruptedException {
long swapWaitStartTime = System.currentTimeMillis();
long maxSwapWaitTime = 30000;
boolean swapProcessed = false;
while (System.currentTimeMillis() - swapWaitStartTime < maxSwapWaitTime) {
if (_clusterVerifier.verifyByPolling()) {
Map<String, ExternalView> currentEVs = getEVs();
- boolean allCorrect = true;
- for (String resource : originalEVs.keySet()) {
- ExternalView current = currentEVs.get(resource);
- ExternalView original = originalEVs.get(resource);
- if (current == null || original == null) {
- allCorrect = false;
- break;
- }
- for (String partition : original.getPartitionSet()) {
- Map<String, String> currentStateMap =
current.getStateMap(partition);
- Map<String, String> originalStateMap =
original.getStateMap(partition);
- // The partition may be transiently missing from an EV while the
controller
- // rewrites it; keep polling instead of throwing NPE out of the
wait loop.
- if (currentStateMap == null || originalStateMap == null) {
- allCorrect = false;
- break;
- }
- // Check if swapOut is still in EV with LEADER state (should be
replaced)
- if (currentStateMap.containsKey(swapOutName)) {
- String currentState = currentStateMap.get(swapOutName);
- String originalState = originalStateMap.get(swapOutName);
- if (currentState.equals(originalState) &&
"LEADER".equals(currentState)) {
- allCorrect = false;
- break;
- }
- }
- // Check if swapIn has taken over the replicas swapOut originally
had.
- // Only partitions that were on swapOut are expected on swapIn;
requiring
- // swapIn in every partition made this wait condition
unsatisfiable.
- if (originalStateMap.containsKey(swapOutName)
- && !currentStateMap.containsKey(swapInName)) {
- allCorrect = false;
- break;
- }
- }
- if (!allCorrect) break;
- }
- if (allCorrect) {
+ // A partition may be transiently missing from an EV while the
controller rewrites it;
+ // treat a null state map as "not hosted" instead of throwing out of
the wait loop.
+ boolean swapOutGone = currentEVs.values().stream().allMatch(ev -> ev
!= null
+ && ev.getPartitionSet().stream().allMatch(partition ->
ev.getStateMap(partition) == null
+ || !ev.getStateMap(partition).containsKey(swapOutName)));
+ boolean swapInPresent = currentEVs.values().stream()
+ .anyMatch(ev -> ev != null &&
getParticipantsInEv(ev).contains(swapInName));
+ if (swapOutGone && swapInPresent) {
swapProcessed = true;
LOG.info("Swap processed and EV stabilized after {} ms",
System.currentTimeMillis() - swapWaitStartTime);
@@ -2147,11 +2134,12 @@ public class TestInstanceOperation extends ZkTestBase {
Assert.assertEquals(actual.getPartitionSet(), original.getPartitionSet(),
"Partition set should match for resource " + resource);
- IdealState is = _gSetupTool.getClusterManagementTool()
- .getResourceIdealState(CLUSTER_NAME, resource);
- StateModelDefinition stateModelDef =
_gSetupTool.getClusterManagementTool()
- .getStateModelDef(CLUSTER_NAME, is.getStateModelDefRef());
- String topState = stateModelDef.getTopState();
+ String topState = _topStateByResource.computeIfAbsent(resource, r -> {
+ IdealState is = _gSetupTool.getClusterManagementTool()
+ .getResourceIdealState(CLUSTER_NAME, r);
+ return _gSetupTool.getClusterManagementTool()
+ .getStateModelDef(CLUSTER_NAME,
is.getStateModelDefRef()).getTopState();
+ });
for (String partition : original.getPartitionSet()) {
Map<String, String> stateMap = actual.getStateMap(partition);
