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 6319543d332f CAMEL-25659: camel-sjms - recovery notices a connection
failure that happens before the consumers are created (#27856)
6319543d332f is described below
commit 6319543d332f8a8544085f1450d06decc82d22a4
Author: allthingssecurity <[email protected]>
AuthorDate: Thu Oct 15 11:16:52 2026 +0530
CAMEL-25659: camel-sjms - recovery notices a connection failure that
happens before the consumers are created (#27856)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../consumer/SimpleMessageListenerContainer.java | 30 +++++++++++--
.../SjmsConsumerRecoveryFailedAttemptTest.java | 52 +++++++++++++++++++++-
2 files changed, 77 insertions(+), 5 deletions(-)
diff --git
a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/SimpleMessageListenerContainer.java
b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/SimpleMessageListenerContainer.java
index 09daa824b399..094fefff131f 100644
---
a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/SimpleMessageListenerContainer.java
+++
b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/SimpleMessageListenerContainer.java
@@ -22,6 +22,7 @@ import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -69,6 +70,9 @@ public class SimpleMessageListenerContainer extends
ServiceSupport
private final Lock consumerLock = new ReentrantLock();
private Set<MessageConsumer> consumers;
private Set<Session> sessions;
+ // the number of onException calls: a connection recovery compares it
before and after its attempt, as an
+ // onException that comes before the consumers of the new connection are
created finds none to drop
+ private final AtomicLong connectionFailures = new AtomicLong();
private ScheduledExecutorService recoverPool;
private BackgroundTask recoverTask;
private Future<?> recoverFuture;
@@ -180,6 +184,7 @@ public class SimpleMessageListenerContainer extends
ServiceSupport
@Override
public void onException(JMSException exception) {
+ connectionFailures.incrementAndGet();
if (exceptionListener != null) {
try {
exceptionListener.onException(exception);
@@ -208,13 +213,17 @@ public class SimpleMessageListenerContainer extends
ServiceSupport
protected boolean recoverConnection(BackgroundTask task) {
LOG.debug("Recovering from JMS Connection exception (attempt: {})",
task.iteration());
try {
- refreshConnection();
+ long failures = refreshConnectionAndCountFailures();
initConsumers();
connectionLock.lock();
try {
- if (consumers == null) {
- // the new connection failed while its consumers were
created: onException dropped them, and
- // did not schedule another recovery as this one is
running. So this one has to try again
+ if (consumers == null || connectionFailures.get() != failures)
{
+ // the new connection failed during this attempt:
onException dropped its consumers while they
+ // were created, or it came before and found none to drop
(then the consumers on the failed
+ // connection are dropped here). onException did not
schedule another recovery as this one is
+ // running. So this one has to try again
+ this.sessions = null;
+ this.consumers = null;
LOG.debug("JMS Connection failed again during recovery
(attempt: {}). Will try again in {} millis",
task.iteration(), endpoint.getRecoveryInterval());
return false;
@@ -455,14 +464,27 @@ public class SimpleMessageListenerContainer extends
ServiceSupport
}
protected final void refreshConnection() throws Exception {
+ refreshConnectionAndCountFailures();
+ }
+
+ /**
+ * Closes the connection and creates a new one, which is started if the
container is started.
+ *
+ * @return the number of connection failures (onException calls) before
the new connection was created, so a failure
+ * that is counted later may be one of the new connection
+ */
+ private long refreshConnectionAndCountFailures() throws Exception {
connectionLock.lock();
try {
closeConnection(connection);
this.connection = null;
+ // read after the old connection is closed and before the new one
has this container as exception listener
+ long failures = connectionFailures.get();
createConnection();
if (this.connectionStarted) {
startConnection();
}
+ return failures;
} finally {
connectionLock.unlock();
}
diff --git
a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/SjmsConsumerRecoveryFailedAttemptTest.java
b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/SjmsConsumerRecoveryFailedAttemptTest.java
index f009447343f8..71d9cc44e453 100644
---
a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/SjmsConsumerRecoveryFailedAttemptTest.java
+++
b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/SjmsConsumerRecoveryFailedAttemptTest.java
@@ -27,6 +27,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import jakarta.jms.Connection;
import jakarta.jms.ConnectionFactory;
+import jakarta.jms.ExceptionListener;
import jakarta.jms.JMSContext;
import jakarta.jms.JMSException;
import jakarta.jms.Session;
@@ -57,6 +58,8 @@ class SjmsConsumerRecoveryFailedAttemptTest extends
JmsTestSupport {
private final List<TrackedConnection> connections = new
CopyOnWriteArrayList<>();
// what the next connection of the tracking connection factory does wrong
(taken by that connection)
private volatile Fault nextFault;
+ // whether the next connection of the tracking connection factory fails
when it is started (taken by it)
+ private volatile boolean nextFailsOnStart;
@BindToRegistry("trackingConnectionFactory")
private final ConnectionFactory trackingConnectionFactory = new
ConnectionFactory() {
@@ -163,6 +166,38 @@ class SjmsConsumerRecoveryFailedAttemptTest extends
JmsTestSupport {
assertEquals(3, connections.size());
}
+ /**
+ * The new connection reports its failure to the container (its exception
listener) when the recovery starts it,
+ * that is before the consumers are created, and it delivers no message;
the JMS provider still creates the sessions
+ * and consumers on it. The recovery has to notice the failure and try
again with another connection, although
+ * onException found no consumers to drop and none of the create calls
failed.
+ */
+ @Test
+ void testRecoveryTriesAgainWhenConnectionFailsBeforeConsumersAreCreated()
throws Exception {
+ context.getRouteController().startRoute("beforeconsumers");
+ MockEndpoint mock = getMockEndpoint("mock:beforeconsumers");
+ mock.expectedBodiesReceived("before");
+ template.sendBody(QUEUE + "beforeconsumers", "before");
+ mock.assertIsSatisfied();
+
+ nextFailsOnStart = true;
+ container("beforeconsumers").onException(new JMSException("Simulated
failure of the connection"));
+ await().atMost(20, TimeUnit.SECONDS).until(() -> connections.size() >=
2 && connections.get(1).failed.get());
+
+ // the first, the one that failed when it was started, the one in use
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(() ->
assertEquals(3, connections.size(),
+ "connections: the recovery must try again, the connection
failed before the consumers were created"));
+ await().atMost(20, TimeUnit.SECONDS)
+ .untilAsserted(() -> assertEquals(CONSUMERS,
last().openConsumers(), "consumers on the new connection"));
+ assertTrue(connections.get(1).closed.get(), "the connection that
failed when it was started is closed");
+
+ mock.reset();
+ mock.expectedBodiesReceived("after");
+ template.sendBody(QUEUE + "beforeconsumers", "after");
+ mock.assertIsSatisfied();
+ assertEquals(3, connections.size());
+ }
+
/**
* A listener container that fails to start (the second consumer cannot be
created) closes the session and consumer
* it did create, and the session whose consumer failed. The connection is
kept for the next start.
@@ -207,7 +242,7 @@ class SjmsConsumerRecoveryFailedAttemptTest extends
JmsTestSupport {
@Override
public void configure() {
// each test starts its own route, so that the tracked
connections are the ones of that route
- for (String routeId : List.of("first", "last", "during")) {
+ for (String routeId : List.of("first", "last", "during",
"beforeconsumers")) {
from(QUEUE + routeId + "?concurrentConsumers=" + CONSUMERS
+ "&recoveryInterval=200"
+ "&connectionFactory=#trackingConnectionFactory")
.routeId(routeId).autoStartup(false).to("mock:" +
routeId);
@@ -223,9 +258,21 @@ class SjmsConsumerRecoveryFailedAttemptTest extends
JmsTestSupport {
TrackedConnection tracked = new TrackedConnection();
tracked.fault = nextFault;
nextFault = null;
+ tracked.failsOnStart = nextFailsOnStart;
+ nextFailsOnStart = false;
connections.add(tracked);
return (Connection)
Proxy.newProxyInstance(getClass().getClassLoader(), new Class<?>[] {
Connection.class },
(proxy, method, args) -> {
+ if ("setExceptionListener".equals(method.getName())) {
+ tracked.exceptionListener = (ExceptionListener)
args[0];
+ }
+ if ("start".equals(method.getName()) &&
tracked.failsOnStart) {
+ // the connection is not started, so its consumers get
no message, and the failure is reported
+ // on the thread that starts it
+ tracked.failed.set(true);
+ tracked.exceptionListener.onException(new
JMSException("Simulated: the connection failed"));
+ return null;
+ }
if ("createSession".equals(method.getName())) {
Fault fault = tracked.fault;
int number = tracked.sessionCalls.incrementAndGet();
@@ -293,7 +340,10 @@ class SjmsConsumerRecoveryFailedAttemptTest extends
JmsTestSupport {
final AtomicInteger sessionCalls = new AtomicInteger();
final List<TrackedSession> sessions = new CopyOnWriteArrayList<>();
final AtomicBoolean closed = new AtomicBoolean();
+ final AtomicBoolean failed = new AtomicBoolean();
volatile Fault fault;
+ volatile boolean failsOnStart;
+ volatile ExceptionListener exceptionListener;
// a closed connection has closed its sessions, and a closed session
its consumer
int openSessions() {
