shashank created CAMEL-25312:
--------------------------------
Summary: camel-sjms - a suspended consumer keeps consuming
(graceful shutdown, suspendRoute, route policies), and a consumer started again
receives nothing
Key: CAMEL-25312
URL: https://issues.apache.org/jira/browse/CAMEL-25312
Project: Camel
Issue Type: Bug
Components: camel-sjms, camel-sjms2
Reporter: shashank
{{SjmsConsumer}} implements {{Suspendable}} but has no
{{doSuspend}}/{{doResume}}, so {{suspend()}} only changes its status: the
listener container keeps receiving and routing messages while the route is
reported as suspended. Everything that suspends a consumer is affected:
* the graceful shutdown (default): {{DefaultShutdownStrategy}} suspends a
{{Suspendable}} consumer instead of stopping it, defers its shutdown and waits
until no exchange is inflight. The sjms consumer keeps taking new messages
meanwhile, so with a steady flow the route (or the CamelContext) only stops at
the shutdown timeout, and the exchanges inflight then are forced. This is what
CAMEL-9567 fixed in 2016 for the old consumer.
* {{suspendRoute}} (route controller, JMX) and the route policies that suspend
the consumer ({{ThrottlingInflightRoutePolicy}},
{{ThrottlingExceptionRoutePolicy}} with an open circuit, ...): nothing is
throttled, an open circuit keeps routing (failing) messages.
Separately, {{SimpleMessageListenerContainer.stopConsumers}} closes the
consumers and sessions but keeps them in the {{consumers}}/{{sessions}} fields,
and {{initConsumers}} only creates consumers when {{consumers == null}}. A
container that is started again therefore keeps its closed consumers and never
receives a message: this happens when the same consumer instance is stopped and
started (the stop/start operations of the managed consumer in JMX; a route
restart creates a new consumer and is not affected).
h3. Reproduction
{{SjmsConsumerRestartTest}} (camel-sjms, embedded Artemis), on main:
{noformat}
testSuspendResumeRoute: suspendRoute("suspend"), send a message
mock://suspend Received message count. Expected: <0> but was: <1>
testSuspendResumeConsumer: ServiceHelper.suspendService(consumer) as the route
policies do
mock://policy Received message count. Expected: <0> but was: <1>
testGracefulStopTakesNoNewMessage: message A blocks in the route, stopRoute
(graceful) begins and suspends
the consumer, the test waits for the connection to stop, then sends message B
and releases A
ConditionTimeout: the connection is never stopped within 20 seconds
(before the waits for the stop thread were added: A message sent after the
route began to stop must not be
consumed ==> expected: <1> but was: <2>)
testStopStartConsumer: the consumer instance stopped and started twice, two
messages sent
mock://restart Received message count. Expected: <2> but was: <0>
{noformat}
{{SjmsConsumerRoutePolicyTest}} (the route policy suspends the consumer when an
exchange is done), on main:
{noformat}
testCircuitBreakerSuspendsConsumer: a failure opens the circuit
(ThrottlingExceptionRoutePolicy) on the listener thread
mock://result Received message count. Expected: <0> but was: <1>
testSuspendWhileThePolicySuspends: suspendService(consumer) while the exchange
is in flight, then the policy opens the circuit
mock://blocking Received message count. Expected: <0> but was: <1>
testCircuitBreakerOnAnotherThread: the same with the route continued by
threads(1)
mock://threads Received message count. Expected: <0> but was: <1>
{noformat}
(These three tests and the waits of {{SjmsConsumerRestartTest}} use two
package-private hooks of the container ({{isSuspendResumePending}},
{{isConnectionStarted}}) and a getter on the consumer
({{getListenerContainer}}), so on main they ran with those three methods
added.) The control {{testStopStartRoute}}, a route restart, passes;
{{testGracefulStopAsyncReply}} guards the fix: an {{asyncConsumer}} InOut
exchange in flight during the graceful stop must still send its reply with the
consumer's session. No sleeps: latches, {{MockEndpoint}} assert period,
Awaitility.
The defect was found with a TLA+ model of the consumer and its listener
container (route start/stop/suspend/resume of one consumer instance, the
connection recovery task, each Java statement one step): "a suspended route
receives nothing" is violated in 5 steps (start, suspend), and "a started route
receives messages" in 13 steps for start, stop, start (the closed consumers are
kept). With both parts of the fix all properties hold, also with connection
failures and recovery during the operations; the model also showed that
{{doShutdown}} must forget the consumers too (the recovery may create consumers
after {{stopConsumers}}). That model makes the suspend one atomic step on a
management thread; a second model of the threads (message listener, route
policy, management thread, the container's stop/start thread, a {{threads()}}
thread; the JMS rule that the listener cannot stop its connection and that
{{Connection.stop()}} waits for it; the consumer lock) shows the ignored stop
on the listener thread and the two waiting cycles of a suspend on the calling
thread, and that a single ordered stop/start thread has none (TLC deadlock
check on; a mutation that starts the connection on the calling thread while a
stop is queued is caught).
h3. Proposed fix
* {{SjmsConsumer.doSuspend}}/{{doResume}} suspend/resume the listener
container, whose {{doSuspend}} stops the connection ({{Connection.stop()}}: no
message is delivered any more) and {{doResume}} starts it again. The consumers
and sessions stay open, as exchanges in flight may still use them (an
{{asyncConsumer}} sends the reply with the consumer's session when the exchange
is done); closing them on suspend breaks that, which the guard test shows.
* The connection is stopped and started by a single thread of the container
({{SjmsSuspendResume}}), in the order of the suspend and resume calls, and
suspend/resume do not wait for it. {{Connection.stop()}} waits for the message
listeners in progress and must not be called by a message listener of the
connection (JMS 2.0; Artemis throws {{IllegalStateException}}), while a route
policy suspends the consumer when an exchange is done: for a synchronous
consumer on the listener thread itself (a suspend on the calling thread did
nothing there: the exception was ignored and the consumer kept receiving), or
on a thread the listener waits for (route continued with {{threads()}}), and a
graceful shutdown or JMX suspend waiting in {{Connection.stop()}} would hold
the lock of the consumer that the listener's route policy needs. Both cycles
waited until Artemis gave up after its {{onMessageCloseTimeout}} (10 s,
"AMQ212002: Timed out after waiting 10000ms for handler to complete
processing"). {{doStop}} lets a pending stop/start finish first.
* {{SimpleMessageListenerContainer.stopConsumers}} sets {{consumers}} and
{{sessions}} to null after closing them; {{doShutdown}} does the same after
closing the connection (the connection recovery may have created consumers
after {{stopConsumers}}; the model found this).
Limits: messages the JMS client has prefetched stay with the stopped connection
until it is started again or closed. A connection factory that hands out a
shared connection and does not stop it ({{JmsPoolConnectionFactory}} of
pooled-jms: {{stop()}} is a no-op; Spring's {{SingleConnectionFactory}} only
stops the shared connection when the last user stops it) keeps delivering to a
suspended consumer, as today. Stopping the connection of one consumer never
suspends another route: each listener container creates its own connection, and
the pooling factories above do not stop a shared one.
Upgrade guide note (next to the other camel-sjms note): a suspended sjms route
now stops consuming, and the limits above.
With the fix the new tests and the camel-sjms (132 tests, 5 skipped) and
camel-sjms2 (23 tests) suites pass; {{MllpTcpServerConsumerTransactionTest}}
(camel-mllp, uses sjms) is disabled ({{@Disabled}}, 2 skipped).
Affected: 4.14.x, 4.18.x and main (same code, GitHub contents API), since 3.8.0.
Duplicate check (2026-10-04): JIRA text "sjms" with "suspend" (CAMEL-9567 only,
the old consumer), "sjms" with restart/stopRoute/startRoute (none),
"SimpleMessageListenerContainer" (CAMEL-22026, thread leak on stop, fixed by
closing the connection on shutdown; not this), sjms issues since 2024 (10,
none). GitHub pull requests "sjms suspend" (none); open PR #27173 (batch
consumer) changes {{SimpleMessageListenerContainer}} in other places.
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
