[jira] [Created] (CAMEL-25312) camel-sjms - a suspended consumer keeps consuming (graceful shutdown, suspendRoute, route policies), and a consumer started again receives nothing

["shashank (Jira)" <ji...@apache.org>]

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)