allthingssecurity opened a new pull request, #27388: URL: https://github.com/apache/camel/pull/27388
# Description [CAMEL-25312](https://issues.apache.org/jira/browse/CAMEL-25312) `SjmsConsumer` implements `Suspendable` but has no `doSuspend`/`doResume`, so suspending it only changes its status: the listener container keeps receiving and routing messages while the route is reported as suspended. This affects the graceful shutdown (the shutdown strategy suspends a `Suspendable` consumer, defers its shutdown and waits for the inflight exchanges, while the sjms consumer keeps taking new messages until the shutdown timeout), `suspendRoute` (route controller, JMX) and the route policies that suspend the consumer (`ThrottlingInflightRoutePolicy`, `ThrottlingExceptionRoutePolicy`, ...). CAMEL-9567 fixed this for the old consumer; the rewrite in CAMEL-15995 (3.8.0) lost it. Also, `SimpleMessageListenerContainer.stopConsumers` closes the consumers and sessions but keeps them in their fields, and `initConsumers` only creates consumers when the field is null, so a container started again keeps the closed consumers and receives nothing. That happens when the same consumer instance is stopped and started (the JMX operations of the managed consumer). This change: - `SjmsConsumer.doSuspend`/`doResume` suspend/resume the listener container, whose `doSuspend` stops its connection (`Connection.stop()`: nothing 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 its reply with the consumer's session when the exchange is done); - the connection is stopped and started by a single thread of the container (`SjmsSuspendResume`), in the order of the suspend and resume calls; suspend and resume do not wait for it. `Connection.stop()` waits for the message listeners in progress and a message listener must not call it (Artemis throws `IllegalStateException`), while a route policy suspends the consumer when an exchange is done, on the listener thread of a synchronous consumer, or on a thread the listener waits for (`threads()`); and a graceful shutdown or JMX suspend that waited in `Connection.stop()` would hold the consumer lock the listener's route policy needs. A suspend on the calling thread was ignored in the first case and waited for Artemis' 10 s `onMessageCloseTimeout` in the other two; - `SimpleMessageListenerContainer` sets `consumers`/`sessions` to null when it closes them, and again in `doShutdown` after closing the connection (the connection recovery may have created consumers after `stopConsumers`); - `SjmsConsumer.getListenerContainer()` (as `JmsConsumer` has), used by the tests. Limits (in the upgrade note): messages already prefetched by the JMS client stay with the stopped connection until it is started again or closed, and a connection factory that shares a connection and ignores `Connection.stop()` (pooled-jms `JmsPoolConnectionFactory`; Spring's `SingleConnectionFactory` while another consumer uses the connection) keeps delivering to a suspended consumer, as today. A suspend never stops the connection of another route. The defect was found with a TLA+ model of the consumer and its container (start/stop/suspend/resume of one consumer instance against the connection recovery task, each Java statement one step): "a suspended route receives nothing" is violated right after start + suspend, and "a started route receives messages" after start, stop, start. With the fix all properties hold, also with connection failures and recovery in between; the model also showed that `doShutdown` must forget the consumers too. A second model of the threads (listener, route policy, management thread, the stop/start thread, a `threads()` thread, the consumer lock, the JMS rule on `Connection.stop()`) shows why the stop/start runs on its own thread, with TLC's deadlock check on. Upgrade guide: new section `=== camel-sjms - suspending a route stops the consumer`, next to the other camel-sjms section. Tests: new `SjmsConsumerRestartTest` and `SjmsConsumerRoutePolicyTest` (embedded Artemis, latches, `MockEndpoint` assert period, Awaitility). Without the main-code change (with the three test hooks added): ``` SjmsConsumerRestartTest.testSuspendResumeRoute:67 mock://suspend Received message count. Expected: <0> but was: <1> SjmsConsumerRestartTest.testSuspendResumeConsumer:87 mock://policy Received message count. Expected: <0> but was: <1> SjmsConsumerRestartTest.testGracefulStopTakesNoNewMessage:130 ConditionTimeout (the connection is never stopped) SjmsConsumerRestartTest.testStopStartConsumer:110 mock://restart Received message count. Expected: <2> but was: <0> SjmsConsumerRoutePolicyTest.testCircuitBreakerSuspendsConsumer:69 mock://result Received message count. Expected: <0> but was: <1> SjmsConsumerRoutePolicyTest.testSuspendWhileThePolicySuspends:99 mock://blocking Received message count. Expected: <0> but was: <1> SjmsConsumerRoutePolicyTest.testCircuitBreakerOnAnotherThread:122 mock://threads Received message count. Expected: <0> but was: <1> ``` `testStopStartRoute` (a route restart, which creates a new consumer) is a control and passes on main. `testGracefulStopAsyncReply` (an `asyncConsumer` InOut exchange in flight during the graceful stop must still reply) guards the fix: it fails when the suspend closes the consumers and sessions. With the change the camel-sjms (132 tests, 5 skipped) and camel-sjms2 (23 tests) suites pass. # Target - [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch) # Tracking - [x] If this is a large change, bug fix, or code improvement, I checked there is a [JIRA issue](https://issues.apache.org/jira/browse/CAMEL) filed for the change (usually before you start working on it). # Apache Camel coding standards and style - [x] I checked that each commit in the pull request has a meaningful subject line and body. - [ ] I have run `mvn clean install -DskipTests` locally from root folder and I have committed all auto-generated changes. (I built and tested the affected modules, including the formatter and import-sort plugins. I did not run the full root build.) # AI-assisted contributions - [x] If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., `Co-authored-by` trailers) and the PR description identifies the AI tool used. This PR was prepared with Claude Code (Claude Opus 5.5). The commit carries a `Co-Authored-By` trailer. _Claude Code on behalf of allthingssecurity_ 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
