This is an automated email from the ASF dual-hosted git repository. apupier pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/camel.git
commit 210bcdeac8ed6f4277c4e18e42eb9b8d98bad6b9 Author: smjain <[email protected]> AuthorDate: Wed Sep 23 18:35:38 2026 +0530 CAMEL-24951: camel-seda - do not wait for the pending messages of a suspended consumer when stopping Cause: a suspended SedaConsumer does not poll its queue, but graceful shutdown waits for it to drain: DefaultShutdownStrategy waits until getPendingExchangesSize() (the queue size) is 0, and SedaConsumer.prepareShutdown then waits for the poll threads, which in the suspended state only exit when shutdown is pending and the queue is empty. CAMEL-6390 only covered a suspended consumer with an empty queue. Effect: stopping a suspended SEDA route, or the CamelContext, when messages were sent to the route while it was suspended always runs into the shutdown timeout (45 s by default) and is then forced, or with abortAfterTimeout the stop fails and the route is restarted. Routes suspended by a RoutePolicy get into this state without user action. Fix: a suspending or suspended consumer reports no pending exchanges, and its poll threads exit as soon as shutdown is pending. The messages are kept on the queue (or purged with purgeWhenStopping) instead of the stop waiting for a drain that cannot happen. Co-Authored-By: Claude Opus 5.5 <[email protected]> --- .../apache/camel/component/seda/SedaConsumer.java | 12 ++- .../SedaSuspendedRouteWithPendingStopTest.java | 87 ++++++++++++++++++++++ 2 files changed, 96 insertions(+), 3 deletions(-) diff --git a/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaConsumer.java b/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaConsumer.java index 6715de6ab0f7..985e80020222 100644 --- a/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaConsumer.java +++ b/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaConsumer.java @@ -97,6 +97,11 @@ public class SedaConsumer extends DefaultConsumer implements Runnable, ShutdownA if (!suspendOnly && getEndpoint().isPurgeWhenStopping()) { getEndpoint().purgeQueue(); } + if (isSuspending() || isSuspended()) { + // a suspended consumer does not poll the queue, so do not wait for it to complete the pending exchanges + // (they are kept on the queue) + return 0; + } return getEndpoint().getQueue().size(); } @@ -179,10 +184,11 @@ public class SedaConsumer extends DefaultConsumer implements Runnable, ShutdownA // do not poll if we are suspended or starting again after resuming if (isSuspending() || isSuspended() || isStarting()) { - if (shutdownPending && queue.isEmpty()) { + if (shutdownPending) { LOG.trace( - "Consumer is suspended and shutdown is pending, so this consumer thread is breaking out because the task queue is empty."); - // we want to shutdown so break out if there queue is empty + "Consumer is suspended and shutdown is pending, so this consumer thread is breaking out."); + // we want to shutdown so break out, as a suspended consumer does not poll the task queue + // (any pending exchanges are kept on the queue) break; } else { LOG.trace("Consumer is suspended so skip polling"); diff --git a/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaSuspendedRouteWithPendingStopTest.java b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaSuspendedRouteWithPendingStopTest.java new file mode 100644 index 000000000000..1a86a0eea08c --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaSuspendedRouteWithPendingStopTest.java @@ -0,0 +1,87 @@ +/* + * 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.seda; + +import java.util.concurrent.TimeUnit; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.ServiceStatus; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Stopping a suspended seda route must not wait for the messages sent to it while it was suspended, as a suspended + * consumer does not consume them. + */ +public class SedaSuspendedRouteWithPendingStopTest extends ContextTestSupport { + + @Test + public void testStopSuspendedRouteWithPendingMessages() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:result"); + mock.expectedMessageCount(0); + + context.getRouteController().suspendRoute("foo"); + assertEquals(ServiceStatus.Suspended, context.getRouteController().getRouteStatus("foo")); + + template.sendBody("seda:start", "A"); + template.sendBody("seda:start", "B"); + template.sendBody("seda:start", "C"); + + // abort the stop if the graceful shutdown times out + boolean stopped = context.getRouteController().stopRoute("foo", 10, TimeUnit.SECONDS, true); + assertTrue(stopped, "Route should be stopped without waiting for the shutdown timeout"); + assertFalse(context.getShutdownStrategy().isTimeoutOccurred()); + assertEquals(ServiceStatus.Stopped, context.getRouteController().getRouteStatus("foo")); + + // the suspended route did not process the messages, they are kept on the queue + mock.assertIsSatisfied(); + assertEquals(3, context.getEndpoint("seda:start", SedaEndpoint.class).getQueue().size()); + + // and they are processed when the route is started again + mock.reset(); + mock.expectedBodiesReceived("A", "B", "C"); + context.getRouteController().startRoute("foo"); + mock.assertIsSatisfied(); + } + + @Test + public void testStopContextWithSuspendedRoute() throws Exception { + context.getRouteController().suspendRoute("foo"); + + template.sendBody("seda:start", "A"); + template.sendBody("seda:start", "B"); + + context.getShutdownStrategy().setTimeout(10); + context.stop(); + assertFalse(context.getShutdownStrategy().isTimeoutOccurred(), "Graceful shutdown should not time out"); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("seda:start?pollTimeout=100").routeId("foo").to("mock:result"); + } + }; + } +}
