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 1bc679e2e Another round of flaky test fixes
1bc679e2e is described below
commit 1bc679e2e00661199b1ed73b6ed04a7cf3c3d06f
Author: Junkai Xue <[email protected]>
AuthorDate: Mon Oct 19 13:41:53 2026 -0700
Another round of flaky test fixes
---
.../rebalancer/TestInstanceOperation.java | 125 ++++++++++++---------
.../apache/helix/manager/zk/TestHandleSession.java | 106 ++++++++++-------
.../messaging/handling/TestHelixTaskExecutor.java | 10 +-
3 files changed, 144 insertions(+), 97 deletions(-)
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 e0c14cae5..e06c3a947 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
@@ -1030,7 +1030,6 @@ public class TestInstanceOperation extends ZkTestBase {
removeOfflineOrInactiveInstances();
Assert.assertTrue(_clusterVerifier.verifyByPolling());
- Map<String, String> swapOutToSwapIn = new HashMap<>();
Map<String, ExternalView> originalEVs = getEVs();
// Enter maintenance mode
@@ -1058,7 +1057,6 @@ public class TestInstanceOperation extends ZkTestBase {
// Add new instance with ENABLE
String swapInName = PARTICIPANT_PREFIX + "_" + _nextStartPort;
LOG.info("Adding new instance: {}", swapInName);
- swapOutToSwapIn.put(swapOutName, swapInName);
addParticipant(swapInName, swapOutConfig.getLogicalId(LOGICAL_ID),
swapOutConfig.getDomainAsMap().get(ZONE),
InstanceConstants.InstanceOperation.ENABLE, -1);
@@ -1070,56 +1068,17 @@ public class TestInstanceOperation extends ZkTestBase {
Assert.assertTrue(_clusterVerifier.verifyByPolling());
- // Ensure swap is fully processed after maintenance mode ends
- // During maintenance mode, EV updates are deferred. After exiting, we
need to wait
- // for the controller to compute and propagate the new ideal state
assignment.
- long swapWaitStartTime = System.currentTimeMillis();
- long maxSwapWaitTime = 30000;
- boolean swapProcessed = false;
- while (System.currentTimeMillis() - swapWaitStartTime < maxSwapWaitTime) {
- Map<String, ExternalView> currentEVs = getEVs();
- boolean allReplaced = true;
- for (String resource : originalEVs.keySet()) {
- ExternalView current = currentEVs.get(resource);
- ExternalView original = originalEVs.get(resource);
- if (current == null || original == null) {
- allReplaced = false;
- break;
- }
- for (String partition : original.getPartitionSet()) {
- Map<String, String> currentStateMap = current.getStateMap(partition);
- Map<String, String> originalStateMap =
original.getStateMap(partition);
- // Check if swapOut is still in EV with higher state than expected
- if (currentStateMap.containsKey(swapOutName)) {
- String currentState = currentStateMap.get(swapOutName);
- String originalState = originalStateMap.get(swapOutName);
- if (currentState.equals(originalState) &&
"LEADER".equals(currentState)) {
- // swapOut still has its original (higher) state - not yet
replaced
- allReplaced = false;
- break;
- }
- }
- }
- if (!allReplaced) break;
- }
- if (allReplaced) {
- swapProcessed = true;
- LOG.info("Swap processed after {} ms", System.currentTimeMillis() -
swapWaitStartTime);
- break;
- }
- Thread.sleep(500);
- }
- if (!swapProcessed) {
- LOG.warn("Swap may not have been fully processed within {} ms",
maxSwapWaitTime);
- }
-
- // Verify swap-in has the same partitions swap-out had.
- // Use verifier() instead of TestHelper.verify directly:
validateEVsCorrect asserts
- // internally, and an AssertionError thrown from the lambda escapes
TestHelper.verify's
- // retry loop, failing the test on the first transient mismatch instead of
polling
- // until the cluster converges.
- verifier(() -> validateEVsCorrect(getEVs(), originalEVs, swapOutToSwapIn,
- Collections.emptySet(), ImmutableSet.of(swapInName)),
VERIFICATION_TIMEOUT * 2);
+ // The new instance is added with ENABLE, not SWAP_IN, so the controller
does not pair it with
+ // the evacuated instance: it is just another assignable node that happens
to share the
+ // evacuated node's logicalId (and therefore its position in the CRUSH
topology). The product
+ // only guarantees that the evacuated instance ends up with no replicas
while every partition
+ // keeps a full, healthy replica set, and that the new instance picks up
load. It does not
+ // guarantee an exact in-place copy of the evacuated instance's
assignment, since the even
+ // distribution rounds and preference-list order can legitimately differ
once the node set
+ // changes (observed on CI as leaders moving on unrelated partitions).
Only the SWAP_IN flow
+ // (see the testSwap* tests) mirrors the swap-out assignment onto the
swap-in instance.
+ verifier(() -> validateEvacuateReplacement(getEVs(), originalEVs,
swapOutName, swapInName),
+ VERIFICATION_TIMEOUT * 2);
// Verify evacuation finished - use verifier to handle timing variations
boolean evacFinished = TestHelper.verify(() ->
@@ -1173,10 +1132,10 @@ public class TestInstanceOperation extends ZkTestBase {
Assert.assertTrue(_clusterVerifier.verifyByPolling());
- // Final validation. verifier() tolerates AssertionError from
validateEVsCorrect and
- // keeps polling until the cluster converges.
- verifier(() -> validateEVsCorrect(getEVs(), originalEVs, swapOutToSwapIn,
- Collections.emptySet(), ImmutableSet.of(swapInName)),
VERIFICATION_TIMEOUT * 2);
+ // Final validation: the evacuated instance (now UNKNOWN) still holds
nothing, every partition
+ // is fully replicated and the new instance keeps its load.
+ verifier(() -> validateEvacuateReplacement(getEVs(), originalEVs,
swapOutName, swapInName),
+ VERIFICATION_TIMEOUT * 2);
LOG.info("END testEvacuateWithAddDuringMaintenanceMode");
}
@@ -2157,6 +2116,60 @@ public class TestInstanceOperation extends ZkTestBase {
}
}
+ /**
+ * Validates the external views after an instance was evacuated and a
replacement instance was
+ * added with ENABLE (not SWAP_IN). Unlike {@link #validateEVsCorrect}, this
does not expect an
+ * exact in-place copy of the evacuated instance's assignment, because an
ENABLE instance is not
+ * paired with the evacuated one by the controller and the rebalancer may
legitimately pick a
+ * different (still valid) assignment and preference-list order. It checks
what EVACUATE does
+ * guarantee: the evacuated instance holds no replicas, every original
partition still has a
+ * full set of healthy replicas with exactly one top-state replica, and the
replacement instance
+ * has taken on load.
+ */
+ private boolean validateEvacuateReplacement(Map<String, ExternalView>
actuals,
+ Map<String, ExternalView> originals, String evacuatedInstance, String
replacementInstance) {
+ Assert.assertEquals(actuals.keySet(), originals.keySet(), "Resource key
sets should match");
+
+ boolean replacementHasReplicas = false;
+ for (String resource : originals.keySet()) {
+ ExternalView actual = actuals.get(resource);
+ ExternalView original = originals.get(resource);
+ Assert.assertNotNull(actual, "Missing external view for resource " +
resource);
+ 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();
+
+ for (String partition : original.getPartitionSet()) {
+ Map<String, String> stateMap = actual.getStateMap(partition);
+ Assert.assertNotNull(stateMap,
+ String.format("Partition %s in resource %s has no state map",
partition, resource));
+ Assert.assertFalse(stateMap.containsKey(evacuatedInstance),
+ String.format("Evacuated instance %s still hosts partition %s in
resource %s: %s",
+ evacuatedInstance, partition, resource, stateMap));
+ Assert.assertEquals(stateMap.size(), REPLICA,
+ String.format("Partition %s in resource %s should have %d
replicas: %s", partition,
+ resource, REPLICA, stateMap));
+
Assert.assertTrue(stateMap.values().stream().allMatch(ACCEPTABLE_STATE_SET::contains),
+ String.format("Partition %s in resource %s has unhealthy replicas:
%s", partition,
+ resource, stateMap));
+
Assert.assertEquals(stateMap.values().stream().filter(topState::equals).count(),
1L,
+ String.format("Partition %s in resource %s should have exactly one
%s replica: %s",
+ partition, resource, topState, stateMap));
+ if (stateMap.containsKey(replacementInstance)) {
+ replacementHasReplicas = true;
+ }
+ }
+ }
+ Assert.assertTrue(replacementHasReplicas,
+ "Replacement instance " + replacementInstance + " should host some
replicas");
+ return true;
+ }
+
private boolean validateEVsCorrect(Map<String, ExternalView> actuals,
Map<String, ExternalView> originals, Map<String, String> swapOutToSwapIn,
Set<String> inFlightSwapIn, Set<String> completedSwapIn) {
diff --git
a/helix-core/src/test/java/org/apache/helix/manager/zk/TestHandleSession.java
b/helix-core/src/test/java/org/apache/helix/manager/zk/TestHandleSession.java
index 2c0838a66..8e15bfa17 100644
---
a/helix-core/src/test/java/org/apache/helix/manager/zk/TestHandleSession.java
+++
b/helix-core/src/test/java/org/apache/helix/manager/zk/TestHandleSession.java
@@ -26,6 +26,7 @@ import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.helix.AccessOption;
import org.apache.helix.HelixException;
@@ -50,6 +51,10 @@ import org.testng.annotations.Test;
public class TestHandleSession extends ZkTestBase {
+ // Generous polling window for session re-establishment, handler reset and
live instance
+ // creation. These are fast normally but take seconds on slow CI machines.
+ private static final long SESSION_WAIT_DURATION = 30 * 1000L;
+
private static final String _className = TestHelper.getTestClassName();
@Test
@@ -397,16 +402,27 @@ public class TestHandleSession extends ZkTestBase {
final int handlerCount = manager.getHandlers().size();
final long originalCreationTime = liveInstance.getStat().getCreationTime();
- final CountDownLatch mainThreadBlocker = new CountDownLatch(1);
+ // Released by the helper thread once expired S1 has completed resetting
handlers and S2 has
+ // started handling its new session, so the main thread can verify S1 did
not create the live
+ // instance.
+ final CountDownLatch s1ResetDoneBlocker = new CountDownLatch(1);
+ // Released by the main thread to let the helper thread proceed S2.
final CountDownLatch helperThreadBlocker = new CountDownLatch(1);
+ // Released by the helper thread once it has let S2 proceed.
+ final CountDownLatch s2ProceededBlocker = new CountDownLatch(1);
+ // Failures in the helper thread are re-thrown on the main thread. They
used to be swallowed,
+ // so a slow session establishment silently turned into a confusing
failure of the final
+ // assertions.
+ final AtomicReference<Throwable> helperFailure = new AtomicReference<>();
// Helper thread to help verify zk session states, async expire S1,
proceed S1 to reset
// handlers and release main thread to verify results.
new Thread(() -> {
try {
// Wait for new session S1 is established and starting to reset
handlers.
- TestHelper.verify(() ->
!(manager.getSessionId().equals(originalSessionId))
- && manager.getResetHandlersStartTime() >
initResetHandlersStartTime, 5000L);
+ Assert.assertTrue(TestHelper.verify(() ->
!(manager.getSessionId().equals(originalSessionId))
+ && manager.getResetHandlersStartTime() >
initResetHandlersStartTime,
+ SESSION_WAIT_DURATION), "S1 did not start resetting handlers in
time");
// S1's info.
final String lastSessionId = manager.getSessionId();
@@ -417,50 +433,53 @@ public class TestHandleSession extends ZkTestBase {
// 4. S1 expired, new session S2 created
ZkTestHelper.asyncExpireSession(manager.getZkClient());
- // Wait and verify the new session S2 is established.
- TestHelper.verify(() ->
!((ZKUtil.toHexSessionId(manager.getZkClient().getSessionId()))
- .equals(lastSessionId)), 5000L);
- } catch (Exception ignored) {
- // Ignored.
+ // Wait and verify the zookeeper is closed. The zk client cannot
reconnect while the
+ // event lock is held, so S2 only gets established after the unlock
below.
+ Assert.assertTrue(TestHelper.verify(
+ () -> !((ZkClient)
manager.getZkClient()).getConnection().getZookeeperState()
+ .isAlive(), SESSION_WAIT_DURATION), "S1 was not expired in
time");
} finally {
// Unlock to start processing event again.
((ZkClient) manager.getZkClient()).getEventLock().unlock();
}
+ // Wait and verify the new session S2 is established.
+ Assert.assertTrue(TestHelper.verify(() -> {
+ try {
+ final String sessionId =
ZKUtil.toHexSessionId(manager.getZkClient().getSessionId());
+ return !"0".equals(sessionId) && !sessionId.equals(lastSessionId);
+ } catch (HelixException ex) {
+ // The zk client may still be re-connecting.
+ return false;
+ }
+ }, SESSION_WAIT_DURATION), "S2 was not established in time");
+
// Proceed S1 to complete reset handlers and try to create live
instance.
manager.proceedResetHandlers();
// Wait for S2 to handle new session.
- TestHelper.verify(() -> !(manager.getSessionId().equals(lastSessionId))
- && manager.getResetHandlersStartTime() >
lastResetHandlersStartTime, 5000L);
-
+ Assert.assertTrue(TestHelper.verify(() ->
!(manager.getSessionId().equals(lastSessionId))
+ && manager.getResetHandlersStartTime() >
lastResetHandlersStartTime,
+ SESSION_WAIT_DURATION), "S2 did not start resetting handlers in
time");
+ } catch (Throwable t) {
+ helperFailure.compareAndSet(null, t);
+ } finally {
// Notify main thread to verify result: expired S1 should not create
live instance.
- mainThreadBlocker.countDown();
+ s1ResetDoneBlocker.countDown();
+ }
+ try {
// Wait for notification from main thread to proceed S2.
helperThreadBlocker.await();
// Proceed S2.
// 6. S2 is valid and creates live instance.
manager.proceedResetHandlers();
-
- final String latestSessionId =
ZKUtil.toHexSessionId(manager.getZkClient().getSessionId());
-
- TestHelper.verify(() -> {
- // Newly created live instance should be created by the latest
session
- // and have a new creation time.
- LiveInstance newLiveInstance =
- accessor.getProperty(keyBuilder.liveInstance(instanceName));
- return newLiveInstance != null
- && newLiveInstance.getStat().getCreationTime() !=
originalCreationTime
- && newLiveInstance.getEphemeralOwner().equals(latestSessionId);
- }, 2000L);
- } catch (Exception ignored) {
- // Ignored.
+ } catch (Throwable t) {
+ helperFailure.compareAndSet(null, t);
+ } finally {
+ s2ProceededBlocker.countDown();
}
-
- // Notify the main thread that live instance is already created by
session S2.
- mainThreadBlocker.countDown();
}).start();
// Lock zk event processing to simulate a long backlog queue.
@@ -473,14 +492,15 @@ public class TestHandleSession extends ZkTestBase {
// Wait and verify the zookeeper is closed.
Assert.assertTrue(TestHelper.verify(
() -> !((ZkClient)
manager.getZkClient()).getConnection().getZookeeperState().isAlive(),
- 3000L));
+ SESSION_WAIT_DURATION));
} finally {
// Unlock to start processing event again.
((ZkClient) manager.getZkClient()).getEventLock().unlock();
}
// Wait for S1 completing resetting handlers.
- mainThreadBlocker.await();
+ s1ResetDoneBlocker.await();
+ Assert.assertNull(helperFailure.get(), "Helper thread failed: " +
helperFailure.get());
// 5. S1 completes resetting handlers, live instance should not be created
by the expired S1
Assert.assertNull(accessor.getProperty(keyBuilder.liveInstance(instanceName)));
@@ -488,13 +508,17 @@ public class TestHandleSession extends ZkTestBase {
// Notify helper thread to proceed S2.
helperThreadBlocker.countDown();
- // Wait for live instance being created by the new session S2.
- mainThreadBlocker.await();
+ // Wait for the helper thread to let S2 proceed. (The previous version
re-used an already
+ // released latch here, so it did not wait at all and then only gave S2 a
2s window.)
+ s2ProceededBlocker.await();
+ Assert.assertNull(helperFailure.get(), "Helper thread failed: " +
helperFailure.get());
- // From now on, the live instance is already created by S2.
+ // From now on, the live instance is going to be created by S2.
// The latest(the final new one S2) session id that is valid.
final String latestSessionId =
ZKUtil.toHexSessionId(manager.getZkClient().getSessionId());
+ // Finishing the handler reset and writing the live instance to ZK can
take seconds on a slow
+ // machine, so poll generously.
Assert.assertTrue(TestHelper.verify(() -> {
// Newly created live instance should be created by the latest session
// and have a new creation time.
@@ -502,12 +526,14 @@ public class TestHandleSession extends ZkTestBase {
return newLiveInstance != null
&& newLiveInstance.getStat().getCreationTime() !=
originalCreationTime && newLiveInstance
.getEphemeralOwner().equals(latestSessionId);
- }, 2000L));
+ }, SESSION_WAIT_DURATION));
// All the callback handlers shall be recovered.
- Assert.assertTrue(TestHelper.verify(() -> manager.getHandlers().size() ==
handlerCount, 1000L));
- Assert.assertTrue(TestHelper
- .verify(() ->
manager.getHandlers().stream().allMatch(CallbackHandler::isReady), 3000L));
+ Assert.assertTrue(TestHelper.verify(() -> manager.getHandlers().size() ==
handlerCount,
+ SESSION_WAIT_DURATION));
+ Assert.assertTrue(TestHelper.verify(
+ () ->
manager.getHandlers().stream().allMatch(CallbackHandler::isReady),
+ SESSION_WAIT_DURATION));
// Clean up.
manager.disconnect();
@@ -644,7 +670,9 @@ public class TestHandleSession extends ZkTestBase {
resetHandlersStartTime = System.currentTimeMillis();
try {
if (!isShutdown) {
- resetHandlersSemaphore.tryAcquire(20L, TimeUnit.SECONDS);
+ // Safety net only: the test releases the semaphore explicitly. Must
exceed the
+ // helper thread's polling windows so a slow machine cannot break
the premise.
+ resetHandlersSemaphore.tryAcquire(2 * SESSION_WAIT_DURATION,
TimeUnit.MILLISECONDS);
}
} catch (InterruptedException ignored) {
// Ignore the exception.
diff --git
a/helix-core/src/test/java/org/apache/helix/messaging/handling/TestHelixTaskExecutor.java
b/helix-core/src/test/java/org/apache/helix/messaging/handling/TestHelixTaskExecutor.java
index a15620048..97f28b003 100644
---
a/helix-core/src/test/java/org/apache/helix/messaging/handling/TestHelixTaskExecutor.java
+++
b/helix-core/src/test/java/org/apache/helix/messaging/handling/TestHelixTaskExecutor.java
@@ -906,7 +906,11 @@ public class TestHelixTaskExecutor {
LOG.info("Created message {} with timeout {} ms", msg.getId(),
msg.getExecutionTimeout());
}
changeContext.setChangeType(HelixConstants.ChangeType.MESSAGE);
- executor.onMessage("someInstance", msgList, changeContext);
+ // Pass a copy: HelixTaskExecutor.onMessage sorts the list it is given in
place, newest message
+ // first (Message.CREATE_TIME_COMPARATOR). On a slow machine the 4
messages can straddle a
+ // millisecond boundary, which would reorder msgList and make the
index-based assertions below
+ // check the wrong messages.
+ executor.onMessage("someInstance", new ArrayList<>(msgList),
changeContext);
LOG.info("Submitted {} messages to executor", nMsgs2);
// Poll until the 2 short-timeout messages have been cancelled (the
handler sets Cancelcount
@@ -993,7 +997,9 @@ public class TestHelixTaskExecutor {
msgList.add(msg);
}
changeContext.setChangeType(HelixConstants.ChangeType.MESSAGE);
- executor.onMessage("someInstance", msgList, changeContext);
+ // Pass a copy: onMessage sorts the given list in place (newest create
time first), which would
+ // reorder msgList on a slow machine and break the index-based assertions
below.
+ executor.onMessage("someInstance", new ArrayList<>(msgList),
changeContext);
// All processing, timeouts and retries happen asynchronously on the
executor threads, so
// poll until every message has reached its final state and the task map
is drained.
