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 0048a6264b9d08915128c7dd497cfa1177967034 Author: smjain <[email protected]> AuthorDate: Thu Sep 24 11:49:15 2026 +0530 CAMEL-24951: camel-seda - Keep waiting for the queue to be empty when a starting consumer is stopped A consumer that is starting will poll the queue, so it only breaks out once the queue is empty, as before. Only a suspended (or suspending) consumer breaks out without waiting. Also checks that no message was lost when CamelContext is stopped with a suspended route. Co-Authored-By: Claude Opus 5.5 <[email protected]> --- .../java/org/apache/camel/component/seda/SedaConsumer.java | 11 ++++++----- .../component/seda/SedaSuspendedRouteWithPendingStopTest.java | 8 ++++++++ 2 files changed, 14 insertions(+), 5 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 985e80020222..28ef5864bdb0 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 @@ -184,11 +184,12 @@ 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) { - LOG.trace( - "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) + // a suspended consumer does not poll the task queue, so break out without waiting for the queue to be + // empty (any pending exchanges are kept on the queue), but a consumer that is starting will poll it, + // so only break out once the queue is empty + if (shutdownPending && (isSuspending() || isSuspended() || queue.isEmpty())) { + LOG.trace("Consumer is suspended or starting and shutdown is pending, so this consumer thread is" + + " breaking out."); 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 index cfe6a63d3452..a37b6d781bb6 100644 --- 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 @@ -16,10 +16,12 @@ */ package org.apache.camel.component.seda; +import java.util.concurrent.BlockingQueue; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; import org.apache.camel.ServiceStatus; import org.apache.camel.builder.RouteBuilder; import org.apache.camel.component.mock.MockEndpoint; @@ -74,6 +76,9 @@ class SedaSuspendedRouteWithPendingStopTest extends ContextTestSupport { @Test void testStopContextWithSuspendedRoute() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:result"); + // keep a reference to the queue, as the endpoint releases it when it is shut down + BlockingQueue<Exchange> queue = ((SedaEndpoint) context.getRoute("foo").getEndpoint()).getQueue(); context.getRouteController().suspendRoute("foo"); template.sendBody("seda:start", "A"); @@ -82,6 +87,9 @@ class SedaSuspendedRouteWithPendingStopTest extends ContextTestSupport { context.getShutdownStrategy().setTimeout(10); context.stop(); assertFalse(context.getShutdownStrategy().isTimeoutOccurred(), "Graceful shutdown should not time out"); + // nothing was lost: a message is either still on the queue (purgeWhenStopping is false), or it was processed + // (a poll that was already in progress when the route was suspended may still take the first one) + assertEquals(2, queue.size() + mock.getReceivedCounter()); } @Override
