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 046f719fb7c4 CAMEL-26536: camel-jms - a consumer stopped and started
again with asyncStopListener=true must keep a running listener container
(#29120)
046f719fb7c4 is described below
commit 046f719fb7c41c88e3cf851199adfb51a0591ea0
Author: allthingssecurity <[email protected]>
AuthorDate: Wed Nov 4 15:06:10 2026 +0530
CAMEL-26536: camel-jms - a consumer stopped and started again with
asyncStopListener=true must keep a running listener container (#29120)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/component/jms/JmsConsumer.java | 166 ++++++++---
.../jms/JmsAsyncStopListenerRestartTest.java | 321 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 8 +
3 files changed, 455 insertions(+), 40 deletions(-)
diff --git
a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConsumer.java
b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConsumer.java
index 7be15c1e46d2..2047ab8d2056 100644
---
a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConsumer.java
+++
b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConsumer.java
@@ -17,6 +17,8 @@
package org.apache.camel.component.jms;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import jakarta.jms.Connection;
import jakarta.jms.ConnectionFactory;
@@ -49,6 +51,8 @@ public class JmsConsumer extends DefaultConsumer implements
Suspendable {
private volatile boolean initialized;
private volatile ExecutorService executorService;
private volatile boolean shutdownExecutorService;
+ // makes the ownership check of the asynchronous start task atomic with
the detach in doStop
+ private final Lock containerLock = new ReentrantLock();
public JmsConsumer(JmsEndpoint endpoint, Processor processor,
AbstractMessageListenerContainer listenerContainer) {
super(endpoint, processor);
@@ -111,9 +115,13 @@ public class JmsConsumer extends DefaultConsumer
implements Suspendable {
* Can be used to start this consumer later if it was configured to not
auto startup.
*/
public void startListenerContainer() {
- LOG.trace("Starting listener container {} on destination {}",
listenerContainer, getDestinationName());
- listenerContainer.start();
- LOG.debug("Started listener container {} on destination {}",
listenerContainer, getDestinationName());
+ startListenerContainer(listenerContainer);
+ }
+
+ private void startListenerContainer(AbstractMessageListenerContainer
container) {
+ LOG.trace("Starting listener container {} on destination {}",
container, getDestinationName(container));
+ container.start();
+ LOG.debug("Started listener container {} on destination {}",
container, getDestinationName(container));
}
/**
@@ -124,9 +132,13 @@ public class JmsConsumer extends DefaultConsumer
implements Suspendable {
* @throws FailedToCreateConsumerException is thrown if testing the
connection failed
*/
protected void testConnectionOnStartup() throws
FailedToCreateConsumerException {
+ testConnectionOnStartup(listenerContainer);
+ }
+
+ private void testConnectionOnStartup(AbstractMessageListenerContainer
container) throws FailedToCreateConsumerException {
try {
- LOG.debug("Testing JMS Connection on startup for destination: {}",
getDestinationName());
- ConnectionFactory connectionfactory =
listenerContainer.getConnectionFactory();
+ LOG.debug("Testing JMS Connection on startup for destination: {}",
getDestinationName(container));
+ ConnectionFactory connectionfactory =
container.getConnectionFactory();
if (connectionfactory != null) {
Connection con = connectionfactory.createConnection();
JmsUtils.closeConnection(con);
@@ -134,9 +146,9 @@ public class JmsConsumer extends DefaultConsumer implements
Suspendable {
LOG.error("connection factory is null");
throw new IllegalStateException("connection factory is null");
}
- LOG.debug("Successfully tested JMS Connection on startup for
destination: {}", getDestinationName());
+ LOG.debug("Successfully tested JMS Connection on startup for
destination: {}", getDestinationName(container));
} catch (Exception e) {
- String msg = "Cannot get JMS Connection on startup for destination
" + getDestinationName();
+ String msg = "Cannot get JMS Connection on startup for destination
" + getDestinationName(container);
throw new FailedToCreateConsumerException(getEndpoint(), msg, e);
}
}
@@ -149,6 +161,7 @@ public class JmsConsumer extends DefaultConsumer implements
Suspendable {
if (listenerContainer == null) {
createMessageListenerContainer();
}
+ final AbstractMessageListenerContainer container = listenerContainer;
getEndpoint().onListenerContainerStarting();
if (getEndpoint().getConfiguration().isAsyncStartListener()) {
@@ -156,84 +169,144 @@ public class JmsConsumer extends DefaultConsumer
implements Suspendable {
@Override
public void run() {
try {
- prepareAndStartListenerContainer();
+ prepareAndStartListenerContainer(container);
} catch (Exception e) {
LOG.warn("Error starting listener container on
destination: {}. This exception will be ignored.",
- getDestinationName(), e);
+ getDestinationName(container), e);
}
}
@Override
public String toString() {
- return "AsyncStartListenerTask[" + getDestinationName() +
"]";
+ return "AsyncStartListenerTask[" +
getDestinationName(container) + "]";
}
});
} else {
- prepareAndStartListenerContainer();
+ prepareAndStartListenerContainer(container);
}
}
protected void prepareAndStartListenerContainer() {
- listenerContainer.afterPropertiesSet();
+ prepareAndStartListenerContainer(listenerContainer);
+ }
+
+ private void
prepareAndStartListenerContainer(AbstractMessageListenerContainer container) {
+ if (listenerContainer != container) {
+ // an early exit without the lock: the check that decides is the
one under containerLock below.
+ // The consumer was stopped (and maybe started again with a new
listener container) before this
+ // asynchronous start task ran
+ LOG.debug("Not starting listener container {} on destination {} as
the consumer no longer uses it", container,
+ getDestinationName(container));
+ return;
+ }
+
+ container.afterPropertiesSet();
// only start listener if auto start is enabled or we are explicit
invoking start later
if (initialized || getEndpoint().isAutoStartup()) {
// should we pre test connections before starting?
if (getEndpoint().isTestConnectionOnStartup()) {
- testConnectionOnStartup();
+ testConnectionOnStartup(container);
}
- startListenerContainer();
+ startListenerContainer(container);
}
- // mark as initialized after the auto-startup check
- initialized = true;
+ containerLock.lock();
+ try {
+ if (listenerContainer == container) {
+ // mark as initialized after the auto-startup check
+ initialized = true;
+ return;
+ }
+ } finally {
+ containerLock.unlock();
+ }
+ // the consumer was stopped while this task was starting the
container, and its stop may have run before
+ // container.start(): stop and destroy it here
+ LOG.debug("Stopping listener container {} on destination {} as the
consumer was stopped while starting it",
+ container, getDestinationName(container));
+ try {
+ container.stop();
+ } finally {
+ container.destroy();
+ }
}
+ /**
+ * Stops and destroys the listener container of this consumer, which
creates a new one when it is started again.
+ */
protected void stopAndDestroyListenerContainer() {
- if (listenerContainer != null) {
+ DetachedListenerContainer detached = detachListenerContainer();
+ if (detached != null) {
+ stopAndDestroyListenerContainer(detached);
+ }
+ }
+
+ /**
+ * Takes the listener container (and its private thread pool) away from
this consumer, so a later start creates a
+ * new one and the (possibly asynchronous) stop of this one cannot touch
it.
+ */
+ private DetachedListenerContainer detachListenerContainer() {
+ containerLock.lock();
+ try {
+ AbstractMessageListenerContainer container = listenerContainer;
+ if (container == null) {
+ return null;
+ }
+ DetachedListenerContainer answer
+ = new DetachedListenerContainer(container,
executorService, shutdownExecutorService);
+ // null container and listener so they are fully re created if
this consumer is restarted
+ // then we will use updated configuration from jms endpoint that
may have been managed using JMX
+ listenerContainer = null;
+ messageListener = null;
+ initialized = false;
+ executorService = null;
+ return answer;
+ } finally {
+ containerLock.unlock();
+ }
+ }
+
+ private void stopAndDestroyListenerContainer(DetachedListenerContainer
detached) {
+ try {
try {
- listenerContainer.stop();
- listenerContainer.destroy();
+ detached.container().stop();
} finally {
- getEndpoint().onListenerContainerStopped();
+ // also when stop fails (a SimpleMessageListenerContainer does
not swallow a failing Connection.stop())
+ detached.container().destroy();
+ }
+ } finally {
+ getEndpoint().onListenerContainerStopped();
+ // shutdown thread pool if listener container was using a private
thread pool
+ if (detached.shutdownExecutorService() &&
detached.executorService() != null) {
+
getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(detached.executorService());
}
}
- // null container and listener so they are fully re created if this
consumer is restarted
- // then we will use updated configuration from jms endpoint that may
have been managed using JMX
- listenerContainer = null;
- messageListener = null;
- initialized = false;
-
- // shutdown thread pool if listener container was using a private
thread pool
- if (shutdownExecutorService && executorService != null) {
-
getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executorService);
- }
- executorService = null;
}
@Override
protected void doStop() throws Exception {
- if (listenerContainer != null) {
-
+ final DetachedListenerContainer detached = detachListenerContainer();
+ if (detached != null) {
if (getEndpoint().getConfiguration().isAsyncStopListener()) {
getEndpoint().getAsyncStartStopExecutorService().submit(new
Runnable() {
@Override
public void run() {
try {
- stopAndDestroyListenerContainer();
+ stopAndDestroyListenerContainer(detached);
} catch (Exception e) {
LOG.warn("Error stopping listener container on
destination: {}. This exception will be ignored.",
- getDestinationName(), e);
+ getDestinationName(detached.container()),
e);
}
}
@Override
public String toString() {
- return "AsyncStopListenerTask[" + getDestinationName()
+ "]";
+ return "AsyncStopListenerTask[" +
getDestinationName(detached.container()) + "]";
}
});
} else {
- stopAndDestroyListenerContainer();
+ stopAndDestroyListenerContainer(detached);
}
}
@@ -263,10 +336,16 @@ public class JmsConsumer extends DefaultConsumer
implements Suspendable {
}
private String getDestinationName() {
- if (listenerContainer.getDestination() != null) {
- return listenerContainer.getDestination().toString();
+ return getDestinationName(listenerContainer);
+ }
+
+ private static String getDestinationName(AbstractMessageListenerContainer
container) {
+ if (container == null) {
+ return null;
+ } else if (container.getDestination() != null) {
+ return container.getDestination().toString();
} else {
- return listenerContainer.getDestinationName();
+ return container.getDestinationName();
}
}
@@ -295,4 +374,11 @@ public class JmsConsumer extends DefaultConsumer
implements Suspendable {
}
}
+ /**
+ * A listener container detached from this consumer by a stop, with the
private thread pool it uses (if any).
+ */
+ private record DetachedListenerContainer(AbstractMessageListenerContainer
container, ExecutorService executorService,
+ boolean shutdownExecutorService) {
+ }
+
}
diff --git
a/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsAsyncStopListenerRestartTest.java
b/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsAsyncStopListenerRestartTest.java
new file mode 100644
index 000000000000..59df575e3295
--- /dev/null
+++
b/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsAsyncStopListenerRestartTest.java
@@ -0,0 +1,321 @@
+/*
+ * 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.jms;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.AbstractExecutorService;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import jakarta.jms.ConnectionFactory;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.infra.core.CamelContextExtension;
+import org.apache.camel.test.infra.core.DefaultCamelContextExtension;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.springframework.jms.listener.AbstractMessageListenerContainer;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Stopping and starting a JMS consumer (as route policies and JMX do) while
the asynchronous stop or start of its
+ * listener container is still pending must leave the consumer consuming.
+ * <p/>
+ * The tasks the consumer submits to the component's async start/stop thread
pool are held, and the test runs them
+ * itself in a chosen order, so the interleaving is deterministic.
+ */
+@TestInstance(TestInstance.Lifecycle.PER_METHOD)
+@Timeout(60)
+public class JmsAsyncStopListenerRestartTest extends AbstractJMSTest {
+
+ @Order(2)
+ @RegisterExtension
+ public static CamelContextExtension camelContextExtension = new
DefaultCamelContextExtension();
+
+ private static final String QUEUE = "JmsAsyncStopListenerRestartTest";
+
+ // static: the CamelContext (and so the component) is created once for the
class
+ private static final HeldExecutorService HELD = new HeldExecutorService();
+ private volatile Runnable initializeHook;
+ private volatile AbstractMessageListenerContainer initializing;
+ protected CamelContext context;
+ protected ProducerTemplate template;
+
+ @Test
+ public void testRestartWhileAsyncStopIsPending() throws Exception {
+ context.getRouteController().startRoute("jms");
+ assertConsumes("Hello World");
+
+ consumer().stop();
+ consumer().start();
+ assertEquals(1, HELD.pending(), "the asynchronous stop of the old
listener container is pending");
+
+ // the stop task of the first run of the route runs after the route
was started again
+ HELD.runPending();
+
+ assertListening();
+ assertConsumes("Bye World");
+ }
+
+ @Test
+ public void testRestartWhileAsyncStopAndStartArePendingStopFirst() throws
Exception {
+ startRouteWithAsyncStart();
+
+ consumer().stop();
+ consumer().start();
+ assertEquals(2, HELD.pending(), "the stop task and the start task are
pending");
+
+ HELD.runPending();
+
+ assertListening();
+ assertConsumes("Bye World");
+ }
+
+ @Test
+ public void testRestartWhileAsyncStopAndStartArePendingStartFirst() throws
Exception {
+ startRouteWithAsyncStart();
+
+ consumer().stop();
+ consumer().start();
+ assertEquals(2, HELD.pending(), "the stop task and the start task are
pending");
+
+ HELD.runPendingInReverse();
+
+ assertListening();
+ assertConsumes("Bye World");
+ }
+
+ @Test
+ public void testStopWhileAsyncStartIsPending() throws Exception {
+ int running = endpoint().getRunningMessageListeners();
+ startRouteWithAsyncStart();
+
+ consumer().stop();
+ consumer().start();
+ AbstractMessageListenerContainer second =
consumer().getListenerContainer();
+ consumer().stop();
+
+ // the start task of the second run of the route runs after the route
was stopped
+ HELD.runPendingInReverse();
+
+ assertTrue(consumer().isStopped(), "the consumer was stopped");
+ assertFalse(second.isActive() && second.isRunning(), "the stopped
route must not keep a listener container running");
+ assertEquals(running, endpoint().getRunningMessageListeners(),
+ "every started listener container is counted as stopped");
+ }
+
+ @Test
+ public void testStopWhileAsyncStartIsInitializingTheListenerContainer()
throws Exception {
+ // the listener containers run the hook (if any) when they are
initialized
+ endpoint().getConfiguration().setMessageListenerContainerFactory(e ->
new DefaultJmsMessageListenerContainer(e) {
+ @Override
+ public void afterPropertiesSet() {
+ Runnable hook = initializeHook;
+ initializeHook = null;
+ if (hook != null) {
+ initializing = this;
+ hook.run();
+ }
+ super.afterPropertiesSet();
+ }
+ });
+ endpoint().getConfiguration().setConsumerType(ConsumerType.Custom);
+ startRouteWithAsyncStart();
+
+ consumer().stop();
+ consumer().start();
+ // the consumer is stopped, and its stop task runs, after the start
task checked that the consumer owns the
+ // listener container and before the start task initializes and starts
it
+ initializeHook = () -> {
+ consumer().stop();
+ HELD.runPending();
+ };
+ HELD.runPending();
+
+ assertTrue(consumer().isStopped(), "the consumer was stopped");
+ assertNotNull(initializing, "the start task initialized the listener
container");
+ assertFalse(initializing.isActive() && initializing.isRunning(),
+ "the stopped consumer must not keep a listener container
running");
+ }
+
+ private void startRouteWithAsyncStart() throws Exception {
+ endpoint().getConfiguration().setAsyncStartListener(true);
+ context.getRouteController().startRoute("jms");
+ HELD.runPending();
+ assertConsumes("Hello World");
+ }
+
+ private JmsEndpoint endpoint() {
+ return context.getEndpoint("activemq:queue:" + QUEUE,
JmsEndpoint.class);
+ }
+
+ private JmsConsumer consumer() {
+ return (JmsConsumer) context.getRoute("jms").getConsumer();
+ }
+
+ private void assertListening() throws Exception {
+ assertTrue(consumer().isStarted(), "the consumer was started");
+ assertTrue(consumer().getListenerContainer().isRunning(),
+ "the started consumer must have a running listener container");
+ }
+
+ private void assertConsumes(String body) throws Exception {
+ MockEndpoint result = getMockEndpoint("mock:result");
+ result.reset();
+ result.expectedBodiesReceived(body);
+ template.sendBody("activemq:queue:" + QUEUE, body);
+ result.assertIsSatisfied();
+ }
+
+ @Override
+ protected String getComponentName() {
+ return "activemq";
+ }
+
+ @Override
+ protected JmsComponent buildComponent(ConnectionFactory connectionFactory)
{
+ JmsComponent jms = new JmsComponent() {
+ @Override
+ protected ExecutorService getAsyncStartStopExecutorService() {
+ return HELD;
+ }
+ };
+ jms.getConfiguration().setConnectionFactory(connectionFactory);
+ jms.getConfiguration().setAsyncStopListener(true);
+ return jms;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ public void configure() {
+ from("activemq:queue:" +
QUEUE).routeId("jms").autoStartup(false).to("mock:result");
+ }
+ };
+ }
+
+ @Override
+ public CamelContextExtension getCamelContextExtension() {
+ return camelContextExtension;
+ }
+
+ @BeforeEach
+ void setUpRequirements() {
+ context = camelContextExtension.getContext();
+ template = camelContextExtension.getProducerTemplate();
+ HELD.hold = true;
+ }
+
+ @AfterEach
+ void stopRoute() throws Exception {
+ // run what is left, and stop the route with the tasks running inline,
for the next test
+ HELD.hold = false;
+ HELD.runPending();
+ context.getRouteController().stopRoute("jms");
+ endpoint().getConfiguration().setAsyncStartListener(false);
+ endpoint().getConfiguration().setMessageListenerContainerFactory(null);
+ endpoint().getConfiguration().setConsumerType(ConsumerType.Default);
+ // drop what a failed test left in the queue, so it does not reach the
next test
+ while
(camelContextExtension.getConsumerTemplate().receiveBody("activemq:queue:" +
QUEUE, 100) != null) {
+ // drained
+ }
+ }
+
+ /**
+ * An executor that keeps the submitted tasks until the test runs them (in
submission order or in reverse order).
+ */
+ static final class HeldExecutorService extends AbstractExecutorService {
+ private final List<Runnable> tasks = new ArrayList<>();
+ volatile boolean hold = true;
+
+ @Override
+ public void execute(Runnable command) {
+ if (hold) {
+ synchronized (tasks) {
+ tasks.add(command);
+ }
+ } else {
+ command.run();
+ }
+ }
+
+ int pending() {
+ synchronized (tasks) {
+ return tasks.size();
+ }
+ }
+
+ void runPending() {
+ runPending(false);
+ }
+
+ void runPendingInReverse() {
+ runPending(true);
+ }
+
+ private void runPending(boolean reverse) {
+ List<Runnable> copy;
+ synchronized (tasks) {
+ copy = new ArrayList<>(tasks);
+ tasks.clear();
+ }
+ if (reverse) {
+ Collections.reverse(copy);
+ }
+ copy.forEach(Runnable::run);
+ }
+
+ @Override
+ public void shutdown() {
+ // noop
+ }
+
+ @Override
+ public List<Runnable> shutdownNow() {
+ return List.of();
+ }
+
+ @Override
+ public boolean isShutdown() {
+ return false;
+ }
+
+ @Override
+ public boolean isTerminated() {
+ return false;
+ }
+
+ @Override
+ public boolean awaitTermination(long timeout, TimeUnit unit) {
+ return true;
+ }
+ }
+}
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 ed00683ba05a..f23049e8871e 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
@@ -3418,6 +3418,14 @@ comment after the transition was made, `watchers` and
`addIssue` can fail after
(`addIssue` sets the created issue as the message body before it adds the
watchers), and a retry or redelivery
repeats the earlier requests.
+=== camel-jms - JmsConsumer starts and stops its listener container through
private methods
+
+`JmsConsumer.doStart` and `doStop` no longer call the protected methods
`prepareAndStartListenerContainer()`,
+`stopAndDestroyListenerContainer()` and `testConnectionOnStartup()`, so that a
consumer stopped and started again
+(with `asyncStopListener=true` or `asyncStartListener=true`) always works on
its own listener container. The methods
+are still there. A subclass of `JmsConsumer` (created by a custom
`JmsEndpoint.createConsumer`) that overrides them
+to change how the listener container is started or stopped should override
`doStart` and `doStop` instead.
+
=== camel-jms - preserveMessageQos uses the time to live of the endpoint for a
message without expiration
With `preserveMessageQos=true`, a message without a `JMSExpiration` (header
missing or 0) was sent with no time to
