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 eef3ea98e65d CAMEL-25312: camel-sjms - a suspended consumer must stop
receiving, and a restarted consumer must receive again (#27388)
eef3ea98e65d is described below
commit eef3ea98e65ddd9b39acf8ed6d5c32aaa74ede0b
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Oct 6 14:52:06 2026 +0530
CAMEL-25312: camel-sjms - a suspended consumer must stop receiving, and a
restarted consumer must receive again (#27388)
* CAMEL-25312: camel-sjms - a suspended consumer must stop receiving, and a
restarted consumer must receive again
SjmsConsumer implements Suspendable but has no doSuspend/doResume, so
suspending it was a no-op: the listener container kept receiving and routing
messages while the route was reported as suspended. This affects a graceful
shutdown (the shutdown strategy suspends a Suspendable consumer and then
waits
for the inflight exchanges, while the consumer keeps taking new messages
until
the shutdown timeout), suspendRoute (route controller, JMX) and the route
policies (ThrottlingInflightRoutePolicy, ThrottlingExceptionRoutePolicy,
...).
CAMEL-9567 fixed this for the old consumer; the consumer rewritten by
CAMEL-15995 (3.8) lost it.
SimpleMessageListenerContainer.stopConsumers closes the consumers and
sessions
but keeps them in their fields, and initConsumers only creates consumers
when
the field is null. A container started again (the same consumer instance
stopped and started, e.g. with the JMX operations of the consumer) kept the
closed consumers and never received a message again.
The consumer now suspends and resumes its listener container, which stops
and
starts its connection: the delivery stops (after the listeners in progress
returned), and the consumers and sessions stay open for the exchanges in
flight (an async consumer sends its reply with the session when the exchange
is done). The container forgets its consumers and sessions when it closes
them,
and on shutdown after closing the connection, as the connection recovery may
have created new ones in between.
Connection.stop() waits for the message listeners in progress, and a
message listener must not call it
(the JMS provider throws an IllegalStateException, which stopConnection()
ignored). A route policy
suspends the consumer when an exchange is done: for a synchronous consumer
on the thread of the message
listener, so the suspend did nothing, or on a thread the listener waits for
(route continued with
threads()), and a graceful shutdown or JMX suspend that waited in
Connection.stop() held the lock of the
consumer that the listener's route policy then needed. Both waited until
Artemis gave up after its
onMessageCloseTimeout (10 s).
The container now stops and starts its connection on a single thread, in
the order of the suspend and
resume calls, and suspend/resume do not wait. The upgrade note is next to
the other camel-sjms note and
says which connection factories ignore Connection.stop().
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25312: camel-sjms - suspend and resume only a started listener
container, and shorter upgrade note
A listener container that is not started yet (autoStartup=false, or
asyncStartListener=true before the start task
runs) was suspended and resumed anyway: a resume then marked it started
without a connection, so it could never be
started (a regression of the previous commit), and a container started
after the suspend received messages while
the consumer was suspended. Suspend and resume now only act on a started
(or suspended) container, and
startListenerContainer suspends the container it starts when the consumer
is suspended.
Review nits: the upgrade note is shorter, the control test that passes on
main is dropped, and the test-only methods
say so in their javadoc.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25312: camel-sjms - Give the suspend note its own upgrade-guide
heading
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/component/sjms/SjmsConsumer.java | 34 +++
.../consumer/SimpleMessageListenerContainer.java | 68 +++++
.../sjms/consumer/SjmsConsumerRestartTest.java | 313 +++++++++++++++++++++
.../sjms/consumer/SjmsConsumerRoutePolicyTest.java | 162 +++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 9 +
5 files changed, 586 insertions(+)
diff --git
a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsConsumer.java
b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsConsumer.java
index e567232cfaa7..badfe1f93e86 100644
---
a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsConsumer.java
+++
b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsConsumer.java
@@ -21,6 +21,7 @@ import jakarta.jms.Connection;
import org.apache.camel.Endpoint;
import org.apache.camel.FailedToCreateConsumerException;
import org.apache.camel.Processor;
+import org.apache.camel.StatefulService;
import org.apache.camel.Suspendable;
import org.apache.camel.support.DefaultConsumer;
import org.apache.camel.support.service.ServiceHelper;
@@ -104,6 +105,27 @@ public class SjmsConsumer extends DefaultConsumer
implements Suspendable {
super.doStop();
}
+ @Override
+ protected void doSuspend() throws Exception {
+ // stop receiving messages while suspended. A listener container that
is not started yet (autoStartup=false,
+ // or asyncStartListener=true and the start task did not run yet) is
left alone: startListenerContainer
+ // suspends it when it starts it
+ if (listenerContainer instanceof StatefulService container &&
container.isStarted()) {
+ container.suspend();
+ }
+ }
+
+ @Override
+ protected void doResume() throws Exception {
+ if (listenerContainer instanceof StatefulService container &&
container.isSuspended()) {
+ container.resume();
+ }
+ }
+
+ public MessageListenerContainer getListenerContainer() {
+ return listenerContainer;
+ }
+
protected void prepareAndStartListenerContainer() {
listenerContainer.afterPropertiesConfigured(getEndpoint().getCamelContext());
@@ -149,6 +171,18 @@ public class SjmsConsumer extends DefaultConsumer
implements Suspendable {
LOG.trace("Starting listener container {} on destination {}",
listenerContainer, getDestinationName());
ServiceHelper.startService(listenerContainer);
LOG.debug("Started listener container {} on destination {}",
listenerContainer, getDestinationName());
+ // the consumer may have been suspended while its listener container
was not started (doSuspend left it alone)
+ lock.lock();
+ try {
+ if (isSuspendingOrSuspended() && listenerContainer instanceof
StatefulService container
+ && container.isStarted()) {
+ LOG.debug("Suspending listener container {} on destination {},
as the consumer is suspended",
+ listenerContainer, getDestinationName());
+ container.suspend();
+ }
+ } finally {
+ lock.unlock();
+ }
}
protected void stopAndDestroyListenerContainer() {
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 e21f401359b8..982e67948741 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
@@ -19,6 +19,7 @@ package org.apache.camel.component.sjms.consumer;
import java.time.Duration;
import java.util.HashSet;
import java.util.Set;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.locks.Lock;
@@ -71,6 +72,10 @@ public class SimpleMessageListenerContainer extends
ServiceSupport
private ScheduledExecutorService recoverPool;
private BackgroundTask recoverTask;
private Future<?> recoverFuture;
+ // stops and starts the connection when the container is suspended and
resumed, in that order (guarded by the
+ // lifecycle lock of the container: suspend, resume and stop)
+ private ExecutorService suspendExecutor;
+ private volatile Future<?> suspendTask;
public SimpleMessageListenerContainer(SjmsEndpoint endpoint) {
this.endpoint = endpoint;
@@ -265,6 +270,12 @@ public class SimpleMessageListenerContainer extends
ServiceSupport
@Override
protected void doStop() throws Exception {
+ if (suspendExecutor != null) {
+ // let a pending stop or start of the connection finish first
+
endpoint.getCamelContext().getExecutorServiceManager().shutdownGraceful(suspendExecutor);
+ suspendExecutor = null;
+ suspendTask = null;
+ }
stopConnection();
stopConsumers();
if (recoverPool != null) {
@@ -281,10 +292,64 @@ public class SimpleMessageListenerContainer extends
ServiceSupport
}
}
+ @Override
+ protected void doSuspend() throws Exception {
+ // stop the delivery of messages; the consumers and sessions stay
open, as the exchanges in flight may still
+ // use them (e.g. to send the reply of an async consumer).
Connection.stop() waits for the message listeners
+ // in progress, and must not be called by a message listener (the JMS
provider throws an
+ // IllegalStateException): a route policy suspends the consumer when
an exchange is done, on the thread of
+ // the message listener or on a thread that the listener waits for,
and the suspend must not wait for the
+ // listeners while it holds the lock of the consumer. So another
thread stops the connection
+ submitSuspendTask(this::stopConnection);
+ }
+
+ @Override
+ protected void doResume() throws Exception {
+ // after a pending stop of the connection
+ submitSuspendTask(() -> {
+ try {
+ startConnection();
+ } catch (Exception e) {
+ LOG.warn("Error starting the JMS connection on resume. This
exception is ignored.", e);
+ }
+ });
+ }
+
+ private void submitSuspendTask(Runnable task) {
+ if (suspendExecutor == null) {
+ suspendExecutor =
endpoint.getCamelContext().getExecutorServiceManager().newSingleThreadExecutor(this,
+ "SjmsSuspendResume");
+ }
+ suspendTask = suspendExecutor.submit(task);
+ }
+
+ /**
+ * Whether a stop or start of the connection requested by a suspend or
resume is still pending. Only for tests.
+ */
+ boolean isSuspendResumePending() {
+ Future<?> task = suspendTask;
+ return task != null && !task.isDone();
+ }
+
+ /**
+ * Whether the connection is started (false as soon as a stop of the
connection began). Only for tests.
+ */
+ boolean isConnectionStarted() {
+ return connectionStarted;
+ }
+
@Override
protected void doShutdown() throws Exception {
closeConnection(connection);
this.connection = null;
+ // the connection recovery may have created consumers after
stopConsumers, they are closed with the connection
+ consumerLock.lock();
+ try {
+ this.consumers = null;
+ this.sessions = null;
+ } finally {
+ consumerLock.unlock();
+ }
}
protected void initConsumers() throws Exception {
@@ -330,6 +395,9 @@ public class SimpleMessageListenerContainer extends
ServiceSupport
closeSession(session);
}
}
+ // so that a restart creates new consumers
+ this.consumers = null;
+ this.sessions = null;
}
} finally {
consumerLock.unlock();
diff --git
a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/SjmsConsumerRestartTest.java
b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/SjmsConsumerRestartTest.java
new file mode 100644
index 000000000000..8fb7f60756f3
--- /dev/null
+++
b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/SjmsConsumerRestartTest.java
@@ -0,0 +1,313 @@
+/*
+ * 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.sjms.consumer;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import jakarta.jms.Connection;
+import jakarta.jms.ConnectionFactory;
+import jakarta.jms.JMSContext;
+import jakarta.jms.JMSException;
+
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.Consumer;
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.component.sjms.SjmsConsumer;
+import org.apache.camel.component.sjms.support.JmsTestSupport;
+import org.apache.camel.support.service.ServiceHelper;
+import org.apache.camel.support.service.ServiceSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A suspended sjms route must not consume, and must consume again when
resumed. The route policies (throttling inflight
+ * / exception, master, scheduled) suspend and resume the consumer, and JMX
can stop and start the same consumer
+ * instance. The listener container may start after the suspend
(asyncStartListener, autoStartup=false).
+ */
+class SjmsConsumerRestartTest extends JmsTestSupport {
+
+ private static final String QUEUE = "sjms:queue:SjmsConsumerRestartTest.";
+
+ private final CountDownLatch entered = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+ private final AtomicInteger processed = new AtomicInteger();
+ private final CountDownLatch connectionEntered = new CountDownLatch(1);
+ // holds the next connection created with the gated connection factory
+ private volatile CountDownLatch connectionGate;
+
+ @BindToRegistry("gatedConnectionFactory")
+ private final ConnectionFactory gatedConnectionFactory = new
ConnectionFactory() {
+ @Override
+ public Connection createConnection() throws JMSException {
+ CountDownLatch gate = connectionGate;
+ connectionGate = null;
+ if (gate != null) {
+ connectionEntered.countDown();
+ try {
+ assertTrue(gate.await(20, TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ return connectionFactory.createConnection();
+ }
+
+ @Override
+ public Connection createConnection(String userName, String password)
throws JMSException {
+ return connectionFactory.createConnection(userName, password);
+ }
+
+ @Override
+ public JMSContext createContext() {
+ return connectionFactory.createContext();
+ }
+
+ @Override
+ public JMSContext createContext(String userName, String password) {
+ return connectionFactory.createContext(userName, password);
+ }
+
+ @Override
+ public JMSContext createContext(String userName, String password, int
sessionMode) {
+ return connectionFactory.createContext(userName, password,
sessionMode);
+ }
+
+ @Override
+ public JMSContext createContext(int sessionMode) {
+ return connectionFactory.createContext(sessionMode);
+ }
+ };
+
+ @Test
+ void testSuspendResumeRoute() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:suspend");
+ mock.expectedBodiesReceived("Hello World");
+ template.sendBody(QUEUE + "suspend", "Hello World");
+ mock.assertIsSatisfied();
+
+ context.getRouteController().suspendRoute("suspend");
+ assertEquals(ServiceStatus.Suspended,
context.getRouteController().getRouteStatus("suspend"));
+ // the connection is stopped by another thread
+ awaitSuspendResume("suspend");
+
+ mock.reset();
+ mock.expectedMessageCount(0);
+ mock.setAssertPeriod(1000);
+ template.sendBody(QUEUE + "suspend", "Bye World");
+ mock.assertIsSatisfied();
+
+ mock.reset();
+ mock.expectedBodiesReceived("Bye World");
+ context.getRouteController().resumeRoute("suspend");
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void testSuspendResumeConsumer() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:policy");
+
+ // what RoutePolicySupport.suspendOrStopConsumer and
resumeOrStartConsumer do
+ Consumer consumer = context.getRoute("policy").getConsumer();
+ ServiceHelper.suspendService(consumer);
+ awaitSuspendResume("policy");
+
+ mock.expectedMessageCount(0);
+ mock.setAssertPeriod(1000);
+ template.sendBody(QUEUE + "policy", "Hello World");
+ mock.assertIsSatisfied();
+
+ mock.reset();
+ mock.expectedBodiesReceived("Hello World", "Bye World");
+ ServiceHelper.resumeService(consumer);
+ template.sendBody(QUEUE + "policy", "Bye World");
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void testStopStartConsumer() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:restart");
+
+ // the same consumer instance, as with the stop and start operations
of the managed consumer (JMX)
+ Consumer consumer = context.getRoute("restart").getConsumer();
+ for (int i = 0; i < 2; i++) {
+ ServiceHelper.stopService(consumer);
+ ServiceHelper.startService(consumer);
+ }
+
+ mock.expectedBodiesReceived("Hello World", "Bye World");
+ template.sendBody(QUEUE + "restart", "Hello World");
+ template.sendBody(QUEUE + "restart", "Bye World");
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void testGracefulStopTakesNoNewMessage() throws Exception {
+ // the graceful shutdown suspends the consumer, then waits for the
inflight exchange
+ template.sendBody(QUEUE + "shutdown", "A");
+ assertTrue(entered.await(20, TimeUnit.SECONDS));
+
+ CompletableFuture<Void> stop = CompletableFuture.runAsync(() -> {
+ try {
+ context.getRouteController().stopRoute("shutdown", 20,
TimeUnit.SECONDS);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ ServiceSupport consumer = (ServiceSupport)
context.getRoute("shutdown").getConsumer();
+ SimpleMessageListenerContainer container = container("shutdown");
+ // the stop of the connection begins (and then waits for the listener
of A)
+ await().atMost(20, TimeUnit.SECONDS)
+ .until(() -> consumer.isSuspendingOrSuspended() &&
!container.isConnectionStarted());
+ template.sendBody(QUEUE + "shutdown", "B");
+ release.countDown();
+ stop.get(30, TimeUnit.SECONDS);
+
+ assertEquals(ServiceStatus.Stopped,
context.getRouteController().getRouteStatus("shutdown"));
+ assertEquals(1, processed.get(), "A message sent after the route began
to stop must not be consumed");
+ }
+
+ @Test
+ void testGracefulStopAsyncReply() throws Exception {
+ // with asyncConsumer the message listener returns before the exchange
is done, and the reply is sent with the
+ // session of the consumer when it is done: suspending the consumer
must keep that session open
+ CompletableFuture<Object> reply = template.asyncRequestBody(QUEUE +
"async?requestTimeout=10000", "A");
+ assertTrue(entered.await(20, TimeUnit.SECONDS));
+
+ CompletableFuture<Void> stop = CompletableFuture.runAsync(() -> {
+ try {
+ context.getRouteController().stopRoute("async", 20,
TimeUnit.SECONDS);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ ServiceSupport consumer = (ServiceSupport)
context.getRoute("async").getConsumer();
+ await().atMost(20,
TimeUnit.SECONDS).until(consumer::isSuspendingOrSuspended);
+ release.countDown();
+
+ assertEquals("Bye A", reply.get(30, TimeUnit.SECONDS));
+ stop.get(30, TimeUnit.SECONDS);
+ }
+
+ @Test
+ void testSuspendedBeforeAsyncStart() throws Exception {
+ // the asynchronous start of the listener container is held while it
tests the connection
+ CountDownLatch proceed = new CountDownLatch(1);
+ connectionGate = proceed;
+ context.addRoutes(new RouteBuilder() {
+ public void configure() {
+ from(QUEUE +
"asyncstart?asyncStartListener=true&testConnectionOnStartup=true"
+ + "&connectionFactory=#gatedConnectionFactory")
+ .routeId("asyncstart").to("mock:asyncstart");
+ }
+ });
+ assertTrue(connectionEntered.await(20, TimeUnit.SECONDS));
+ context.getRouteController().suspendRoute("asyncstart");
+ proceed.countDown();
+
+ // the container starts after the suspend, and must then be suspended
too
+ SimpleMessageListenerContainer container = container("asyncstart");
+ await().atMost(20, TimeUnit.SECONDS)
+ .until(() -> container.isSuspended() &&
!container.isSuspendResumePending());
+ MockEndpoint mock = getMockEndpoint("mock:asyncstart");
+ mock.expectedMessageCount(0);
+ mock.setAssertPeriod(1000);
+ template.sendBody(QUEUE + "asyncstart", "Hello World");
+ mock.assertIsSatisfied();
+
+ mock.reset();
+ mock.expectedBodiesReceived("Hello World");
+ context.getRouteController().resumeRoute("asyncstart");
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void testSuspendResumeBeforeListenerStarted() throws Exception {
+ // autoStartup=false: the consumer starts without its listener
container, which is started later
+ context.getRouteController().suspendRoute("manual");
+ context.getRouteController().resumeRoute("manual");
+ ((SjmsConsumer)
context.getRoute("manual").getConsumer()).startListenerContainer();
+
+ MockEndpoint mock = getMockEndpoint("mock:manual");
+ mock.expectedBodiesReceived("Hello World");
+ template.sendBody(QUEUE + "manual", "Hello World");
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void testListenerStartedWhileSuspended() throws Exception {
+ // autoStartup=false, and the listener container is started while the
route is suspended
+ context.getRouteController().suspendRoute("manualsuspended");
+ ((SjmsConsumer)
context.getRoute("manualsuspended").getConsumer()).startListenerContainer();
+ awaitSuspendResume("manualsuspended");
+
+ MockEndpoint mock = getMockEndpoint("mock:manualsuspended");
+ mock.expectedMessageCount(0);
+ mock.setAssertPeriod(1000);
+ template.sendBody(QUEUE + "manualsuspended", "Hello World");
+ mock.assertIsSatisfied();
+
+ mock.reset();
+ mock.expectedBodiesReceived("Hello World");
+ context.getRouteController().resumeRoute("manualsuspended");
+ mock.assertIsSatisfied();
+ }
+
+ private SimpleMessageListenerContainer container(String routeId) {
+ return (SimpleMessageListenerContainer) ((SjmsConsumer)
context.getRoute(routeId).getConsumer())
+ .getListenerContainer();
+ }
+
+ private void awaitSuspendResume(String routeId) {
+ SimpleMessageListenerContainer container = container(routeId);
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
!container.isSuspendResumePending());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ public void configure() {
+ from(QUEUE + "suspend").routeId("suspend").to("mock:suspend");
+ from(QUEUE + "policy").routeId("policy").to("mock:policy");
+ from(QUEUE + "restart").routeId("restart").to("mock:restart");
+ from(QUEUE +
"manual?autoStartup=false").routeId("manual").to("mock:manual");
+ from(QUEUE +
"manualsuspended?autoStartup=false").routeId("manualsuspended").to("mock:manualsuspended");
+ from(QUEUE + "async?asyncConsumer=true").routeId("async")
+ .threads(1)
+ .process(e -> {
+ entered.countDown();
+ release.await(20, TimeUnit.SECONDS);
+ })
+ .transform(body().prepend("Bye "));
+ from(QUEUE + "shutdown").routeId("shutdown")
+ .process(e -> {
+ processed.incrementAndGet();
+ entered.countDown();
+ release.await(20, TimeUnit.SECONDS);
+ });
+ }
+ };
+ }
+
+}
diff --git
a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/SjmsConsumerRoutePolicyTest.java
b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/SjmsConsumerRoutePolicyTest.java
new file mode 100644
index 000000000000..b05e678c5470
--- /dev/null
+++
b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/SjmsConsumerRoutePolicyTest.java
@@ -0,0 +1,162 @@
+/*
+ * 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.sjms.consumer;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.component.sjms.SjmsConsumer;
+import org.apache.camel.component.sjms.support.JmsTestSupport;
+import org.apache.camel.support.service.ServiceHelper;
+import org.apache.camel.support.service.ServiceSupport;
+import org.apache.camel.throttling.ThrottlingExceptionRoutePolicy;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A route policy suspends the consumer when an exchange is done, which for a
synchronous sjms consumer runs on the
+ * thread of the JMS message listener, or on a thread the listener waits for.
Connection.stop() waits for the message
+ * listeners in progress and a message listener must not call it (the JMS
provider throws an IllegalStateException), so
+ * the consumer must stop receiving without the suspend waiting for the
listener.
+ */
+class SjmsConsumerRoutePolicyTest extends JmsTestSupport {
+
+ private static final String QUEUE =
"sjms:queue:SjmsConsumerRoutePolicyTest";
+
+ private final ThrottlingExceptionRoutePolicy policy = new
ThrottlingExceptionRoutePolicy(1, 60000, 60000, null, false);
+ private final ThrottlingExceptionRoutePolicy blockingPolicy
+ = new ThrottlingExceptionRoutePolicy(1, 60000, 60000, null, false);
+ private final ThrottlingExceptionRoutePolicy threadsPolicy
+ = new ThrottlingExceptionRoutePolicy(1, 60000, 60000, null, false);
+ private final CountDownLatch entered = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+
+ @Test
+ void testCircuitBreakerSuspendsConsumer() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ Consumer consumer = context.getRoute("breaker").getConsumer();
+ SimpleMessageListenerContainer container
+ = (SimpleMessageListenerContainer) ((SjmsConsumer)
consumer).getListenerContainer();
+
+ // the failure opens the circuit on the thread of the message listener
+ template.sendBody(QUEUE, "Kaboom");
+ await().atMost(20, TimeUnit.SECONDS)
+ .until(() -> ((ServiceSupport) consumer).isSuspended() &&
!container.isSuspendResumePending());
+
+ mock.expectedMessageCount(0);
+ mock.setAssertPeriod(1000);
+ template.sendBody(QUEUE, "Hello World");
+ mock.assertIsSatisfied();
+
+ mock.reset();
+ mock.expectedBodiesReceived("Hello World");
+ ServiceHelper.resumeService(consumer);
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void testSuspendWhileThePolicySuspends() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:blocking");
+ Consumer consumer = context.getRoute("blocking").getConsumer();
+ SimpleMessageListenerContainer container
+ = (SimpleMessageListenerContainer) ((SjmsConsumer)
consumer).getListenerContainer();
+
+ template.sendBody(QUEUE + "Blocking", "Kaboom");
+ assertTrue(entered.await(20, TimeUnit.SECONDS));
+
+ // the consumer is suspended (graceful shutdown, JMX) while the
exchange is in flight, then the exchange fails
+ // and the policy opens the circuit, which suspends the consumer on
the thread of the message listener
+ CompletableFuture<Boolean> suspend = CompletableFuture.supplyAsync(()
-> ServiceHelper.suspendService(consumer));
+ await().atMost(20, TimeUnit.SECONDS).until(((ServiceSupport)
consumer)::isSuspendingOrSuspended);
+ release.countDown();
+ // must not wait for the message listener, which waits for the lock of
the consumer
+ assertTrue(suspend.get(5, TimeUnit.SECONDS));
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
!container.isSuspendResumePending());
+
+ mock.expectedMessageCount(0);
+ mock.setAssertPeriod(1000);
+ template.sendBody(QUEUE + "Blocking", "Hello World");
+ mock.assertIsSatisfied();
+
+ mock.reset();
+ mock.expectedBodiesReceived("Hello World");
+ ServiceHelper.resumeService(consumer);
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void testCircuitBreakerOnAnotherThread() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:threads");
+ Consumer consumer = context.getRoute("threads").getConsumer();
+ SimpleMessageListenerContainer container
+ = (SimpleMessageListenerContainer) ((SjmsConsumer)
consumer).getListenerContainer();
+
+ // the route continues on another thread, which opens the circuit
while the message listener waits for it
+ template.sendBody(QUEUE + "Threads", "Kaboom");
+ await().atMost(5, TimeUnit.SECONDS)
+ .until(() -> ((ServiceSupport) consumer).isSuspended() &&
!container.isSuspendResumePending());
+
+ mock.expectedMessageCount(0);
+ mock.setAssertPeriod(1000);
+ template.sendBody(QUEUE + "Threads", "Hello World");
+ mock.assertIsSatisfied();
+
+ mock.reset();
+ mock.expectedBodiesReceived("Hello World");
+ ServiceHelper.resumeService(consumer);
+ mock.assertIsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ public void configure() {
+ from(QUEUE).routeId("breaker").routePolicy(policy)
+ .process(e -> {
+ if
("Kaboom".equals(e.getMessage().getBody(String.class))) {
+ throw new IllegalArgumentException("Forced");
+ }
+ })
+ .to("mock:result");
+ from(QUEUE +
"Blocking").routeId("blocking").routePolicy(blockingPolicy)
+ .process(e -> {
+ if
("Kaboom".equals(e.getMessage().getBody(String.class))) {
+ entered.countDown();
+ release.await(20, TimeUnit.SECONDS);
+ throw new IllegalArgumentException("Forced");
+ }
+ })
+ .to("mock:blocking");
+ from(QUEUE +
"Threads").routeId("threads").routePolicy(threadsPolicy)
+ .threads(1)
+ .process(e -> {
+ if
("Kaboom".equals(e.getMessage().getBody(String.class))) {
+ throw new IllegalArgumentException("Forced");
+ }
+ })
+ .to("mock:threads");
+ }
+ };
+ }
+
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
index 03ddbbe59e7c..dce46b5c5d5b 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
@@ -1491,6 +1491,15 @@ timeout or the reply, and the send failure is logged at
WARN level.
which `ReplyManagerSupport` implements. A custom `ReplyManager` implementation
that does not override it keeps the
previous behaviour.
+=== camel-sjms - suspending a route stops the consumer
+
+A suspended sjms (or sjms2) consumer used to keep receiving and routing
messages, also during a graceful shutdown and
+under route policies such as `ThrottlingInflightRoutePolicy`. It now stops the
JMS connection of its listener container
+shortly after the suspend (the connection is stopped by another thread), and
starts it again on resume. Messages that
+the JMS client has already prefetched are delivered after the resume. A
connection factory that shares one connection
+and does not stop it on `Connection.stop()` (such as the pooled-jms
`JmsPoolConnectionFactory`) keeps delivering to a
+suspended consumer, as before.
+
=== camel-tika
The Tika dependency has been upgraded from 3.x to 4.x. Tika 4 removed the
`TikaConfig` class and XML
