Skip to content

CAMEL-24598: Fix flaky test JGroupsRaftProducerTest.shouldSendBody #26193

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
gnodet wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

CAMEL-24598: Fix flaky test JGroupsRaftProducerTest.shouldSendBody #26193

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 25 additions & 11 deletions ...s-raft/src/test/java/org/apache/camel/component/jgroups/raft/JGroupsRaftAbstractTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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;
});
}
}
2 changes: 1 addition & 1 deletion ...s-raft/src/test/java/org/apache/camel/component/jgroups/raft/JGroupsRaftConsumerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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");
Expand Down
2 changes: 1 addition & 1 deletion ...s-raft/src/test/java/org/apache/camel/component/jgroups/raft/JGroupsRaftProducerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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());
Expand Down
50 changes: 39 additions & 11 deletions .../java/org/apache/camel/component/jgroups/raft/cluster/JGroupsRaftClusterAbstractTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
}
11 changes: 8 additions & 3 deletions .../org/apache/camel/component/jgroups/raft/cluster/JGroupsRaftClusteredRoutePolicyTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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));
}

Expand Down
18 changes: 10 additions & 8 deletions .../src/test/java/org/apache/camel/component/jgroups/raft/cluster/JGroupsRaftMasterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)));
}

Expand Down