This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 7b58ce0cdaf6 CAMEL-25488: camel-jgroups-raft - tests close their
channels and report the state of the nodes when no leader is elected (#28243)
7b58ce0cdaf6 is described below
commit 7b58ce0cdaf659e0f26597a4a8a29f8071311fce
Author: Aurélien Pupier <[email protected]>
AuthorDate: Fri Oct 23 14:27:31 2026 +0200
CAMEL-25488: camel-jgroups-raft - tests close their channels and report the
state of the nodes when no leader is elected (#28243)
The cluster tests never closed their channels when they failed. The
channels stay
registered in the static routing table of SHARED_LOOPBACK for the whole
JVM, so
every re-run of the test failed on joining with "join of A rejected as
it would
create a view with duplicate members". The re-runs were thus always
failing with
a different error than the first attempt, and hid the original failure.
Track the channels created by the tests and close them (and stop the
contexts)
after each test, so that a re-run is a real re-run.
Also, when waiting for a leader times out, report for each node whether
it is
connected, its leader, its term and its view, to find out why the
election stalls.
Co-authored-by: IBM Bob 2.2.1
Signed-off-by: Aurélien Pupier <[email protected]>
---
.../jgroups/raft/JGroupsRaftAbstractTest.java | 37 +++++-----
.../component/jgroups/raft/RaftNodesState.java | 44 ++++++++++++
.../cluster/JGroupsRaftClusterAbstractTest.java | 81 +++++++++++++++++-----
.../JGroupsRaftClusteredRoutePolicyTest.java | 52 ++++++--------
.../raft/cluster/JGroupsRaftMasterTest.java | 58 ++++++----------
5 files changed, 173 insertions(+), 99 deletions(-)
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 aba49ba251ac..6792f939ae0b 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
@@ -20,6 +20,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.camel.Exchange;
import org.apache.camel.test.junit6.CamelTestSupport;
+import org.awaitility.core.ConditionTimeoutException;
import org.jgroups.raft.RaftHandle;
import static org.awaitility.Awaitility.await;
@@ -43,22 +44,26 @@ public abstract class JGroupsRaftAbstractTest extends
CamelTestSupport {
* 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;
+ try {
+ 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;
+ }
}
- if (rh.isLeader()) {
- hasLeader = true;
- }
- if (rh.leader() == null) {
- return false;
- }
- }
- return hasLeader;
- });
+ return hasLeader;
+ });
+ } catch (ConditionTimeoutException e) {
+ throw new ConditionTimeoutException(e.getMessage() +
System.lineSeparator() + RaftNodesState.describe(handles), e);
+ }
}
}
diff --git
a/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/RaftNodesState.java
b/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/RaftNodesState.java
new file mode 100644
index 000000000000..8cfb5ec8e3b7
--- /dev/null
+++
b/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/RaftNodesState.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.jgroups.raft;
+
+import org.jgroups.JChannel;
+import org.jgroups.raft.RaftHandle;
+
+public final class RaftNodesState {
+
+ private RaftNodesState() {
+ }
+
+ /**
+ * Describe the state of the given nodes, to find out why no leader has
been elected.
+ */
+ public static String describe(RaftHandle... handles) {
+ StringBuilder sb = new StringBuilder("State of the nodes when giving
up waiting for a leader:");
+ for (RaftHandle rh : handles) {
+ JChannel channel = rh.channel();
+ sb.append(System.lineSeparator())
+ .append(" raftId=").append(rh.raftId())
+ .append(" connected=").append(channel.isConnected())
+ .append(" isLeader=").append(rh.isLeader())
+ .append(" leader=").append(rh.leader())
+ .append(" term=").append(rh.currentTerm())
+ .append(" view=").append(channel.getView());
+ }
+ return sb.toString();
+ }
+}
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 62e0901db6d7..e77ef2799ae3 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
@@ -16,15 +16,60 @@
*/
package org.apache.camel.component.jgroups.raft.cluster;
+import java.util.ArrayList;
+import java.util.List;
import java.util.concurrent.TimeUnit;
+import org.apache.camel.CamelContext;
+import org.apache.camel.component.jgroups.raft.RaftNodesState;
+import org.awaitility.core.ConditionTimeoutException;
import org.jgroups.JChannel;
import org.jgroups.raft.RaftHandle;
+import org.jgroups.raft.StateMachine;
+import org.junit.jupiter.api.AfterEach;
import static org.awaitility.Awaitility.await;
public abstract class JGroupsRaftClusterAbstractTest {
+ protected final List<CamelContext> contexts = new ArrayList<>();
+ protected final List<String> routeNames = new ArrayList<>();
+ private final List<RaftHandle> openHandles = new ArrayList<>();
+
+ /**
+ * The channels are registered in the static routing table of
SHARED_LOOPBACK, so a channel left open by a failed
+ * test would make a rerun of the test fail on joining with the same
member names ("duplicate members"), and would
+ * hide the reason of the first failure.
+ */
+ @AfterEach
+ void closeAll() {
+ for (CamelContext context : contexts) {
+ try {
+ context.stop();
+ } catch (Exception e) {
+ // best effort cleanup
+ }
+ }
+ for (RaftHandle handle : openHandles) {
+ JChannel channel = handle.channel();
+ if (channel != null) {
+ channel.close();
+ }
+ }
+ contexts.clear();
+ routeNames.clear();
+ openHandles.clear();
+ }
+
+ /**
+ * Create the Raft handle of a node and keep track of it, so that its
channel is closed at the end of the test.
+ */
+ protected RaftHandle createHandle(String id, StateMachine stateMachine)
throws Exception {
+ RaftHandle handle = new RaftHandle(new
JChannel("raftABC.xml").name(id), stateMachine).raftId(id);
+ openHandles.add(handle);
+ return handle;
+ }
+
/**
* 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
@@ -32,23 +77,27 @@ public abstract class JGroupsRaftClusterAbstractTest {
* 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;
+ try {
+ 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;
- });
+ return hasLeader;
+ });
+ } catch (ConditionTimeoutException e) {
+ throw new ConditionTimeoutException(e.getMessage() +
System.lineSeparator() + RaftNodesState.describe(handles), e);
+ }
}
/**
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 f91b85872d1e..4acf48192d21 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
@@ -16,7 +16,7 @@
*/
package org.apache.camel.component.jgroups.raft.cluster;
-import java.util.ArrayList;
+import java.util.List;
import java.util.concurrent.TimeUnit;
import org.apache.camel.CamelContext;
@@ -25,7 +25,6 @@ import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.jgroups.raft.utils.NopStateMachine;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.impl.cluster.ClusteredRoutePolicy;
-import org.jgroups.JChannel;
import org.jgroups.raft.RaftHandle;
import org.junit.jupiter.api.Test;
@@ -34,33 +33,24 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
public class JGroupsRaftClusteredRoutePolicyTest extends
JGroupsRaftClusterAbstractTest {
- private ArrayList<CamelContext> lcc = new ArrayList<>();
- private ArrayList<String> rn = new ArrayList<>();
-
@Test
@SuppressWarnings("resource")
public void test() throws Exception {
- // NOTE: to be closed by component lifecycle.
- JChannel chA = new JChannel("raftABC.xml").name("A");
- RaftHandle handleA = new RaftHandle(chA, new
NopStateMachine()).raftId("A");
+ RaftHandle handleA = createHandle("A", new NopStateMachine());
CamelContext contextA = createContext("A", handleA);
- // NOTE: to be closed by component lifecycle.
- JChannel chB = new JChannel("raftABC.xml").name("B");
- RaftHandle handleB = new RaftHandle(chB, new
NopStateMachine()).raftId("B");
+ RaftHandle handleB = createHandle("B", new NopStateMachine());
CamelContext contextB = createContext("B", handleB);
- // NOTE: to be closed by component lifecycle.
- JChannel chC = new JChannel("raftABC.xml").name("C");
- RaftHandle handleC = new RaftHandle(chC, new
NopStateMachine()).raftId("C");
+ RaftHandle handleC = createHandle("C", new NopStateMachine());
CamelContext contextC = createContext("C", handleC);
- lcc.add(contextA);
- rn.add("route-A");
- lcc.add(contextB);
- rn.add("route-B");
- lcc.add(contextC);
- rn.add("route-C");
+ contexts.add(contextA);
+ routeNames.add("route-A");
+ contexts.add(contextB);
+ routeNames.add("route-B");
+ contexts.add(contextC);
+ routeNames.add("route-C");
contextA.start();
contextB.start();
@@ -70,18 +60,17 @@ public class JGroupsRaftClusteredRoutePolicyTest extends
JGroupsRaftClusterAbstr
contextA.stop();
// Ensure channel A is fully closed before checking for a new leader
- chA.close();
+ handleA.channel().close();
waitForLeader(50, handleB, handleC);
awaitOneActiveRoute();
contextB.stop();
// 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");
+ handleB.channel().close();
+ waitForViewSize(handleC.channel(), 1, 30);
+ handleA = createHandle("A", new NopStateMachine());
contextA = createContext("A", handleA);
- lcc.set(0, contextA);
+ contexts.set(0, contextA);
contextA.start();
waitForLeader(50, handleA, handleC);
awaitOneActiveRoute();
@@ -113,17 +102,18 @@ public class JGroupsRaftClusteredRoutePolicyTest extends
JGroupsRaftClusterAbstr
private void awaitOneActiveRoute() {
// the policy starts and stops the routes on its own thread, shortly
after the leadership change
- await().atMost(10, TimeUnit.SECONDS).untilAsserted(() ->
assertEquals(1, countActiveFromEndpoints(lcc, rn)));
+ await().atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() -> assertEquals(1,
countActiveFromEndpoints(contexts, routeNames)));
}
- private int countActiveFromEndpoints(ArrayList<CamelContext> lcc,
ArrayList<String> rn) {
+ private int countActiveFromEndpoints(List<CamelContext> contexts,
List<String> routeNames) {
int result = 0;
- if (lcc.size() != rn.size()) {
+ if (contexts.size() != routeNames.size()) {
throw new UnsupportedOperationException(
"CamelContext list and Route ids list must have the same
number of elements!");
}
- for (int i = 0; i < lcc.size(); i++) {
- ServiceStatus status =
lcc.get(i).getRouteController().getRouteStatus(rn.get(i));
+ for (int i = 0; i < contexts.size(); i++) {
+ ServiceStatus status =
contexts.get(i).getRouteController().getRouteStatus(routeNames.get(i));
if (ServiceStatus.Starting.equals(status) ||
ServiceStatus.Started.equals(status)) {
result++;
}
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 5ca1ac5f1567..1a9637880783 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
@@ -17,7 +17,7 @@
package org.apache.camel.component.jgroups.raft.cluster;
import java.lang.reflect.Field;
-import java.util.ArrayList;
+import java.util.List;
import org.apache.camel.CamelContext;
import org.apache.camel.Route;
@@ -26,7 +26,6 @@ import
org.apache.camel.component.jgroups.raft.utils.NopStateMachine;
import org.apache.camel.component.master.MasterConsumer;
import org.apache.camel.impl.DefaultCamelContext;
import org.awaitility.Awaitility;
-import org.jgroups.JChannel;
import org.jgroups.raft.RaftHandle;
import org.junit.jupiter.api.Test;
@@ -34,58 +33,45 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
public class JGroupsRaftMasterTest extends JGroupsRaftClusterAbstractTest {
- private ArrayList<CamelContext> lcc = new ArrayList<>();
- private ArrayList<String> rn = new ArrayList<>();
-
@Test
public void test() throws Exception {
- @SuppressWarnings("resource")
- // NOTE: to be closed by component lifecycle.
- JChannel chA = new JChannel("raftABC.xml").name("A");
- RaftHandle handleA = new RaftHandle(chA, new
NopStateMachine()).raftId("A");
+ RaftHandle handleA = createHandle("A", new NopStateMachine());
CamelContext contextA = createContext("A", handleA);
- @SuppressWarnings("resource")
- // NOTE: to be closed by component lifecycle.
- JChannel chB = new JChannel("raftABC.xml").name("B");
- RaftHandle handleB = new RaftHandle(chB, new
NopStateMachine()).raftId("B");
+ RaftHandle handleB = createHandle("B", new NopStateMachine());
CamelContext contextB = createContext("B", handleB);
- @SuppressWarnings("resource")
- // NOTE: to be closed by component lifecycle.
- JChannel chC = new JChannel("raftABC.xml").name("C");
- RaftHandle handleC = new RaftHandle(chC, new
NopStateMachine()).raftId("C");
+ RaftHandle handleC = createHandle("C", new NopStateMachine());
CamelContext contextC = createContext("C", handleC);
- lcc.add(contextA);
- rn.add("route-A");
- lcc.add(contextB);
- rn.add("route-B");
- lcc.add(contextC);
- rn.add("route-C");
+ contexts.add(contextA);
+ routeNames.add("route-A");
+ contexts.add(contextB);
+ routeNames.add("route-B");
+ contexts.add(contextC);
+ routeNames.add("route-C");
contextA.start();
contextB.start();
contextC.start();
waitForLeader(50, handleA, handleB, handleC);
- Awaitility.await().untilAsserted(() -> assertEquals(1,
countActiveFromEndpoints(lcc, rn)));
+ Awaitility.await().untilAsserted(() -> assertEquals(1,
countActiveFromEndpoints(contexts, routeNames)));
contextA.stop();
- chA.close();
+ handleA.channel().close();
waitForLeader(50, handleB, handleC);
- Awaitility.await().untilAsserted(() -> assertEquals(1,
countActiveFromEndpoints(lcc, rn)));
+ Awaitility.await().untilAsserted(() -> assertEquals(1,
countActiveFromEndpoints(contexts, routeNames)));
contextB.stop();
- chB.close();
- waitForViewSize(chC, 1, 30);
+ handleB.channel().close();
+ waitForViewSize(handleC.channel(), 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");
+ handleA = createHandle("A", new NopStateMachine());
contextA = createContext("A", handleA);
- lcc.set(0, contextA);
+ contexts.set(0, contextA);
contextA.start();
waitForLeader(50, handleA, handleC);
- Awaitility.await().untilAsserted(() -> assertEquals(1,
countActiveFromEndpoints(lcc, rn)));
+ Awaitility.await().untilAsserted(() -> assertEquals(1,
countActiveFromEndpoints(contexts, routeNames)));
}
private CamelContext createContext(String id, RaftHandle rh) throws
Exception {
@@ -111,15 +97,15 @@ public class JGroupsRaftMasterTest extends
JGroupsRaftClusterAbstractTest {
return context;
}
- private int countActiveFromEndpoints(ArrayList<CamelContext> lcc,
ArrayList<String> rn)
+ private int countActiveFromEndpoints(List<CamelContext> contexts,
List<String> routeNames)
throws NoSuchFieldException, IllegalAccessException {
int result = 0;
- if (lcc.size() != rn.size()) {
+ if (contexts.size() != routeNames.size()) {
throw new UnsupportedOperationException(
"CamelContext list and Route ids list must have the same
number of elements!");
}
- for (int i = 0; i < lcc.size(); i++) {
- Route route = lcc.get(i).getRoute(rn.get(i));
+ for (int i = 0; i < contexts.size(); i++) {
+ Route route = contexts.get(i).getRoute(routeNames.get(i));
if (route != null) {
MasterConsumer mc = (MasterConsumer) route.getConsumer();
Field delegatedConsumer =
mc.getClass().getDeclaredField("delegatedConsumer");
