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 dc95b0ed26d94754afb45a958419b863dc6016d5 Author: smjain <[email protected]> AuthorDate: Wed Sep 23 18:10:49 2026 +0530 CAMEL-24949: camel-seda - do not purge the queue when suspending a route with purgeWhenStopping Cause: DefaultShutdownStrategy runs the same wait loop for suspend and shutdown, and asks every ShutdownAware for getPendingExchangesSize(). SedaConsumer purges its queue there whenever purgeWhenStopping=true, as it cannot know that the route is only being suspended. Effect: suspendRoute, CamelContext.suspend() or a JMX suspend of a SEDA route with purgeWhenStopping=true silently discards all queued messages, although the option is documented to purge when stopping the consumer/route. Fix: add a default method ShutdownAware.getPendingExchangesSize(boolean suspendOnly), delegating to getPendingExchangesSize(), and let the shutdown strategy call it with its suspendOnly flag. SedaConsumer only purges when not suspending; stopping still purges as before. Co-Authored-By: Claude Opus 5.5 <[email protected]> --- .../apache/camel/component/seda/SedaConsumer.java | 8 +- .../java/org/apache/camel/spi/ShutdownAware.java | 18 +++ .../camel/impl/engine/DefaultShutdownStrategy.java | 15 ++- .../seda/SedaPurgeWhenStoppingSuspendTest.java | 122 +++++++++++++++++++++ 4 files changed, 160 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 e4503f9f7ca8..6715de6ab0f7 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 @@ -86,9 +86,15 @@ public class SedaConsumer extends DefaultConsumer implements Runnable, ShutdownA @Override public int getPendingExchangesSize() { + return getPendingExchangesSize(false); + } + + @Override + public int getPendingExchangesSize(boolean suspendOnly) { // the route is shutting down, so either we should purge the queue, // or return how many exchanges are still on the queue - if (getEndpoint().isPurgeWhenStopping()) { + // (a suspended route must keep its pending exchanges, so only purge when stopping) + if (!suspendOnly && getEndpoint().isPurgeWhenStopping()) { getEndpoint().purgeQueue(); } return getEndpoint().getQueue().size(); diff --git a/core/camel-api/src/main/java/org/apache/camel/spi/ShutdownAware.java b/core/camel-api/src/main/java/org/apache/camel/spi/ShutdownAware.java index 1439d0c3e0c0..373f9fd3635f 100644 --- a/core/camel-api/src/main/java/org/apache/camel/spi/ShutdownAware.java +++ b/core/camel-api/src/main/java/org/apache/camel/spi/ShutdownAware.java @@ -51,4 +51,22 @@ public interface ShutdownAware extends ShutdownPrepared { */ int getPendingExchangesSize(); + /** + * Gets the number of pending exchanges, while the route is being suspended or shutdown. + * <p/> + * This is invoked by the {@link org.apache.camel.spi.ShutdownStrategy} while it waits for the pending exchanges to + * complete. Consumers which discard their pending exchanges on shutdown (for example the + * {@link org.apache.camel.component.seda.SedaConsumer} with the <tt>purgeWhenStopping</tt> option) must only do so + * when the route is being shutdown, and not when it is only being suspended. + * <p/> + * By default, this delegates to {@link #getPendingExchangesSize()}. + * + * @param suspendOnly <tt>true</tt> if the route is only being suspended, <tt>false</tt> if it is being shutdown + * @return number of pending exchanges + * @since 4.23 + */ + default int getPendingExchangesSize(boolean suspendOnly) { + return getPendingExchangesSize(); + } + } diff --git a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultShutdownStrategy.java b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultShutdownStrategy.java index 40640bee5fa5..d69112208646 100644 --- a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultShutdownStrategy.java +++ b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultShutdownStrategy.java @@ -671,7 +671,7 @@ public class DefaultShutdownStrategy extends ServiceSupport implements ShutdownS for (RouteStartupOrder order : routes) { int inflight = context.getInflightRepository().size(order.getRoute().getId()); - inflight += getPendingInflightExchanges(order); + inflight += getPendingInflightExchanges(order, suspendOnly); if (inflight > 0) { String routeId = order.getRoute().getId(); routeInflight.put(routeId, inflight); @@ -770,6 +770,17 @@ public class DefaultShutdownStrategy extends ServiceSupport implements ShutdownS * @return number of inflight exchanges */ protected static int getPendingInflightExchanges(RouteStartupOrder order) { + return getPendingInflightExchanges(order, false); + } + + /** + * Calculates the total number of inflight exchanges for the given route + * + * @param order the route + * @param suspendOnly whether the route is only being suspended (and not shutdown) + * @return number of inflight exchanges + */ + protected static int getPendingInflightExchanges(RouteStartupOrder order, boolean suspendOnly) { int inflight = 0; // the consumer is the 1st service so we always get the consumer @@ -779,7 +790,7 @@ public class DefaultShutdownStrategy extends ServiceSupport implements ShutdownS Set<Service> children = ServiceHelper.getChildServices(service); for (Service child : children) { if (child instanceof ShutdownAware shutdownAware) { - inflight += shutdownAware.getPendingExchangesSize(); + inflight += shutdownAware.getPendingExchangesSize(suspendOnly); } } } diff --git a/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaPurgeWhenStoppingSuspendTest.java b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaPurgeWhenStoppingSuspendTest.java new file mode 100644 index 000000000000..6ee915c278b8 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaPurgeWhenStoppingSuspendTest.java @@ -0,0 +1,122 @@ +/* + * 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.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +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; +import org.apache.camel.spi.Registry; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Suspending a route must not discard the pending messages of a seda consumer with purgeWhenStopping=true, the option + * only applies when stopping. + */ +public class SedaPurgeWhenStoppingSuspendTest extends ContextTestSupport { + + private final CountDownLatch firstStarted = new CountDownLatch(1); + private final CountDownLatch releaseFirst = new CountDownLatch(1); + private final CountDownLatch pendingChecked = new CountDownLatch(1); + private final AtomicBoolean suspending = new AtomicBoolean(); + + private final LinkedBlockingQueue<Exchange> queue = new LinkedBlockingQueue<>() { + @Override + public int size() { + // the shutdown strategy asks the seda consumer for its pending exchanges (the queue size) + if (suspending.get()) { + pendingChecked.countDown(); + } + return super.size(); + } + }; + + @Override + protected Registry createCamelRegistry() throws Exception { + Registry registry = super.createCamelRegistry(); + registry.bind("myQueue", queue); + return registry; + } + + @Test + public void testSuspendDoesNotPurge() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:result"); + mock.expectedBodiesReceived("A", "B", "C", "D", "E"); + + for (String body : new String[] { "A", "B", "C", "D", "E" }) { + template.sendBody("seda:foo", body); + } + // A is being processed and B..E are pending on the queue + assertTrue(firstStarted.await(10, TimeUnit.SECONDS)); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + suspending.set(true); + Future<?> suspend = executor.submit(() -> { + context.getRouteController().suspendRoute("myRoute"); + return null; + }); + // let A complete when the shutdown strategy is waiting for the pending exchanges + assertTrue(pendingChecked.await(10, TimeUnit.SECONDS)); + releaseFirst.countDown(); + + suspend.get(20, TimeUnit.SECONDS); + } finally { + executor.shutdownNow(); + } + + // the pending messages were not discarded + mock.assertIsSatisfied(); + assertEquals(ServiceStatus.Suspended, context.getRouteController().getRouteStatus("myRoute")); + + // and the route works after resume + mock.reset(); + mock.expectedBodiesReceived("F"); + context.getRouteController().resumeRoute("myRoute"); + template.sendBody("seda:foo", "F"); + mock.assertIsSatisfied(); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("seda:foo?queue=#myQueue&purgeWhenStopping=true").routeId("myRoute") + .process(exchange -> { + if ("A".equals(exchange.getMessage().getBody(String.class))) { + firstStarted.countDown(); + releaseFirst.await(10, TimeUnit.SECONDS); + } + }) + .to("mock:result"); + } + }; + } +}
