diff --git a/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/JGroupsRaftAbstractTest.java b/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/JGroupsRaftAbstractTest.java index 388b58423e3e5..aba49ba251acd 100644 --- a/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/JGroupsRaftAbstractTest.java +++ b/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/JGroupsRaftAbstractTest.java @@ -19,10 +19,10 @@ import java.util.concurrent.TimeUnit; import org.apache.camel.Exchange; -import org.apache.camel.RuntimeCamelException; import org.apache.camel.test.junit6.CamelTestSupport; import org.jgroups.raft.RaftHandle; +import static org.awaitility.Awaitility.await; import static org.junit.jupiter.api.Assertions.assertNotNull; public abstract class JGroupsRaftAbstractTest extends CamelTestSupport { @@ -36,15 +36,29 @@ protected void checkHeaders(Exchange exchange) { assertNotNull(exchange.getIn().getHeader(JGroupsRaftConstants.HEADER_JGROUPSRAFT_RAFT_ID, String.class)); } - protected void waitForLeader(int attempts, RaftHandle rh, RaftHandle rh2, RaftHandle rh3) throws InterruptedException { - boolean thereIsLeader = rh.isLeader() || rh2.isLeader() || rh3.isLeader(); - while (!thereIsLeader && attempts > 0) { - thereIsLeader = rh.isLeader() || rh2.isLeader() || rh3.isLeader(); - TimeUnit.SECONDS.sleep(1); - attempts--; - } - if (attempts <= 0) { - throw new RuntimeCamelException("No leader in time!"); - } + /** + * Wait until a leader has been elected AND all given handles know who the leader is. Only connected handles are + * checked; disconnected or closed handles are skipped. Without checking leader() on every active handle, a follower + * node may not yet have discovered the leader, causing set() to throw when the REDIRECT protocol has no leader + * address to forward to. + */ + protected void waitForLeader(int attempts, RaftHandle... handles) { + await().atMost(attempts, TimeUnit.SECONDS) + .pollInterval(500, TimeUnit.MILLISECONDS) + .until(() -> { + boolean hasLeader = false; + for (RaftHandle rh : handles) { + if (!rh.channel().isConnected()) { + continue; + } + if (rh.isLeader()) { + hasLeader = true; + } + if (rh.leader() == null) { + return false; + } + } + return hasLeader; + }); } } diff --git a/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/JGroupsRaftConsumerTest.java b/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/JGroupsRaftConsumerTest.java index 0024ffa84bdfb..ad29806818f40 100644 --- a/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/JGroupsRaftConsumerTest.java +++ b/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/JGroupsRaftConsumerTest.java @@ -52,7 +52,7 @@ public void shouldReceiveChangeRoleEvents() throws Exception { JGroupsRaftEndpoint endpoint2 = getMandatoryEndpoint(CONFIGURED_ENDPOINT_URI2, JGroupsRaftEndpoint.class); JGroupsRaftEndpoint endpoint3 = getMandatoryEndpoint(CONFIGURED_ENDPOINT_URI3, JGroupsRaftEndpoint.class); - waitForLeader(5, endpoint.getResolvedRaftHandle(), endpoint2.getResolvedRaftHandle(), + waitForLeader(30, endpoint.getResolvedRaftHandle(), endpoint2.getResolvedRaftHandle(), endpoint3.getResolvedRaftHandle()); MockEndpoint mock = getMockEndpoint("mock:out"); diff --git a/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/JGroupsRaftProducerTest.java b/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/JGroupsRaftProducerTest.java index dfe5bee0ea347..183f6111cecdb 100644 --- a/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/JGroupsRaftProducerTest.java +++ b/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/JGroupsRaftProducerTest.java @@ -51,7 +51,7 @@ public void shouldSendBody() throws Exception { JGroupsRaftEndpoint endpoint2 = getMandatoryEndpoint(CONFIGURED_ENDPOINT_URI2, JGroupsRaftEndpoint.class); JGroupsRaftEndpoint endpoint3 = getMandatoryEndpoint(CONFIGURED_ENDPOINT_URI3, JGroupsRaftEndpoint.class); - waitForLeader(5, endpoint.getResolvedRaftHandle(), endpoint2.getResolvedRaftHandle(), + waitForLeader(30, endpoint.getResolvedRaftHandle(), endpoint2.getResolvedRaftHandle(), endpoint3.getResolvedRaftHandle()); Processor processor = exchange -> exchange.getIn().setBody(MESSAGE.getBytes()); diff --git a/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/cluster/JGroupsRaftClusterAbstractTest.java b/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/cluster/JGroupsRaftClusterAbstractTest.java index 0d9e0c4bc2a5c..62e0901db6d72 100644 --- a/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/cluster/JGroupsRaftClusterAbstractTest.java +++ b/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/cluster/JGroupsRaftClusterAbstractTest.java @@ -18,19 +18,47 @@ import java.util.concurrent.TimeUnit; -import org.apache.camel.RuntimeCamelException; +import org.jgroups.JChannel; import org.jgroups.raft.RaftHandle; +import static org.awaitility.Awaitility.await; + public abstract class JGroupsRaftClusterAbstractTest { - protected void waitForLeader(int attempts, RaftHandle rh, RaftHandle rh2, RaftHandle rh3) throws InterruptedException { - boolean thereIsLeader = rh.isLeader() || rh2.isLeader() || rh3.isLeader(); - while (!thereIsLeader && attempts > 0) { - thereIsLeader = rh.isLeader() || rh2.isLeader() || rh3.isLeader(); - TimeUnit.SECONDS.sleep(1); - attempts--; - } - if (attempts <= 0) { - throw new RuntimeCamelException("No leader in time!"); - } + + /** + * Wait until a leader has been elected AND all given handles know who the leader is. Only connected handles are + * checked; disconnected or closed handles are skipped. Without checking leader() on every active handle, a follower + * node may not yet have discovered the leader, causing set() to throw when the REDIRECT protocol has no leader + * address to forward to. + */ + protected void waitForLeader(int attempts, RaftHandle... handles) { + await().atMost(attempts, TimeUnit.SECONDS) + .pollInterval(500, TimeUnit.MILLISECONDS) + .until(() -> { + boolean hasLeader = false; + for (RaftHandle rh : handles) { + if (!rh.channel().isConnected()) { + continue; + } + if (rh.isLeader()) { + hasLeader = true; + } + if (rh.leader() == null) { + return false; + } + } + return hasLeader; + }); + } + + /** + * Wait until the given channel's view has exactly the expected number of members. Use this after closing a channel + * to ensure the remaining nodes have processed the LEAVE before creating new channels with the same member name. + */ + protected void waitForViewSize(JChannel channel, int expectedSize, int timeoutSeconds) { + await().atMost(timeoutSeconds, TimeUnit.SECONDS) + .pollInterval(500, TimeUnit.MILLISECONDS) + .until(() -> channel.isConnected() && channel.getView() != null + && channel.getView().size() == expectedSize); } } diff --git a/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/cluster/JGroupsRaftClusteredRoutePolicyTest.java b/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/cluster/JGroupsRaftClusteredRoutePolicyTest.java index f9be1fb98aadd..48e2d9a88800f 100644 --- a/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/cluster/JGroupsRaftClusteredRoutePolicyTest.java +++ b/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/cluster/JGroupsRaftClusteredRoutePolicyTest.java @@ -67,16 +67,21 @@ public void test() throws Exception { assertEquals(1, countActiveFromEndpoints(lcc, rn)); contextA.stop(); - waitForLeader(50, handleA, handleB, handleC); + // Ensure channel A is fully closed before checking for a new leader + chA.close(); + waitForLeader(50, handleB, handleC); assertEquals(1, countActiveFromEndpoints(lcc, rn)); contextB.stop(); - // NOTE: to be closed by component lifecycle. + // Ensure channel B is fully closed before creating a new channel with the same member name + chB.close(); + waitForViewSize(chC, 1, 30); chA = new JChannel("raftABC.xml").name("A"); handleA = new RaftHandle(chA, new NopStateMachine()).raftId("A"); contextA = createContext("A", handleA); + lcc.set(0, contextA); contextA.start(); - waitForLeader(50, handleA, handleB, handleC); + waitForLeader(50, handleA, handleC); assertEquals(1, countActiveFromEndpoints(lcc, rn)); } diff --git a/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/cluster/JGroupsRaftMasterTest.java b/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/cluster/JGroupsRaftMasterTest.java index 6a63b649d2248..5ca1ac5f1567c 100644 --- a/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/cluster/JGroupsRaftMasterTest.java +++ b/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/cluster/JGroupsRaftMasterTest.java @@ -71,18 +71,20 @@ public void test() throws Exception { Awaitility.await().untilAsserted(() -> assertEquals(1, countActiveFromEndpoints(lcc, rn))); contextA.stop(); - waitForLeader(50, handleA, handleB, handleC); + chA.close(); + waitForLeader(50, handleB, handleC); Awaitility.await().untilAsserted(() -> assertEquals(1, countActiveFromEndpoints(lcc, rn))); contextB.stop(); - JGroupsRaftClusterService service = new JGroupsRaftClusterService(); - service.setId("A"); - service.setRaftId("A"); - service.setRaftHandle(handleA); - service.setJgroupsClusterName("JGroupsRaftMasterTest"); - contextA.addService(service); + chB.close(); + waitForViewSize(chC, 1, 30); + // Create a completely new channel, handle and context for A + chA = new JChannel("raftABC.xml").name("A"); + handleA = new RaftHandle(chA, new NopStateMachine()).raftId("A"); + contextA = createContext("A", handleA); + lcc.set(0, contextA); contextA.start(); - waitForLeader(50, handleA, handleB, handleC); + waitForLeader(50, handleA, handleC); Awaitility.await().untilAsserted(() -> assertEquals(1, countActiveFromEndpoints(lcc, rn))); }