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 63ab9eb9d1d8 CAMEL-25613: camel-jmx - consumer reconnects once per
lost connection and keeps the new connection (#27800)
63ab9eb9d1d8 is described below
commit 63ab9eb9d1d8d53ef537dde2bd604f5d75e9f926
Author: allthingssecurity <[email protected]>
AuthorDate: Wed Oct 14 17:16:04 2026 +0530
CAMEL-25613: camel-jmx - consumer reconnects once per lost connection and
keeps the new connection (#27800)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/component/jmx/JMXConsumer.java | 112 ++++++-
.../component/jmx/JMXConsumerReconnectTest.java | 362 +++++++++++++++++++++
2 files changed, 469 insertions(+), 5 deletions(-)
diff --git
a/components/camel-jmx/src/main/java/org/apache/camel/component/jmx/JMXConsumer.java
b/components/camel-jmx/src/main/java/org/apache/camel/component/jmx/JMXConsumer.java
index 2ddca7354c3f..2debd042d147 100644
---
a/components/camel-jmx/src/main/java/org/apache/camel/component/jmx/JMXConsumer.java
+++
b/components/camel-jmx/src/main/java/org/apache/camel/component/jmx/JMXConsumer.java
@@ -23,7 +23,9 @@ import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import javax.management.ListenerNotFoundException;
import javax.management.MBeanServerConnection;
import javax.management.Notification;
import javax.management.NotificationFilter;
@@ -51,8 +53,8 @@ public class JMXConsumer extends DefaultConsumer implements
NotificationListener
private static final Logger LOG =
LoggerFactory.getLogger(JMXConsumer.class);
private JMXEndpoint mJmxEndpoint;
- private JMXConnector mConnector;
- private String mConnectionId;
+ private volatile JMXConnector mConnector;
+ private volatile String mConnectionId;
/**
* Used for processing notifications (should not block notification thread)
@@ -63,7 +65,13 @@ public class JMXConsumer extends DefaultConsumer implements
NotificationListener
/**
* Used to schedule delayed connection attempts
*/
- private ScheduledExecutorService mScheduledExecutor;
+ private volatile ScheduledExecutorService mScheduledExecutor;
+
+ /**
+ * Whether a reconnect is scheduled and has not begun: a lost connection
is reported more than once (the connector
+ * sends "failed" and then "closed" for it) and must be followed by one
reconnect
+ */
+ private final AtomicBoolean mReconnectPending = new AtomicBoolean();
/**
* Used to receive notifications about lost connections
@@ -138,6 +146,13 @@ public class JMXConsumer extends DefaultConsumer
implements NotificationListener
*/
private void initNetworkConnection() throws IOException {
if (mConnector != null) {
+ try {
+ // this consumer closes the connection itself, so its "closed"
notification is not a lost connection:
+ // without the listener it cannot schedule a reconnect for a
connection that is being replaced
+
mConnector.removeConnectionNotificationListener(getConnectionNotificationListener());
+ } catch (Exception e) {
+ // ignore, the listener is not registered on a connector that
could not be opened
+ }
try {
mConnector.close();
} catch (Exception e) {
@@ -189,10 +204,18 @@ public class JMXConsumer extends DefaultConsumer
implements NotificationListener
@Override
public void handleNotification(Notification notification, Object
handback) {
JMXConnectionNotification connectionNotification =
(JMXConnectionNotification) notification;
+ if
(connectionNotification.getType().equals(JMXConnectionNotification.OPENED)) {
+ connectorReconnected(connectionNotification.getConnectionId());
+ return;
+ }
// only reset the connection if the notification is for the
connection from this endpoint
if
(!connectionNotification.getConnectionId().equals(mConnectionId)) {
return;
}
+ if (!isRunAllowed()) {
+ // a notification that arrives after the consumer was stopped
+ return;
+ }
if
(connectionNotification.getType().equals(JMXConnectionNotification.NOTIFS_LOST)
||
connectionNotification.getType().equals(JMXConnectionNotification.CLOSED)
||
connectionNotification.getType().equals(JMXConnectionNotification.FAILED)) {
@@ -208,10 +231,59 @@ public class JMXConsumer extends DefaultConsumer
implements NotificationListener
}
/**
- * Schedules an attempt to re-initialize a lost connection after the
reconnect delay
+ * The connector lost its connection and opened another one by itself (the
listener is added after the connector is
+ * connected, so the first connection is not reported here). The new
connection has another id, which the following
+ * notifications carry, and the connector has registered the listener of
this consumer again only if the MBean
+ * existed at that moment. So the consumer subscribes again on that
connection, and connects again only if that
+ * fails.
+ */
+ private void connectorReconnected(String connectionId) {
+ if (connectionId == null || connectionId.equals(mConnectionId)) {
+ return;
+ }
+ mConnectionId = connectionId;
+ if (!isRunAllowed()) {
+ return;
+ }
+ if (mJmxEndpoint.isReconnectOnConnectionFailure()) {
+ LOG.warn("The JMX connector has opened a new connection for: {}.
Subscribing again on that connection.",
+ URISupport.sanitizeUri(mJmxEndpoint.getEndpointUri()));
+ scheduleReconnect(true);
+ } else {
+ LOG.warn("The JMX connector has opened a new connection for: {}.
The JMX consumer may not be subscribed"
+ + " anymore. Use 'reconnectOnConnectionFailure' to enable
reconnections.",
+ URISupport.sanitizeUri(mJmxEndpoint.getEndpointUri()));
+ }
+ }
+
+ /**
+ * Schedules an attempt to re-initialize a lost connection after the
reconnect delay, unless an attempt is already
+ * scheduled or the consumer is stopped
*/
protected void scheduleReconnect() {
+ scheduleReconnect(false);
+ }
+
+ /**
+ * Schedules an attempt to re-initialize the connection after the
reconnect delay, unless an attempt is already
+ * scheduled or the consumer is stopped.
+ *
+ * @param resubscribeFirst whether the attempt first subscribes again on
the current connection, which the connector
+ * has restored by itself, and connects again only
when that fails
+ */
+ private void scheduleReconnect(boolean resubscribeFirst) {
+ if (!isRunAllowed() || !mReconnectPending.compareAndSet(false, true)) {
+ return;
+ }
Runnable startRunnable = () -> {
+ // from here on a lost connection needs another attempt, also when
it is reported while this one runs
+ mReconnectPending.set(false);
+ if (!isRunAllowed()) {
+ return;
+ }
+ if (resubscribeFirst && resubscribe()) {
+ return;
+ }
try {
initNetworkConnection();
addNotificationListener();
@@ -222,7 +294,35 @@ public class JMXConsumer extends DefaultConsumer
implements NotificationListener
};
LOG.info("Delaying JMX consumer reconnection for endpoint {}. Trying
again in {} seconds.",
URISupport.sanitizeUri(mJmxEndpoint.getEndpointUri()),
mJmxEndpoint.getReconnectDelay());
- getExecutor().schedule(startRunnable,
mJmxEndpoint.getReconnectDelay(), TimeUnit.SECONDS);
+ try {
+ getExecutor().schedule(startRunnable,
mJmxEndpoint.getReconnectDelay(), TimeUnit.SECONDS);
+ } catch (RuntimeException e) {
+ mReconnectPending.set(false);
+ throw e;
+ }
+ }
+
+ /**
+ * Subscribes again on the current connection: removes every registration
of this listener (the connector has
+ * registered it again if the MBean existed when it reconnected) and adds
one. So the consumer is subscribed once,
+ * without closing a connection that works.
+ *
+ * @return false if that failed, then the connection has to be replaced
+ */
+ private boolean resubscribe() {
+ try {
+ try {
+
getServerConnection().removeNotificationListener(getEndpoint().getJMXObjectName(),
this);
+ } catch (ListenerNotFoundException e) {
+ // the connector did not register the listener again
+ }
+ addNotificationListener();
+ return true;
+ } catch (Exception e) {
+ LOG.warn("Failed to subscribe again on the connection opened by
the JMX connector, reconnecting. >> {}",
+ e.getMessage());
+ return false;
+ }
}
/**
@@ -267,6 +367,8 @@ public class JMXConsumer extends DefaultConsumer implements
NotificationListener
getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(mScheduledExecutor);
mScheduledExecutor = null;
}
+ // the scheduled reconnect, if there was one, is cancelled
+ mReconnectPending.set(false);
removeNotificationListeners();
diff --git
a/components/camel-jmx/src/test/java/org/apache/camel/component/jmx/JMXConsumerReconnectTest.java
b/components/camel-jmx/src/test/java/org/apache/camel/component/jmx/JMXConsumerReconnectTest.java
new file mode 100644
index 000000000000..003e5c1971b7
--- /dev/null
+++
b/components/camel-jmx/src/test/java/org/apache/camel/component/jmx/JMXConsumerReconnectTest.java
@@ -0,0 +1,362 @@
+/*
+ * 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.jmx;
+
+import java.rmi.NoSuchObjectException;
+import java.rmi.registry.LocateRegistry;
+import java.rmi.registry.Registry;
+import java.rmi.server.UnicastRemoteObject;
+import java.util.Collections;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
+import javax.management.InstanceNotFoundException;
+import javax.management.MBeanServerFactory;
+import javax.management.NotificationListener;
+import javax.management.remote.JMXConnectionNotification;
+import javax.management.remote.JMXConnectorServer;
+import javax.management.remote.JMXConnectorServerFactory;
+import javax.management.remote.JMXServiceURL;
+
+import org.apache.camel.Route;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.AvailablePortFinder;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A remote JMX consumer with reconnectOnConnectionFailure=true must replace a
lost connection once: one reconnect for
+ * the "failed" and "closed" notifications that the connector sends for one
lost connection, no reconnect for the
+ * connection it closes itself, and none after it was stopped. The connector
server counts the client connections that
+ * it accepts.
+ */
+class JMXConsumerReconnectTest extends SimpleBeanFixture {
+
+ @RegisterExtension
+ AvailablePortFinder.Port portHolder = AvailablePortFinder.find();
+
+ /**
+ * How long no connection may be opened: three times the reconnect delay
of one second
+ */
+ private static final int QUIET_SECONDS = 3;
+
+ private final Semaphore opened = new Semaphore(0);
+ private JMXServiceURL url;
+ private JMXConnectorServer connector;
+ private Registry registry;
+ private boolean beanRegistered;
+
+ @BeforeEach
+ @Override
+ public void setUp() throws Exception {
+ url = new JMXServiceURL("service:jmx:rmi:///jndi/rmi://localhost:" +
portHolder.getPort() + "/" + DOMAIN);
+ registry = LocateRegistry.createRegistry(portHolder.getPort());
+ initServer();
+ initBean();
+ initContext();
+ startContext();
+ }
+
+ @Override
+ @AfterEach
+ public void tearDown() throws Exception {
+ if (!getContext().isStopped()) {
+ getContext().stop();
+ }
+ if (beanRegistered) {
+ unregisterBean(makeObjectName("simpleBean"));
+ }
+ if (connector != null) {
+ connector.stop();
+ }
+ if (registry != null) {
+ try {
+ UnicastRemoteObject.unexportObject(registry, true);
+ } catch (NoSuchObjectException e) {
+ // already unexported
+ }
+ }
+ }
+
+ @Override
+ protected void initServer() throws Exception {
+ server = MBeanServerFactory.newMBeanServer(DOMAIN);
+ connector = JMXConnectorServerFactory.newJMXConnectorServer(url,
Collections.<String, Object> emptyMap(), server);
+ // one permit for every client connection that this connector server
accepts
+ connector.addNotificationListener((notification, handback) -> {
+ if
(JMXConnectionNotification.OPENED.equals(notification.getType())) {
+ opened.release();
+ }
+ }, null, null);
+ connector.start();
+ }
+
+ @Override
+ protected void initBean() throws Exception {
+ super.initBean();
+ beanRegistered = true;
+ }
+
+ @Override
+ protected JMXUriBuilder buildFromURI() {
+ return
super.buildFromURI().withServerName(url.toString()).withReconnectDelay(1)
+ .withReconnectOnConnectionFailure(true);
+ }
+
+ /**
+ * A lost connection: the connector reports it with "failed" and then,
when it closes itself, with "closed", both
+ * with the id of that connection (observed with the RMI connector of JDK
21 when the server can no longer be
+ * reached). That is one lost connection, so the consumer must connect
once.
+ */
+ @Test
+ void testReconnectsOnceWhenTheConnectionFailed() throws Exception {
+ MockEndpoint mock = getMockFixture().getMockEndpoint();
+ awaitFirstConnection(mock);
+
+ String connectionId = connector.getConnectionIds()[0];
+ NotificationListener listener = connectionListener();
+ listener.handleNotification(new JMXConnectionNotification(
+ JMXConnectionNotification.FAILED, connector, connectionId, 1,
"Failed to communicate with the server",
+ null), null);
+ listener.handleNotification(new JMXConnectionNotification(
+ JMXConnectionNotification.CLOSED, connector, connectionId, 2,
"Client has been closed", null), null);
+
+ assertTrue(opened.tryAcquire(20, TimeUnit.SECONDS), "the consumer did
not replace its connection");
+
+ assertKeepsConnectionAndIsSubscribedOnce(mock);
+ }
+
+ /**
+ * The server restarts and the MBean is registered a little later than the
connector is back (the usual order when
+ * an application restarts). Whether the connector re-establishes the
connection itself or reports it as lost
+ * depends on timing; the consumer must keep trying to subscribe either
way, and keep the connection it subscribed
+ * on.
+ */
+ @Test
+ void testSubscribesWhenTheBeanIsRegisteredAfterTheServerIsBack() throws
Exception {
+ MockEndpoint mock = getMockFixture().getMockEndpoint();
+ awaitFirstConnection(mock);
+
+ // the server goes away and comes back without the MBean
+ connector.stop();
+ beanRegistered = false;
+ initServer();
+
+ // two connections: a connection without a subscription is not the
end, the consumer connects again
+ assertTrue(opened.tryAcquire(2, 30, TimeUnit.SECONDS),
+ "the consumer did not keep trying to subscribe after the
server came back without the MBean");
+ assertFalse(server.isRegistered(makeObjectName("simpleBean")));
+
+ // now the MBean is there: one of the next attempts subscribes
+ mock.reset();
+ mock.setMinimumExpectedMessageCount(1);
+ initBean();
+ awaitSubscribed(mock);
+
+ // The consumer is subscribed, so it has no reason to connect again.
The check does not depend on the moment
+ // at which it starts: it tolerates a connection that is being opened
right now and fails only when the
+ // consumer keeps connecting.
+ assertTrue(awaitNoConnectionFor(QUIET_SECONDS, 3),
+ "the consumer kept connecting although it was subscribed on a
working connection");
+
+ assertDeliversOnce(mock);
+ }
+
+ /**
+ * The consumer replaces its connection when the server reports lost
notifications. It closes the old connection
+ * itself, which must not count as one more lost connection.
+ */
+ @Test
+ void testReconnectsOnceWhenNotificationsWereLost() throws Exception {
+ MockEndpoint mock = getMockFixture().getMockEndpoint();
+ awaitFirstConnection(mock);
+
+ String connectionId = connector.getConnectionIds()[0];
+ connectionListener().handleNotification(new JMXConnectionNotification(
+ JMXConnectionNotification.NOTIFS_LOST, connector,
connectionId, 1, "lost notifications", 1L), null);
+
+ assertTrue(opened.tryAcquire(20, TimeUnit.SECONDS), "the consumer did
not replace its connection");
+
+ assertKeepsConnectionAndIsSubscribedOnce(mock);
+ }
+
+ /**
+ * The connector reports that it has opened a new connection by itself (as
the RMI connector does after an I/O error
+ * when the server can be reached), and has registered the listener of the
consumer again because the MBean was
+ * there. The consumer subscribes again on that connection: it does not
close a connection that works, and it is
+ * subscribed once.
+ */
+ @Test
+ void testSubscribesAgainOnTheConnectionThatTheConnectorOpenedByItself()
throws Exception {
+ MockEndpoint mock = getMockFixture().getMockEndpoint();
+ awaitFirstConnection(mock);
+
+ connectorReconnected();
+
+ assertKeepsConnectionAndIsSubscribedOnce(mock);
+ }
+
+ /**
+ * As above, but the connector could not register the listener of the
consumer again (the MBean was not there when
+ * it reconnected). The consumer subscribes again on the connection that
the connector opened.
+ */
+ @Test
+ void testSubscribesAgainWhenTheConnectorReconnectedWithoutTheListener()
throws Exception {
+ MockEndpoint mock = getMockFixture().getMockEndpoint();
+ awaitFirstConnection(mock);
+
+ JMXConsumer consumer = consumer();
+
consumer.getServerConnection().removeNotificationListener(makeObjectName("simpleBean"),
consumer);
+ connectorReconnected();
+
+ mock.reset();
+ mock.setMinimumExpectedMessageCount(1);
+ awaitSubscribed(mock);
+
+ assertKeepsConnectionAndIsSubscribedOnce(mock);
+ }
+
+ /**
+ * As above, but the consumer cannot subscribe on the connection that the
connector opened (here: the MBean is not
+ * registered). Then it connects again, until it can subscribe.
+ */
+ @Test
+ void
testReconnectsWhenItCannotSubscribeAgainOnTheConnectionThatTheConnectorOpened()
throws Exception {
+ MockEndpoint mock = getMockFixture().getMockEndpoint();
+ awaitFirstConnection(mock);
+
+ unregisterBean(makeObjectName("simpleBean"));
+ beanRegistered = false;
+ connectorReconnected();
+
+ assertTrue(opened.tryAcquire(20, TimeUnit.SECONDS),
+ "the consumer did not connect again after it could not
subscribe on the connection of the connector");
+
+ mock.reset();
+ mock.setMinimumExpectedMessageCount(1);
+ initBean();
+ awaitSubscribed(mock);
+
+ assertTrue(awaitNoConnectionFor(QUIET_SECONDS, 3),
+ "the consumer kept connecting although it was subscribed on a
working connection");
+
+ assertDeliversOnce(mock);
+ }
+
+ /**
+ * A notification about a lost connection that arrives after the consumer
was stopped must not make it connect
+ * again.
+ */
+ @Test
+ void testDoesNotReconnectAfterStop() throws Exception {
+ MockEndpoint mock = getMockFixture().getMockEndpoint();
+ awaitFirstConnection(mock);
+
+ String connectionId = connector.getConnectionIds()[0];
+ NotificationListener listener = connectionListener();
+ Route route = getContext().getRoutes().get(0);
+ getContext().getRouteController().stopRoute(route.getRouteId());
+
+ listener.handleNotification(new JMXConnectionNotification(
+ JMXConnectionNotification.FAILED, connector, connectionId, 1,
"Failed to communicate with the server",
+ null), null);
+
+ assertFalse(opened.tryAcquire(QUIET_SECONDS, TimeUnit.SECONDS),
+ "the consumer connected again although it was stopped");
+ }
+
+ private JMXConsumer consumer() {
+ return (JMXConsumer) getContext().getRoutes().get(0).getConsumer();
+ }
+
+ private NotificationListener connectionListener() {
+ return consumer().getConnectionNotificationListener();
+ }
+
+ /**
+ * Sends the notification of a connector that has opened a new connection
by itself.
+ */
+ private void connectorReconnected() {
+ connectionListener().handleNotification(new JMXConnectionNotification(
+ JMXConnectionNotification.OPENED, connector,
"rmi://reconnected 2", 1, "Reconnected to server", null),
+ null);
+ }
+
+ private void awaitFirstConnection(MockEndpoint mock) throws Exception {
+ assertTrue(opened.tryAcquire(20, TimeUnit.SECONDS), "the consumer did
not connect at start");
+ getSimpleMXBean().touch();
+ mock.assertIsSatisfied(20000);
+ }
+
+ /**
+ * The consumer is subscribed on a connection that works: it must not
connect again (waits three times the reconnect
+ * delay of one second for a connection that must not come), and one
notification is delivered once.
+ */
+ private void assertKeepsConnectionAndIsSubscribedOnce(MockEndpoint mock)
throws Exception {
+ assertFalse(opened.tryAcquire(QUIET_SECONDS, TimeUnit.SECONDS),
+ "the consumer connected again although it was subscribed on a
working connection");
+
+ assertDeliversOnce(mock);
+ }
+
+ private void assertDeliversOnce(MockEndpoint mock) throws Exception {
+ mock.reset();
+ mock.setExpectedMessageCount(1);
+ // keep checking after the first message, so that a duplicate
subscription is detected
+ mock.setAssertPeriod(1000);
+ getSimpleMXBean().touch();
+ mock.assertIsSatisfied(20000);
+ }
+
+ /**
+ * Waits until the connector server has accepted no connection for the
given number of seconds; gives up after the
+ * given number of periods that had a connection.
+ */
+ private boolean awaitNoConnectionFor(int seconds, int attempts) throws
InterruptedException {
+ for (int i = 0; i < attempts; i++) {
+ if (!opened.tryAcquire(seconds, TimeUnit.SECONDS)) {
+ return true;
+ }
+ opened.drainPermits();
+ }
+ return false;
+ }
+
+ /**
+ * Touches the bean until the consumer delivers a notification (bounded:
150 rounds of at most 200 ms).
+ */
+ private void awaitSubscribed(MockEndpoint mock) throws Exception {
+ for (int i = 0; i < 150; i++) {
+ try {
+ getSimpleMXBean().touch();
+ } catch (Exception e) {
+ if (!(e.getCause() instanceof InstanceNotFoundException)) {
+ throw e;
+ }
+ }
+ if (mock.await(200, TimeUnit.MILLISECONDS)) {
+ return;
+ }
+ }
+ mock.assertIsSatisfied(0);
+ }
+}
