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 c96869e6300f CAMEL-26080: camel-jgroups - do not close a channel that
the user gave to the component or an endpoint when Camel stops (#29201)
c96869e6300f is described below
commit c96869e6300f14dbc9f8e45c7588f6d3062ecf5d
Author: allthingssecurity <[email protected]>
AuthorDate: Thu Nov 5 12:34:34 2026 +0530
CAMEL-26080: camel-jgroups - do not close a channel that the user gave to
the component or an endpoint when Camel stops (#29201)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/component/jgroups/JGroupsComponent.java | 32 ++++++------
.../camel/component/jgroups/JGroupsEndpoint.java | 2 +-
.../JGroupsComponentChannelEndpointsTest.java | 57 ++++++++++++++++++++--
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 9 ++++
4 files changed, 80 insertions(+), 20 deletions(-)
diff --git
a/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsComponent.java
b/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsComponent.java
index 264a3738c861..8ac4ed72cbe8 100644
---
a/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsComponent.java
+++
b/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsComponent.java
@@ -17,11 +17,9 @@
package org.apache.camel.component.jgroups;
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;
@@ -30,6 +28,8 @@ import org.apache.camel.spi.Metadata;
import org.apache.camel.spi.annotations.Component;
import org.apache.camel.support.DefaultComponent;
import org.jgroups.JChannel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Component providing support for messages multicasted from- or to JGroups
channels ({@code org.jgroups.Channel}).
@@ -37,6 +37,8 @@ import org.jgroups.JChannel;
@Component("jgroups")
public class JGroupsComponent extends DefaultComponent {
+ private static final Logger LOG =
LoggerFactory.getLogger(JGroupsComponent.class);
+
@Metadata
private JChannel channel;
@Metadata
@@ -66,8 +68,6 @@ public class JGroupsComponent extends DefaultComponent {
// the connected producers and consumers of each channel that the
endpoints did not create (such as the channel of
// this component), as several endpoints use such a channel
private final Map<JChannel, Integer> sharedChannelUsers = new
IdentityHashMap<>();
- // the channels that the endpoints did not create and connected, closed
when this component stops
- private final Set<JChannel> sharedChannels = Collections.newSetFromMap(new
IdentityHashMap<>());
private final Lock sharedChannelLock = new ReentrantLock();
public JGroupsComponent() {
@@ -158,7 +158,6 @@ public class JGroupsComponent extends DefaultComponent {
try {
sharedChannel.connect(clusterName);
sharedChannelUsers.merge(sharedChannel, 1, Integer::sum);
- sharedChannels.add(sharedChannel);
} finally {
sharedChannelLock.unlock();
}
@@ -188,22 +187,25 @@ public class JGroupsComponent extends DefaultComponent {
@Override
protected void doStop() throws Exception {
super.doStop();
- List<JChannel> channels;
+ List<JChannel> connected;
sharedChannelLock.lock();
try {
- channels = new ArrayList<>(sharedChannels);
+ connected = new ArrayList<>(sharedChannelUsers.keySet());
sharedChannelUsers.clear();
- sharedChannels.clear();
} finally {
sharedChannelLock.unlock();
}
- // the channel of the component is used by all its endpoints, so it is
closed with the component, as is a
- // channel given to an endpoint
- if (channel != null && !channels.contains(channel)) {
- channels.add(channel);
- }
- for (JChannel sharedChannel : channels) {
- sharedChannel.close();
+ // the channels that the endpoints did not create (the channel of this
component, or a channel given to an
+ // endpoint) belong to the user, who closes them: a closed JChannel
cannot be connected again, so they are only
+ // disconnected, and the routes can connect them again when Camel
starts again. A channel is still connected
+ // here when a producer that Camel does not stop (such as one that the
application created and started itself)
+ // still uses it; it is disconnected, as the node must leave the
cluster when Camel stops
+ for (JChannel sharedChannel : connected) {
+ try {
+ sharedChannel.disconnect();
+ } catch (Exception e) {
+ LOG.warn("Error disconnecting the JGroups channel {}: {}",
sharedChannel, e.getMessage(), e);
+ }
}
}
diff --git
a/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsEndpoint.java
b/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsEndpoint.java
index 6dbd0d58c75a..f78e0edb68ed 100644
---
a/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsEndpoint.java
+++
b/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsEndpoint.java
@@ -190,7 +190,7 @@ public class JGroupsEndpoint extends DefaultEndpoint {
@Override
protected void doStop() throws Exception {
// a channel this endpoint did not create (such as the channel of the
component) is used by other endpoints,
- // and is closed by the component
+ // and is disconnected by its last user or by the component, but not
closed (it belongs to the user)
if (resolvedChannel != null && isOwnChannel()) {
connectLock.lock();
try {
diff --git
a/components/camel-jgroups/src/test/java/org/apache/camel/component/jgroups/JGroupsComponentChannelEndpointsTest.java
b/components/camel-jgroups/src/test/java/org/apache/camel/component/jgroups/JGroupsComponentChannelEndpointsTest.java
index 74498c81132c..6621ea677367 100644
---
a/components/camel-jgroups/src/test/java/org/apache/camel/component/jgroups/JGroupsComponentChannelEndpointsTest.java
+++
b/components/camel-jgroups/src/test/java/org/apache/camel/component/jgroups/JGroupsComponentChannelEndpointsTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.jgroups;
+import org.apache.camel.Producer;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.spi.Registry;
@@ -24,13 +25,16 @@ import org.jgroups.JChannel;
import org.jgroups.ObjectMessage;
import org.junit.jupiter.api.Test;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* The channel 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 channel
while the other endpoints use it. The
- * channel is closed when the CamelContext stops, as is a channel given to an
endpoint.
+ * channel and a channel given to an endpoint belong to the user: the
CamelContext disconnects them when it stops, but
+ * does not close them, so they can be used again when it starts again.
*/
class JGroupsComponentChannelEndpointsTest extends CamelTestSupport {
@@ -140,12 +144,57 @@ class JGroupsComponentChannelEndpointsTest extends
CamelTestSupport {
}
@Test
- void stoppingTheCamelContextClosesTheChannels() {
+ void stoppingTheCamelContextDisconnectsTheChannelsButDoesNotCloseThem() {
+ assertTrue(componentChannel.isConnected());
assertTrue(endpointChannel.isConnected());
context.stop();
- assertTrue(componentChannel.isClosed());
- assertTrue(endpointChannel.isClosed());
+ assertFalse(componentChannel.isConnected());
+ assertFalse(endpointChannel.isConnected());
+ assertFalse(componentChannel.isClosed());
+ assertFalse(endpointChannel.isClosed());
+ }
+
+ @Test
+ void
stoppingTheCamelContextDisconnectsTheChannelThatAProducerNotStoppedByCamelStillUses()
throws Exception {
+ // a producer that the application created and started itself: Camel
does not stop it
+ JGroupsEndpoint endpoint = context.getEndpoint("jgroups:" + CLUSTER,
JGroupsEndpoint.class);
+ Producer producer = endpoint.createProducer();
+ producer.start();
+
+ context.stop();
+
+ assertFalse(componentChannel.isConnected(), "the node leaves the
cluster when Camel stops");
+ assertFalse(componentChannel.isClosed());
+ // the late stop of that producer finds the channel disconnected
+ assertDoesNotThrow(producer::stop);
+ assertFalse(componentChannel.isClosed());
+ }
+
+ @Test
+ void theRoutesReceiveAgainAfterTheCamelContextIsStoppedAndStartedAgain()
throws Exception {
+ context.stop();
+ // the CamelContext of the test drops the component added in the route
builder and the beans of its registry
+ // when it stops: configure them again with the same channels, as an
application that keeps them has them
+ JGroupsComponent component = new JGroupsComponent();
+ component.setChannel(componentChannel);
+ context.addComponent("jgroups", component);
+ context.getRegistry().bind("endpointChannel", endpointChannel);
+ context.start();
+ assertTrue(componentChannel.isConnected());
+ assertTrue(endpointChannel.isConnected());
+
+ MockEndpoint mock = getMockEndpoint("mock:received");
+ mock.expectedBodiesReceived("after a restart");
+ MockEndpoint endpointChannelMock =
getMockEndpoint("mock:endpointChannel");
+ endpointChannelMock.expectedBodiesReceived("after a restart, endpoint
channel");
+
+ sender.send(new ObjectMessage(null, "after a restart"));
+ try (JChannel senderB = new JChannel()) {
+ senderB.connect(CLUSTER + "B");
+ senderB.send(new ObjectMessage(null, "after a restart, endpoint
channel"));
+ MockEndpoint.assertIsSatisfied(context);
+ }
}
}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index aa2c10c7a18d..9162438e1c97 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -3583,6 +3583,15 @@ content, so the members of the cluster receive a
`byte[]`. Any other body that i
`InvalidPayloadException` that says to convert it first, before it reaches
JGroups. An `onException` for
`NotSerializableException` must handle `InvalidPayloadException` instead.
+=== camel-jgroups - a channel given by the user is not closed when Camel stops
+
+When the CamelContext stops, the `jgroups` component no longer closes the
channel of the component (its `channel`
+option) and the channels given to an endpoint (`channel=#myChannel`). It
disconnects them, so the node leaves the
+cluster, and leaves them open, as a closed `JChannel` cannot be connected
again: the routes can now use them again when
+the CamelContext is started again. The channels that the endpoints create
(with `channelProperties` or the default
+configuration) are closed as before. Close a channel you give to Camel
yourself when the application no longer needs
+it, for example as a bean with a destroy method (`close`).
+
=== camel-jira - the producers wait for the answer of Jira
The `addComment`, `addIssueLink`, `addWorkLog`, `attach`, `deleteIssue`,
`transitionIssue` and `watchers` producers
