allthingssecurity commented on code in PR #27388: URL: https://github.com/apache/camel/pull/27388#discussion_r4192068793
########## docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc: ########## @@ -1380,6 +1380,23 @@ 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 Review Comment: Done in 099426dd12cc: cut to the behaviour change, "shortly after the suspend" (the connection is stopped by another thread), and the prefetch and pooled-connection caveats. It is now a `====` subsection under the existing camel-sjms heading. _Claude Code on behalf of allthingssecurity_ ########## components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/SjmsConsumerRestartTest.java: ########## @@ -0,0 +1,209 @@ +/* + * 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 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. + */ +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(); + + @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 testStopStartRoute() throws Exception { Review Comment: Dropped `testStopStartRoute` in 099426dd12cc. _Claude Code on behalf of allthingssecurity_ ########## components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/SimpleMessageListenerContainer.java: ########## @@ -281,10 +292,64 @@ protected void doStop() throws Exception { } } + @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 + */ + boolean isSuspendResumePending() { Review Comment: Done in 099426dd12cc: the javadoc of `isSuspendResumePending()` and `isConnectionStarted()` now says "Only for tests". _Claude Code on behalf of allthingssecurity_ -- 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]
