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
The following commit(s) were added to refs/heads/main by this push:
new 7705d668cba0 CAMEL-24950: camel-seda - complete the exchanges
discarded by purging the queue
7705d668cba0 is described below
commit 7705d668cba0cb63699c911a4dbe6782c30611b1
Author: smjain <[email protected]>
AuthorDate: Wed Sep 23 18:17:43 2026 +0530
CAMEL-24950: camel-seda - complete the exchanges discarded by purging the
queue
Cause: SedaEndpoint.purgeQueue() (used by purgeWhenStopping and the JMX
operation) just cleared the queue. A request/reply exchange sent with
waitForTaskToComplete is a copy whose onDone synchronization releases the
waiting SedaProducer, and that synchronization never ran for a discarded
copy.
Effect: a producer waiting for a purged exchange waits its full timeout
and then fails with ExchangeTimedOutException, or waits forever with
timeout=0 ("disable timeout"). The producer's own route then cannot shut
down gracefully either, as its exchange stays inflight. On completions
handed over to discarded InOnly exchanges never ran.
Fix: drain the queue instead of clearing it, fail each discarded exchange
with a RejectedExecutionException and run its on completions, as the
consumer does after routing a multicast exchange.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/component/seda/SedaEndpoint.java | 14 ++-
.../SedaPurgeWhenStoppingWaitingProducerTest.java | 128 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 8 ++
3 files changed, 148 insertions(+), 2 deletions(-)
diff --git
a/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java
b/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java
index 4ac74dcd5a43..57e0eafbd376 100644
---
a/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java
+++
b/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java
@@ -23,6 +23,7 @@ import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
import org.apache.camel.AsyncEndpoint;
import org.apache.camel.AsyncProcessor;
@@ -45,6 +46,7 @@ import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriPath;
import org.apache.camel.support.DefaultEndpoint;
import org.apache.camel.support.PluginHelper;
+import org.apache.camel.support.UnitOfWorkHelper;
import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.util.URISupport;
import org.slf4j.Logger;
@@ -582,12 +584,20 @@ public class SedaEndpoint extends DefaultEndpoint
implements AsyncEndpoint, Brow
}
/**
- * Purges the queue
+ * Purges the queue.
+ * <p/>
+ * The discarded exchanges are failed with a {@link
RejectedExecutionException} and their on completions are
+ * executed, so a producer waiting for the reply of a discarded exchange
is released.
*/
@ManagedOperation(description = "Purges the seda queue")
public void purgeQueue() {
LOG.debug("Purging queue with {} exchanges", queue.size());
- queue.clear();
+ List<Exchange> discarded = new ArrayList<>();
+ queue.drainTo(discarded);
+ for (Exchange exchange : discarded) {
+ exchange.setException(new RejectedExecutionException("Exchange
discarded as the SEDA queue was purged"));
+ UnitOfWorkHelper.doneSynchronizations(exchange,
exchange.getExchangeExtension().handoverCompletions());
+ }
}
/**
diff --git
a/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaPurgeWhenStoppingWaitingProducerTest.java
b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaPurgeWhenStoppingWaitingProducerTest.java
new file mode 100644
index 000000000000..2e2cf1eecb0b
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaPurgeWhenStoppingWaitingProducerTest.java
@@ -0,0 +1,128 @@
+/*
+ * 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.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.support.SynchronizationAdapter;
+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.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Exchanges discarded by purging the seda queue must be completed, so a
producer waiting for their reply is released.
+ */
+public class SedaPurgeWhenStoppingWaitingProducerTest extends
ContextTestSupport {
+
+ private final CountDownLatch busyStarted = new CountDownLatch(1);
+ private final CountDownLatch releaseBusy = new CountDownLatch(1);
+
+ @Test
+ public void testWaitingProducerReleasedWhenStopping() throws Exception {
+ template.sendBody("seda:svc", "busy");
+ assertTrue(busyStarted.await(10, TimeUnit.SECONDS));
+
+ // request/reply without timeout, queued behind the busy message
+ Exchange request =
context.getEndpoint("seda:svc").createExchange(ExchangePattern.InOut);
+ request.getMessage().setBody("request");
+ Future<Exchange> reply = template.asyncSend("seda:svc?timeout=0",
request);
+ SedaEndpoint endpoint = context.getEndpoint("seda:svc",
SedaEndpoint.class);
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
endpoint.getQueue().size() == 1);
+
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future<?> stop = executor.submit(() -> {
+ context.getRouteController().stopRoute("svc");
+ return null;
+ });
+
+ // the request is discarded by the purge, and the waiting producer
is released
+ Exchange out = reply.get(10, TimeUnit.SECONDS);
+ assertInstanceOf(RejectedExecutionException.class,
out.getException());
+
+ releaseBusy.countDown();
+ stop.get(20, TimeUnit.SECONDS);
+ } finally {
+ releaseBusy.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testPurgeQueueCompletesDiscardedExchanges() throws Exception {
+ template.sendBody("seda:svc", "busy");
+ assertTrue(busyStarted.await(10, TimeUnit.SECONDS));
+
+ CountDownLatch failed = new CountDownLatch(1);
+ Exchange inOnly =
context.getEndpoint("seda:svc").createExchange(ExchangePattern.InOnly);
+ inOnly.getMessage().setBody("inOnly");
+ inOnly.getExchangeExtension().addOnCompletion(new
SynchronizationAdapter() {
+ @Override
+ public void onFailure(Exchange exchange) {
+ failed.countDown();
+ }
+ });
+ template.send("seda:svc", inOnly);
+
+ Exchange request =
context.getEndpoint("seda:svc").createExchange(ExchangePattern.InOut);
+ request.getMessage().setBody("request");
+ Future<Exchange> reply = template.asyncSend("seda:svc?timeout=0",
request);
+
+ SedaEndpoint endpoint = context.getEndpoint("seda:svc",
SedaEndpoint.class);
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
endpoint.getQueue().size() == 2);
+
+ try {
+ // as done from JMX
+ endpoint.purgeQueue();
+
+ assertEquals(0, endpoint.getQueue().size());
+ assertTrue(failed.await(10, TimeUnit.SECONDS), "on completion of
the discarded InOnly exchange should be done");
+ Exchange out = reply.get(10, TimeUnit.SECONDS);
+ assertInstanceOf(RejectedExecutionException.class,
out.getException());
+ } finally {
+ releaseBusy.countDown();
+ }
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("seda:svc?purgeWhenStopping=true").routeId("svc")
+ .process(exchange -> {
+ if
("busy".equals(exchange.getMessage().getBody(String.class))) {
+ busyStarted.countDown();
+ releaseBusy.await(20, TimeUnit.SECONDS);
+ }
+ });
+ }
+ };
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
index eace13660003..1e20dc11dd8e 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
@@ -2622,3 +2622,11 @@ Routes that reference the constants (for example
`setHeader(MustacheConstants.MU
are unaffected. Routes that set the header by its literal string name, or that
use
`allowTemplateFromHeader=true` with the old header names, must switch to the
new `Camel`-prefixed
names.
+
+=== camel-seda - purging the queue completes the discarded exchanges
+
+When a SEDA queue is purged (with `purgeWhenStopping=true` or the `purgeQueue`
JMX operation), the discarded exchanges
+are now failed with a `RejectedExecutionException` and their on completions
are executed. A producer waiting for the
+reply of a discarded exchange (`waitForTaskToComplete`) is released with that
exception, instead of waiting until its
+`timeout`, or forever when the timeout is disabled. On completions handed over
to a discarded InOnly exchange, such as
+the commit or rollback of the consumer that received the message, now run as a
failure, where previously they never ran.
