From 38f77895346900090933975beee2fd62adfef2bd Mon Sep 17 00:00:00 2001 From: samlightfoot Date: Wed, 25 Feb 2026 08:42:55 +0000 Subject: [PATCH] Fix flaky InProgressSequenceCoordinationTest by increasing request_timeout and ensuring background threads are joined before test cleanup Patch by Sam Lightfoot; reviewed by Dmitry Konstantinov and Sam Tunnicliffe for CASSANDRA-21189 --- .../InProgressSequenceCoordinationTest.java | 178 +++++++++++------- 1 file changed, 109 insertions(+), 69 deletions(-) diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/InProgressSequenceCoordinationTest.java b/test/distributed/org/apache/cassandra/distributed/test/log/InProgressSequenceCoordinationTest.java index c23bd3d751..94558ac66e 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/log/InProgressSequenceCoordinationTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/log/InProgressSequenceCoordinationTest.java @@ -70,7 +70,7 @@ public class InProgressSequenceCoordinationTest extends FuzzTestBase { try (Cluster cluster = builder().withNodes(3) .appendConfig(cfg -> cfg.set("progress_barrier_timeout", "5000ms") - .set("request_timeout", "1000ms") + .set("request_timeout", "5000ms") .set("progress_barrier_backoff", "100ms") .with(Feature.NETWORK, Feature.GOSSIP)) .withTokenSupplier(TokenSupplier.evenlyDistributedTokens(4)) @@ -100,25 +100,33 @@ public class InProgressSequenceCoordinationTest extends FuzzTestBase // The StartJoin event has no prerequisite, so we should see a CONTINUING state, but execution of the // MidJoin should be BLOCKED as nodes 2 & 3 can't ack the StartJoin. Callable progressBlocked = waitForListener(newInstance, continuable(), blocked()); - new Thread(() -> newInstance.startup()).start(); - progressBlocked.call(); + Thread startupThread = new Thread(newInstance::startup); + startupThread.start(); + try + { + progressBlocked.call(); - // Remove the partition between nodes 2 & 3 and the CMS and have them catch up. - cluster.filters().reset(); - cluster.get(2).runOnInstance(() -> ClusterMetadataService.instance().processor().fetchLogAndWait()); - cluster.get(3).runOnInstance(() -> ClusterMetadataService.instance().processor().fetchLogAndWait()); + // Remove the partition between nodes 2 & 3 and the CMS and have them catch up. + cluster.filters().reset(); + cluster.get(2).runOnInstance(() -> ClusterMetadataService.instance().processor().fetchLogAndWait()); + cluster.get(3).runOnInstance(() -> ClusterMetadataService.instance().processor().fetchLogAndWait()); - // Unpause the joining node and have it retry the bootstrap sequence from MidJoin, this time the - // expectation is that it will succeed, so we can just clear out the listener. - newInstance.runOnInstance(() -> { - TestExecutionListener listener = (TestExecutionListener) InProgressSequences.listener; - listener.restorePrevious(); - listener.releaseAndRetry(); - }); + // Unpause the joining node and have it retry the bootstrap sequence from MidJoin, this time the + // expectation is that it will succeed, so we can just clear out the listener. + newInstance.runOnInstance(() -> { + TestExecutionListener listener = (TestExecutionListener) InProgressSequences.listener; + listener.restorePrevious(); + listener.releaseAndRetry(); + }); - // Wait for the cluster to all witness the finish join event. - Epoch finalEpoch = finishJoinEpoch.call(); - ClusterUtils.waitForCMSToQuiesce(cluster, finalEpoch); + // Wait for the cluster to all witness the finish join event. + Epoch finalEpoch = finishJoinEpoch.call(); + ClusterUtils.waitForCMSToQuiesce(cluster, finalEpoch); + } + finally + { + releaseAndJoin(newInstance, startupThread); + } } } @@ -128,7 +136,7 @@ public class InProgressSequenceCoordinationTest extends FuzzTestBase try (Cluster cluster = builder().withNodes(4) .withTokenSupplier(TokenSupplier.evenlyDistributedTokens(4)) .appendConfig(cfg -> cfg.set("progress_barrier_timeout", "5000ms") - .set("request_timeout", "1000ms") + .set("request_timeout", "5000ms") .set("progress_barrier_backoff", "100ms") .with(Feature.NETWORK, Feature.GOSSIP)) .withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(4, "dc0", "rack0")) @@ -151,44 +159,51 @@ public class InProgressSequenceCoordinationTest extends FuzzTestBase Callable progressBlocked = waitForListener(leavingInstance, blocked()); Thread t = new Thread(() -> leavingInstance.runOnInstance(() -> StorageService.instance.decommission(true))); t.start(); - progressBlocked.call(); - // Remove the partition between nodes 2 & 3 and the CMS and have them catch up. - cluster.filters().reset(); - cluster.get(2).runOnInstance(() -> ClusterMetadataService.instance().processor().fetchLogAndWait()); - cluster.get(3).runOnInstance(() -> ClusterMetadataService.instance().processor().fetchLogAndWait()); + try + { + progressBlocked.call(); + // Remove the partition between nodes 2 & 3 and the CMS and have them catch up. + cluster.filters().reset(); + cluster.get(2).runOnInstance(() -> ClusterMetadataService.instance().processor().fetchLogAndWait()); + cluster.get(3).runOnInstance(() -> ClusterMetadataService.instance().processor().fetchLogAndWait()); - // Now re-partition nodes 2 & 3 so that the MidLeave event cannot be submitted by node 1 as 2 & 3 won't - // receive/ack the StartLeave. - cluster.filters().allVerbs().from(1).to(2,3).drop(); - cluster.filters().verbs(TCM_FETCH_PEER_LOG_RSP.id).from(4).to(2, 3).drop(); + // Now re-partition nodes 2 & 3 so that the MidLeave event cannot be submitted by node 1 as 2 & 3 won't + // receive/ack the StartLeave. + cluster.filters().allVerbs().from(1).to(2,3).drop(); + cluster.filters().verbs(TCM_FETCH_PEER_LOG_RSP.id).from(4).to(2, 3).drop(); - // Unpause the leaving node and have it retry the StartLeave, which should now be able to proceed as 2 & 3 - // will ack the PrepareJoin. Its progress should be blocked as it comes to submit the next event, its - // MidJoin, so set a new BLOCKED expectation. - progressBlocked = waitForExistingListener(leavingInstance, blocked()); - leavingInstance.runOnInstance(() -> { - TestExecutionListener listener = (TestExecutionListener) InProgressSequences.listener; - listener.releaseAndRetry(); - }); - progressBlocked.call(); + // Unpause the leaving node and have it retry the StartLeave, which should now be able to proceed as 2 & 3 + // will ack the PrepareJoin. Its progress should be blocked as it comes to submit the next event, its + // MidJoin, so set a new BLOCKED expectation. + progressBlocked = waitForExistingListener(leavingInstance, blocked()); + leavingInstance.runOnInstance(() -> { + TestExecutionListener listener = (TestExecutionListener) InProgressSequences.listener; + listener.releaseAndRetry(); + }); + progressBlocked.call(); - // Heal the partition again and force a catch up. - cluster.filters().reset(); - cluster.get(2).runOnInstance(() -> ClusterMetadataService.instance().processor().fetchLogAndWait()); - cluster.get(3).runOnInstance(() -> ClusterMetadataService.instance().processor().fetchLogAndWait()); + // Heal the partition again and force a catch up. + cluster.filters().reset(); + cluster.get(2).runOnInstance(() -> ClusterMetadataService.instance().processor().fetchLogAndWait()); + cluster.get(3).runOnInstance(() -> ClusterMetadataService.instance().processor().fetchLogAndWait()); - // Unpause the leaving node and have it retry the MidLeave, which will now be able to proceed as 2 & 3 will - // ack the StartJoin. This time the expectation is that the remaining events will be successfully acked, so - // we can just clear out the listener. - leavingInstance.runOnInstance(() -> { - TestExecutionListener listener = (TestExecutionListener) InProgressSequences.listener; - listener.restorePrevious(); - listener.releaseAndRetry(); - }); + // Unpause the leaving node and have it retry the MidLeave, which will now be able to proceed as 2 & 3 will + // ack the StartJoin. This time the expectation is that the remaining events will be successfully acked, so + // we can just clear out the listener. + leavingInstance.runOnInstance(() -> { + TestExecutionListener listener = (TestExecutionListener) InProgressSequences.listener; + listener.restorePrevious(); + listener.releaseAndRetry(); + }); - // Wait for the cluster to all witness the finish join event. - Epoch finalEpoch = finishLeaveEpoch.call(); - ClusterUtils.waitForCMSToQuiesce(cluster, finalEpoch); + // Wait for the cluster to all witness the finish join event. + Epoch finalEpoch = finishLeaveEpoch.call(); + ClusterUtils.waitForCMSToQuiesce(cluster, finalEpoch); + } + finally + { + releaseAndJoin(leavingInstance, t); + } } } @@ -197,7 +212,7 @@ public class InProgressSequenceCoordinationTest extends FuzzTestBase { try (Cluster cluster = builder().withNodes(3) .appendConfig(cfg -> cfg.set("progress_barrier_timeout", "5000ms") - .set("request_timeout", "1000ms") + .set("request_timeout", "5000ms") .set("progress_barrier_backoff", "100ms") .with(Feature.NETWORK, Feature.GOSSIP)) .withTokenSupplier(TokenSupplier.evenlyDistributedTokens(4)) @@ -226,31 +241,39 @@ public class InProgressSequenceCoordinationTest extends FuzzTestBase // Have the joining node pause when the StartReplace event fails due to ack timeout. Callable progressBlocked = waitForListener(replacement, continuable(), blocked()); - new Thread(() -> { + Thread startupThread = new Thread(() -> { try (WithProperties replacementProps = new WithProperties()) { replacementProps.set(REPLACE_ADDRESS_FIRST_BOOT, toReplace.config().broadcastAddress().getAddress().getHostAddress()); replacement.startup(); } - }).start(); - progressBlocked.call(); - - // Remove the partition between node 2 and the CMS and have it catch up. - cluster.filters().reset(); - cluster.get(2).runOnInstance(() -> ClusterMetadataService.instance().processor().fetchLogAndWait()); - - // Unpause the joining node and have it retry the bootstrap sequence from StartReplace, this time the - // expectation is that it will succeed, so we can just clear out the listener. - replacement.runOnInstance(() -> { - TestExecutionListener listener = (TestExecutionListener) InProgressSequences.listener; - listener.restorePrevious(); - listener.releaseAndRetry(); }); + startupThread.start(); + try + { + progressBlocked.call(); - // Wait for the cluster to all witness the finish join event. - Epoch finalEpoch = finishReplaceEpoch.call(); - ClusterUtils.waitForCMSToQuiesce(cluster, finalEpoch); + // Remove the partition between node 2 and the CMS and have it catch up. + cluster.filters().reset(); + cluster.get(2).runOnInstance(() -> ClusterMetadataService.instance().processor().fetchLogAndWait()); + + // Unpause the joining node and have it retry the bootstrap sequence from StartReplace, this time the + // expectation is that it will succeed, so we can just clear out the listener. + replacement.runOnInstance(() -> { + TestExecutionListener listener = (TestExecutionListener) InProgressSequences.listener; + listener.restorePrevious(); + listener.releaseAndRetry(); + }); + + // Wait for the cluster to all witness the finish join event. + Epoch finalEpoch = finishReplaceEpoch.call(); + ClusterUtils.waitForCMSToQuiesce(cluster, finalEpoch); + } + finally + { + releaseAndJoin(replacement, startupThread); + } } } @@ -338,6 +361,16 @@ public class InProgressSequenceCoordinationTest extends FuzzTestBase return remoteCallable::call; } + // Unblock the thread if it's still waiting on the listener barrier, then join to ensure clean shutdown. + private void releaseAndJoin(IInvokableInstance instance, Thread thread) throws InterruptedException + { + instance.runOnInstance(() -> { + if (InProgressSequences.listener instanceof TestExecutionListener) + ((TestExecutionListener) InProgressSequences.listener).releaseWithoutRetry(); + }); + thread.join(); + } + public static class TestExecutionListener implements BiFunction, SequenceState, SequenceState> { volatile boolean retry = true; @@ -405,5 +438,12 @@ public class InProgressSequenceCoordinationTest extends FuzzTestBase retry = true; barrier.signal(); } + + public void releaseWithoutRetry() + { + retry = false; + if (barrier != null) + barrier.signal(); + } } }