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 86f1e5f46277 CAMEL-25883: camel-jgroups-raft - an endpoint no longer
disconnects or closes the RaftHandle of the component while other endpoints use
it (#28425)
86f1e5f46277 is described below
commit 86f1e5f46277b70c9ab7723a87aede9be2f8ed03
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Oct 25 13:07:01 2026 +0530
CAMEL-25883: camel-jgroups-raft - an endpoint no longer disconnects or
closes the RaftHandle of the component while other endpoints use it (#28425)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../jgroups/raft/JGroupsRaftComponent.java | 90 ++++++++++++++++
.../jgroups/raft/JGroupsRaftConsumer.java | 9 +-
.../jgroups/raft/JGroupsRaftEndpoint.java | 40 ++++---
.../jgroups/raft/JGroupsRaftProducer.java | 8 +-
.../JGroupsRaftComponentHandleEndpointsTest.java | 118 +++++++++++++++++++++
.../raft/JGroupsRaftNeverStartedConsumerTest.java | 62 +++++++++++
6 files changed, 313 insertions(+), 14 deletions(-)
diff --git
a/components/camel-jgroups-raft/src/main/java/org/apache/camel/component/jgroups/raft/JGroupsRaftComponent.java
b/components/camel-jgroups-raft/src/main/java/org/apache/camel/component/jgroups/raft/JGroupsRaftComponent.java
index b83cdb53cf52..4e7e256d2c17 100644
---
a/components/camel-jgroups-raft/src/main/java/org/apache/camel/component/jgroups/raft/JGroupsRaftComponent.java
+++
b/components/camel-jgroups-raft/src/main/java/org/apache/camel/component/jgroups/raft/JGroupsRaftComponent.java
@@ -16,7 +16,14 @@
*/
package org.apache.camel.component.jgroups.raft;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.IdentityHashMap;
+import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import org.apache.camel.Endpoint;
import org.apache.camel.component.jgroups.raft.utils.NopStateMachine;
@@ -26,6 +33,8 @@ import org.apache.camel.spi.annotations.Component;
import org.apache.camel.support.DefaultComponent;
import org.jgroups.raft.RaftHandle;
import org.jgroups.raft.StateMachine;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Component providing support for JGroups-raft leader election and shared
state machine implementation
@@ -33,6 +42,8 @@ import org.jgroups.raft.StateMachine;
*/
@Component("jgroups-raft")
public class JGroupsRaftComponent extends DefaultComponent {
+ private static final Logger LOG =
LoggerFactory.getLogger(JGroupsRaftComponent.class);
+
@UriParam
@Metadata(defaultValue = "null")
private RaftHandle raftHandle;
@@ -46,6 +57,13 @@ public class JGroupsRaftComponent extends DefaultComponent {
@Metadata(defaultValue = "raft.xml")
private String channelProperties;
+ // the connected producers and consumers of each RaftHandle that the
endpoints did not create (such as the
+ // raftHandle of this component), as several endpoints use such a handle
+ private final Map<RaftHandle, Integer> sharedHandleUsers = new
IdentityHashMap<>();
+ // the handles that the endpoints did not create and connected, closed
when this component stops
+ private final Set<RaftHandle> sharedHandles =
Collections.newSetFromMap(new IdentityHashMap<>());
+ private final Lock sharedHandleLock = new ReentrantLock();
+
@Override
protected Endpoint createEndpoint(String uri, String remaining,
Map<String, Object> parameters) {
return new JGroupsRaftEndpoint(
@@ -99,5 +117,77 @@ public class JGroupsRaftComponent extends DefaultComponent {
this.channelProperties = channelProperties;
}
+ /**
+ * Connects a RaftHandle that several endpoints can use, for a producer or
a consumer of one of them.
+ */
+ void connectSharedHandle(RaftHandle sharedHandle, String clusterName)
throws Exception {
+ sharedHandleLock.lock();
+ try {
+ sharedHandle.channel().connect(clusterName);
+ sharedHandleUsers.merge(sharedHandle, 1, Integer::sum);
+ sharedHandles.add(sharedHandle);
+ } finally {
+ sharedHandleLock.unlock();
+ }
+ }
+
+ /**
+ * Disconnects a RaftHandle that several endpoints can use, when its last
producer or consumer stops.
+ */
+ void disconnectSharedHandle(RaftHandle sharedHandle) {
+ sharedHandleLock.lock();
+ try {
+ Integer users = sharedHandleUsers.get(sharedHandle);
+ if (users == null) {
+ return;
+ }
+ if (users > 1) {
+ sharedHandleUsers.put(sharedHandle, users - 1);
+ } else {
+ sharedHandleUsers.remove(sharedHandle);
+ LOG.trace("Disconnecting JGroupsraft Channel {}",
sharedHandle.channel());
+ sharedHandle.channel().disconnect();
+ }
+ } finally {
+ sharedHandleLock.unlock();
+ }
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ super.doStop();
+ List<RaftHandle> handles;
+ sharedHandleLock.lock();
+ try {
+ handles = new ArrayList<>(sharedHandles);
+ sharedHandleUsers.clear();
+ sharedHandles.clear();
+ } finally {
+ sharedHandleLock.unlock();
+ }
+ // the raftHandle of the component is used by all its endpoints, so it
is closed with the component, as is a
+ // handle given to an endpoint
+ if (raftHandle != null && !handles.contains(raftHandle)) {
+ handles.add(raftHandle);
+ }
+ for (RaftHandle handle : handles) {
+ // each handle is closed, also when closing another one fails
+ try {
+ if (handle.channel() != null) {
+ handle.channel().close();
+ }
+ } catch (Exception e) {
+ LOG.warn("Error closing the channel of the RaftHandle {}: {}",
handle, e.getMessage(), e);
+ }
+ try {
+ if (handle.log() != null) {
+ handle.log().close();
+ }
+ } catch (Exception e) {
+ LOG.warn("Error closing the log of the RaftHandle {}: {}",
handle, e.getMessage(), e);
+ }
+ }
+ }
+
//TODO: implement a org.jgroups.protocols.raft.StateMachine as a Camel
Consumer.
}
diff --git
a/components/camel-jgroups-raft/src/main/java/org/apache/camel/component/jgroups/raft/JGroupsRaftConsumer.java
b/components/camel-jgroups-raft/src/main/java/org/apache/camel/component/jgroups/raft/JGroupsRaftConsumer.java
index 623b4d94bbdc..0587dc1e356f 100644
---
a/components/camel-jgroups-raft/src/main/java/org/apache/camel/component/jgroups/raft/JGroupsRaftConsumer.java
+++
b/components/camel-jgroups-raft/src/main/java/org/apache/camel/component/jgroups/raft/JGroupsRaftConsumer.java
@@ -33,6 +33,9 @@ public class JGroupsRaftConsumer extends DefaultConsumer {
private final CamelRoleChangeListener roleListener;
private final JGroupsRaftEndpoint endpoint;
+ // whether this consumer connected the handle, so that only a consumer
that connected it disconnects it (a consumer
+ // that never started, or failed to start, is stopped as well, while the
handle can be used by other endpoints)
+ private volatile boolean connected;
public JGroupsRaftConsumer(JGroupsRaftEndpoint endpoint, Processor
processor, String clusterName,
boolean enableRoleChangeEvents) {
@@ -53,6 +56,7 @@ public class JGroupsRaftConsumer extends DefaultConsumer {
endpoint.getResolvedRaftHandle().addRoleListener(roleListener);
}
endpoint.connect();
+ connected = true;
}
@Override
@@ -61,7 +65,10 @@ public class JGroupsRaftConsumer extends DefaultConsumer {
LOG.debug("Closing connection to cluster: {} from roleListener:
{}.", clusterName, roleListener);
endpoint.getResolvedRaftHandle().removeRoleListener(roleListener);
}
- endpoint.disconnect();
+ if (connected) {
+ connected = false;
+ endpoint.disconnect();
+ }
super.doStop();
}
}
diff --git
a/components/camel-jgroups-raft/src/main/java/org/apache/camel/component/jgroups/raft/JGroupsRaftEndpoint.java
b/components/camel-jgroups-raft/src/main/java/org/apache/camel/component/jgroups/raft/JGroupsRaftEndpoint.java
index b0d16e839aae..6885d8f913ca 100644
---
a/components/camel-jgroups-raft/src/main/java/org/apache/camel/component/jgroups/raft/JGroupsRaftEndpoint.java
+++
b/components/camel-jgroups-raft/src/main/java/org/apache/camel/component/jgroups/raft/JGroupsRaftEndpoint.java
@@ -114,19 +114,27 @@ public class JGroupsRaftEndpoint extends DefaultEndpoint {
@Override
protected void doStop() throws Exception {
- LOG.info("Closing JGroupsraft Channel {}", getEndpointUri());
- if (resolvedRaftHandle != null && resolvedRaftHandle.channel() !=
null) {
- resolvedRaftHandle.channel().close();
- LOG.info("Closed JGroupsraft Channel {}", getEndpointUri());
- }
- LOG.info("Closing Log {}", getEndpointUri());
- if (resolvedRaftHandle != null && resolvedRaftHandle.log() != null) {
- resolvedRaftHandle.log().close();
- LOG.info("Closed Log Channel {}", getEndpointUri());
+ // a handle this endpoint did not create (such as the raftHandle of
the component) is used by other endpoints,
+ // and is closed by the component
+ if (resolvedRaftHandle != null && isOwnHandle()) {
+ LOG.info("Closing JGroupsraft Channel {}", getEndpointUri());
+ if (resolvedRaftHandle.channel() != null) {
+ resolvedRaftHandle.channel().close();
+ LOG.info("Closed JGroupsraft Channel {}", getEndpointUri());
+ }
+ LOG.info("Closing Log {}", getEndpointUri());
+ if (resolvedRaftHandle.log() != null) {
+ resolvedRaftHandle.log().close();
+ LOG.info("Closed Log Channel {}", getEndpointUri());
+ }
}
super.doStop();
}
+ private boolean isOwnHandle() {
+ return raftHandle == null;
+ }
+
private RaftHandle resolveHandle() throws Exception {
if (raftHandle != null) {
LOG.trace("Raft Handle resolved as passed by Component: {}",
raftHandle);
@@ -136,14 +144,14 @@ public class JGroupsRaftEndpoint extends DefaultEndpoint {
LOG.trace("Raft Handle created with configured channelProperties:
{} and state machine: {}", channelProperties,
stateMachine);
@SuppressWarnings("resource")
- // NOTE: channel will be closed by the component during
+ // NOTE: channel will be closed by this endpoint during
// stop() lifecycle step.
JChannel channel = new JChannel(channelProperties).name(raftId);
// NOSONAR
return new RaftHandle(channel, stateMachine).raftId(raftId);
}
LOG.trace("Raft Handle created with defaults: {}, {},",
JGroupsRaftConstants.DEFAULT_JGROUPSRAFT_CONFIG, stateMachine);
@SuppressWarnings("resource")
- // NOTE: channel will be closed by the component during
+ // NOTE: channel will be closed by this endpoint during
// stop() lifecycle step.
JChannel channel = new
JChannel(JGroupsRaftConstants.DEFAULT_JGROUPSRAFT_CONFIG).name(raftId); //
NOSONAR
return new RaftHandle(channel, stateMachine).raftId(raftId);
@@ -155,9 +163,13 @@ public class JGroupsRaftEndpoint extends DefaultEndpoint {
* @throws Exception
*/
public void connect() throws Exception {
- connectCount.incrementAndGet();
LOG.trace("Connecting JGroups-raft Channel {} with cluster name: {},
raftHandle: {} and using config: {}",
getEndpointUri(), clusterName, resolvedRaftHandle,
channelProperties == null ? "default" : channelProperties);
+ if (!isOwnHandle()) {
+ ((JGroupsRaftComponent)
getComponent()).connectSharedHandle(resolvedRaftHandle, clusterName);
+ return;
+ }
+ connectCount.incrementAndGet();
resolvedRaftHandle.channel().connect(clusterName);
}
@@ -165,6 +177,10 @@ public class JGroupsRaftEndpoint extends DefaultEndpoint {
* Disconnect shared RaftHandle channel, called by producer and consumer.
*/
public void disconnect() {
+ if (!isOwnHandle()) {
+ ((JGroupsRaftComponent)
getComponent()).disconnectSharedHandle(resolvedRaftHandle);
+ return;
+ }
if (connectCount.decrementAndGet() == 0) {
LOG.trace("Disconnecting JGroupsraft Channel {}",
getEndpointUri());
resolvedRaftHandle.channel().disconnect();
diff --git
a/components/camel-jgroups-raft/src/main/java/org/apache/camel/component/jgroups/raft/JGroupsRaftProducer.java
b/components/camel-jgroups-raft/src/main/java/org/apache/camel/component/jgroups/raft/JGroupsRaftProducer.java
index dca07dfc7b92..5ee0e999d49b 100644
---
a/components/camel-jgroups-raft/src/main/java/org/apache/camel/component/jgroups/raft/JGroupsRaftProducer.java
+++
b/components/camel-jgroups-raft/src/main/java/org/apache/camel/component/jgroups/raft/JGroupsRaftProducer.java
@@ -32,6 +32,8 @@ public class JGroupsRaftProducer extends DefaultProducer {
// Producer settings
private final JGroupsRaftEndpoint endpoint;
+ // whether this producer connected the handle, so that a producer that
never started does not disconnect it
+ private volatile boolean connected;
// Constructor
public JGroupsRaftProducer(JGroupsRaftEndpoint endpoint) {
@@ -45,11 +47,15 @@ public class JGroupsRaftProducer extends DefaultProducer {
protected void doStart() throws Exception {
super.doStart();
endpoint.connect();
+ connected = true;
}
@Override
protected void doStop() throws Exception {
- endpoint.disconnect();
+ if (connected) {
+ connected = false;
+ endpoint.disconnect();
+ }
super.doStop();
}
diff --git
a/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/JGroupsRaftComponentHandleEndpointsTest.java
b/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/JGroupsRaftComponentHandleEndpointsTest.java
new file mode 100644
index 000000000000..1c53c4667678
--- /dev/null
+++
b/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/JGroupsRaftComponentHandleEndpointsTest.java
@@ -0,0 +1,118 @@
+/*
+ * 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 java.util.concurrent.TimeUnit;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.jgroups.raft.utils.NopStateMachine;
+import org.apache.camel.spi.Registry;
+import org.jgroups.JChannel;
+import org.jgroups.raft.RaftHandle;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The raftHandle of the component is used by every endpoint of the component:
an endpoint whose producer or consumer
+ * stops, or that is stopped itself, must not disconnect or close the handle
while the other endpoints use it.
+ */
+class JGroupsRaftComponentHandleEndpointsTest extends JGroupsRaftAbstractTest {
+
+ private static final String CLUSTER_NAME =
"JGroupsRaftComponentHandleEndpointsTest";
+ private static final String PRODUCER_URI = "jgroups-raft:" + CLUSTER_NAME;
+ private static final String CONSUMER_URI = "jgroups-raft:" + CLUSTER_NAME
+ "?enableRoleChangeEvents=true";
+
+ private RaftHandle componentHandle;
+ private RaftHandle endpointHandle;
+
+ @Override
+ protected void bindToRegistry(Registry registry) throws Exception {
+ // a handle given to one endpoint (raftHandle=#endpointHandle) is not
created by the endpoint either
+ endpointHandle = new RaftHandle(new JChannel("raftB.xml").name("B"),
new NopStateMachine()).raftId("B");
+ registry.bind("endpointHandle", endpointHandle);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ // a single member cluster (raft.xml), so the handle becomes
leader on its own and can commit
+ componentHandle = new RaftHandle(new
JChannel("raft.xml").name("A"), new NopStateMachine()).raftId("A");
+ JGroupsRaftComponent component = new JGroupsRaftComponent();
+ component.setRaftHandle(componentHandle);
+ context.addComponent("jgroups-raft", component);
+
+ // two endpoints of the same cluster: the consumer has an
option the producer has not
+ from("direct:send").routeId("producer").to(PRODUCER_URI);
+ from(CONSUMER_URI).routeId("consumer").to("mock:events");
+
+ from("jgroups-raft:" + CLUSTER_NAME +
"B?raftHandle=#endpointHandle").to("mock:endpointHandle");
+ }
+ };
+ }
+
+ @BeforeEach
+ void endpointsShareTheHandleOfTheComponent() {
+ JGroupsRaftEndpoint producerEndpoint =
getMandatoryEndpoint(PRODUCER_URI, JGroupsRaftEndpoint.class);
+ JGroupsRaftEndpoint consumerEndpoint =
getMandatoryEndpoint(CONSUMER_URI, JGroupsRaftEndpoint.class);
+ assertNotSame(producerEndpoint, consumerEndpoint);
+ assertSame(componentHandle, producerEndpoint.getResolvedRaftHandle());
+ assertSame(componentHandle, consumerEndpoint.getResolvedRaftHandle());
+ waitForLeader(30, componentHandle);
+ }
+
+ @Test
+ void stoppingTheProducerKeepsTheHandleConnected() throws Exception {
+ context.getRouteController().stopRoute("producer");
+
+ assertTrue(componentHandle.channel().isConnected(), "the consumer
route still uses the handle");
+ assertHandleCommits();
+ }
+
+ @Test
+ void removingTheProducerRouteKeepsTheHandleOpen() throws Exception {
+ context.getRouteController().stopRoute("producer");
+ context.removeRoute("producer");
+
+ assertFalse(componentHandle.channel().isClosed(), "the consumer route
still uses the handle");
+ assertTrue(componentHandle.channel().isConnected(), "the consumer
route still uses the handle");
+ assertHandleCommits();
+ }
+
+ @Test
+ void stoppingTheCamelContextClosesTheHandles() {
+ assertTrue(endpointHandle.channel().isConnected());
+
+ context.stop();
+
+ assertTrue(componentHandle.channel().isClosed());
+ assertTrue(endpointHandle.channel().isClosed());
+ }
+
+ private void assertHandleCommits() throws Exception {
+ waitForLeader(30, componentHandle);
+ long commitIndex = componentHandle.commitIndex();
+ componentHandle.set(new byte[] { 1 }, 0, 1, 10, TimeUnit.SECONDS);
+ assertTrue(componentHandle.commitIndex() > commitIndex);
+ }
+}
diff --git
a/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/JGroupsRaftNeverStartedConsumerTest.java
b/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/JGroupsRaftNeverStartedConsumerTest.java
new file mode 100644
index 000000000000..7c778539b558
--- /dev/null
+++
b/components/camel-jgroups-raft/src/test/java/org/apache/camel/component/jgroups/raft/JGroupsRaftNeverStartedConsumerTest.java
@@ -0,0 +1,62 @@
+/*
+ * 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.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.jgroups.raft.utils.NopStateMachine;
+import org.jgroups.JChannel;
+import org.jgroups.raft.RaftHandle;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A route whose consumer never started (autoStartup false) is stopped or
removed: it must not disconnect the raftHandle
+ * of the component, which the producer of another endpoint uses.
+ */
+class JGroupsRaftNeverStartedConsumerTest extends JGroupsRaftAbstractTest {
+
+ private static final String CLUSTER_NAME =
"JGroupsRaftNeverStartedConsumerTest";
+
+ private RaftHandle componentHandle;
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ componentHandle = new RaftHandle(new
JChannel("raft.xml").name("A"), new NopStateMachine()).raftId("A");
+ JGroupsRaftComponent component = new JGroupsRaftComponent();
+ component.setRaftHandle(componentHandle);
+ context.addComponent("jgroups-raft", component);
+
+ from("direct:send").routeId("producer").to("jgroups-raft:" +
CLUSTER_NAME);
+ from("jgroups-raft:" + CLUSTER_NAME +
"?enableRoleChangeEvents=true").routeId("lazy").autoStartup(false)
+ .to("mock:events");
+ }
+ };
+ }
+
+ @Test
+ void stoppingANeverStartedConsumerRouteKeepsTheHandleConnected() throws
Exception {
+ assertTrue(componentHandle.channel().isConnected(), "the producer
connected the handle");
+ context.getRouteController().stopRoute("lazy");
+ assertTrue(componentHandle.channel().isConnected(), "after stopRoute
of the never started route");
+ context.removeRoute("lazy");
+ assertTrue(componentHandle.channel().isConnected(), "after removeRoute
of the never started route");
+ }
+}
