This is an automated email from the ASF dual-hosted git repository.
davsclaus 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 e8a5c53c89d3 CAMEL-26601: camel-core - Recipient list in a doTry: a
producer that throws fails the exchange instead of hanging (#29295)
e8a5c53c89d3 is described below
commit e8a5c53c89d327833617f116fed2c577c9e36f60
Author: allthingssecurity <[email protected]>
AuthorDate: Fri Nov 6 13:06:27 2026 +0530
CAMEL-26601: camel-core - Recipient list in a doTry: a producer that throws
fails the exchange instead of hanging (#29295)
Inside a doTry, the recipient list now guards the unit-of-work processor of
each recipient, so a producer that throws before calling the callback fails the
exchange instead of leaving it hanging.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/impl/engine/CamelInternalProcessor.java | 13 +-
.../apache/camel/processor/MulticastProcessor.java | 6 +
.../RecipientListProducerThrowsInDoTryTest.java | 287 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 4 +-
4 files changed, 306 insertions(+), 4 deletions(-)
diff --git
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/CamelInternalProcessor.java
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/CamelInternalProcessor.java
index 0fa6cd81b25a..3718aee4f3c3 100644
---
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/CamelInternalProcessor.java
+++
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/CamelInternalProcessor.java
@@ -131,8 +131,8 @@ public class CamelInternalProcessor extends
DelegateAsyncProcessor implements In
// the id of the step, when this is the channel of a step that asks the
step hook itself, as it has no error handler
// that does (set by DefaultChannel)
String stepHookNodeId;
- // whether the processor is the step itself, with no error handler around
it (set by DefaultChannel, such as for the
- // steps of a doTry or an onException): this then catches an exception
that the processor throws instead of
+ // whether the processor has no error handler around it (set by
DefaultChannel for a step without an error handler,
+ // such as the steps of a doTry or an onException, or through
setCatchProcessorThrows): this then catches an exception that the processor
throws instead of
// calling the callback, which would otherwise be dropped by the reactive
executor and leave the exchange hanging
boolean catchProcessorThrows;
@@ -225,6 +225,15 @@ public class CamelInternalProcessor extends
DelegateAsyncProcessor implements In
addAdvice(CamelInternalProcessor.wrap(processor));
}
+ /**
+ * Whether to catch an exception that the wrapped processor throws from
{@code process(exchange, callback)}, instead
+ * of setting it on the exchange, and fail the exchange with it. For a
processor with no error handler around it, as
+ * {@link DefaultChannel} does for a step without an error handler. Must
be set before this processor is used.
+ */
+ public void setCatchProcessorThrows(boolean catchProcessorThrows) {
+ this.catchProcessorThrows = catchProcessorThrows;
+ }
+
@Override
public void setRouteOnAdvices(Route route) {
RoutePolicyAdvice task = getAdvice(RoutePolicyAdvice.class);
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java
index 72a7b82882e1..78eb0db6132e 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java
@@ -52,6 +52,7 @@ import org.apache.camel.Route;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.StreamCache;
import org.apache.camel.Traceable;
+import org.apache.camel.impl.engine.CamelInternalProcessor;
import org.apache.camel.processor.aggregate.ShareUnitOfWorkAggregationStrategy;
import org.apache.camel.processor.aggregate.UseOriginalAggregationStrategy;
import org.apache.camel.processor.errorhandler.ErrorHandlerSupport;
@@ -1311,6 +1312,11 @@ public class MulticastProcessor extends
BaseProcessorSupport
} else {
// and wrap in unit of work processor so the copy exchange also
can run under UoW
answer = createUnitOfWorkProcessor(route, withSendHook(processor),
exchange);
+ if (answer instanceof CamelInternalProcessor internal) {
+ // no error handler around the output (such as in a doTry): a
recipient producer that throws from
+ // process(exchange, callback) fails the copy instead of
hanging the recipient list, as in a channel
+ internal.setCatchProcessorThrows(true);
+ }
}
return answer;
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/RecipientListProducerThrowsInDoTryTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/RecipientListProducerThrowsInDoTryTest.java
new file mode 100644
index 000000000000..ff1231bb20d7
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/RecipientListProducerThrowsInDoTryTest.java
@@ -0,0 +1,287 @@
+/*
+ * 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.processor;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Consumer;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.model.RecipientListDefinition;
+import org.apache.camel.spi.CamelEvent;
+import org.apache.camel.spi.CamelEvent.ExchangeCompletedEvent;
+import org.apache.camel.spi.CamelEvent.ExchangeFailedEvent;
+import org.apache.camel.support.DefaultAsyncProducer;
+import org.apache.camel.support.DefaultEndpoint;
+import org.apache.camel.support.EventNotifierSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * A recipient of a recipient list in a doTry has no error handler around its
producer. When the producer throws from
+ * {@code process(exchange, callback)} instead of setting the exception on the
exchange, the exception reached the task
+ * of the recipient, which the reactive executor runs and which dropped it:
the recipient list waited for good. The
+ * exchange must go on as when the producer sets the exception, and be
continued exactly once. Each send is bounded, so
+ * a hang fails the test.
+ */
+class RecipientListProducerThrowsInDoTryTest extends ContextTestSupport {
+
+ private final AtomicInteger calls = new AtomicInteger();
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ /**
+ * What the producer does with the exchange and its callback.
+ */
+ private interface Behaviour {
+ boolean process(Exchange exchange, AsyncCallback callback);
+ }
+
+ private final class TestEndpoint extends DefaultEndpoint {
+ private final Behaviour behaviour;
+
+ private TestEndpoint(String uri, Behaviour behaviour) {
+ this.behaviour = behaviour;
+ setEndpointUri(uri);
+ }
+
+ @Override
+ public Producer createProducer() {
+ return new DefaultAsyncProducer(this) {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback
callback) {
+ calls.incrementAndGet();
+ return behaviour.process(exchange, callback);
+ }
+ };
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ protected String createEndpointUri() {
+ return getEndpointUri();
+ }
+ }
+
+ private static boolean sets(Exchange exchange, AsyncCallback callback) {
+ exchange.setException(new IllegalStateException("failed"));
+ callback.done(true);
+ return true;
+ }
+
+ private static boolean throwsFailure(Exchange exchange, AsyncCallback
callback) {
+ throw new IllegalStateException("failed");
+ }
+
+ private interface Options {
+ void apply(RecipientListDefinition<?> recipientList);
+ }
+
+ /**
+ * The routes direct:{id} with a recipient list of test:{id} in a doTry,
to mock:{id}-result, and a doCatch to
+ * mock:{id}-caught.
+ */
+ private void addRoute(RouteBuilder b, String id, Options options) {
+ RecipientListDefinition<?> recipientList = b.from("direct:" + id)
+ .doTry()
+ .recipientList(b.constant("test:" + id));
+ options.apply(recipientList);
+ recipientList.end()
+ .to("mock:" + id + "-result")
+ .endDoTry()
+ .doCatch(IllegalStateException.class)
+ .to("mock:" + id + "-caught")
+ .end();
+ }
+
+ private void start(Options options, Behaviour behaviour) throws Exception {
+ context.addEndpoint("test:sets", new TestEndpoint("test:sets",
RecipientListProducerThrowsInDoTryTest::sets));
+ context.addEndpoint("test:start", new TestEndpoint("test:start",
behaviour));
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ addRoute(this, "sets", options);
+ addRoute(this, "start", options);
+ }
+ });
+ context.start();
+ }
+
+ private AtomicInteger countDoneExchanges() {
+ AtomicInteger done = new AtomicInteger();
+ context.getManagementStrategy().addEventNotifier(new
EventNotifierSupport() {
+ @Override
+ public void notify(CamelEvent event) {
+ done.incrementAndGet();
+ }
+
+ @Override
+ public boolean isEnabled(CamelEvent event) {
+ return event instanceof ExchangeCompletedEvent || event
instanceof ExchangeFailedEvent;
+ }
+ });
+ return done;
+ }
+
+ private String outcome(Exchange out, String id) {
+ Exception caught = out.getProperty(Exchange.EXCEPTION_CAUGHT,
Exception.class);
+ return "exception=" + describe(out.getException()) + " caught=" +
describe(caught)
+ + " result=" + getMockEndpoint("mock:" + id +
"-result").getReceivedCounter()
+ + " caughtMock=" + getMockEndpoint("mock:" + id +
"-caught").getReceivedCounter();
+ }
+
+ private static String describe(Throwable t) {
+ // the message of a CamelExchangeException ends with the exchange,
which differs between the routes
+ return t == null ? "none" : t.getClass().getSimpleName() + ":" +
t.getMessage().replaceAll("\\. Exchange\\[.*", "");
+ }
+
+ /**
+ * The route whose producer sets the exception (direct:sets) and the one
whose producer behaves as given
+ * (direct:start) must do the same, and be done as many times.
+ */
+ private void assertSameAsSettingTheException(Options options, Behaviour
behaviour, String expected)
+ throws Exception {
+ start(options, behaviour);
+ AtomicInteger done = countDoneExchanges();
+ Exchange reference = template.asyncSend("direct:sets", e ->
e.getMessage().setBody("Hello")).get(10,
+ TimeUnit.SECONDS);
+ assertEquals(expected, outcome(reference, "sets"), "the producer that
sets the exception");
+ await().atMost(10, TimeUnit.SECONDS).until(() -> done.get() >= 1);
+ int referenceDone = done.getAndSet(0);
+ calls.set(0);
+
+ Exchange out = template.asyncSend("direct:start", e ->
e.getMessage().setBody("Hello")).get(10, TimeUnit.SECONDS);
+ assertEquals(expected, outcome(out, "start"), "the route whose
producer does not set the exception");
+ await().atMost(10, TimeUnit.SECONDS).until(() -> done.get() >=
referenceDone);
+ // wait until nothing is in flight, so a second continuation that
comes later is counted as well
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
context.getInflightRepository().size() == 0);
+ assertEquals(1, calls.get(), "the producer is called once");
+ // and send one more exchange, so a second continuation queued before
it is counted as well
+ template.asyncSend("direct:sets", e ->
e.getMessage().setBody("Hello")).get(10, TimeUnit.SECONDS);
+ await().atMost(10, TimeUnit.SECONDS).until(() -> done.get() >= 2 *
referenceDone);
+ assertEquals(2 * referenceDone, done.get(),
+ "the exchanges done, as with the producer that sets the
exception (and once more for the later exchange)");
+ }
+
+ private static final String CAUGHT
+ = "exception=none caught=IllegalStateException:failed result=0
caughtMock=1";
+
+ private static final String CAUGHT_STOP_ON_EXCEPTION
+ = "exception=none caught=CamelExchangeException:Multicast
processing failed for number 0 result=0 caughtMock=1";
+
+ @Test
+ void producerThrows() throws Exception {
+ assertSameAsSettingTheException(r -> {
+ }, RecipientListProducerThrowsInDoTryTest::throwsFailure, CAUGHT);
+ }
+
+ @Test
+ void producerThrowsSynchronous() throws Exception {
+ assertSameAsSettingTheException(r -> r.synchronous(),
RecipientListProducerThrowsInDoTryTest::throwsFailure,
+ CAUGHT);
+ }
+
+ @Test
+ void producerThrowsParallel() throws Exception {
+ assertSameAsSettingTheException(r -> r.parallelProcessing(),
RecipientListProducerThrowsInDoTryTest::throwsFailure,
+ CAUGHT);
+ }
+
+ @Test
+ void producerThrowsStopOnException() throws Exception {
+ assertSameAsSettingTheException(r -> r.stopOnException(),
RecipientListProducerThrowsInDoTryTest::throwsFailure,
+ CAUGHT_STOP_ON_EXCEPTION);
+ }
+
+ @Test
+ void producerThrowsParallelTimeout() throws Exception {
+ // with a timeout the recipient list did not hang, but returned after
the timeout without the failure
+ assertSameAsSettingTheException(r ->
r.parallelProcessing().timeout(5000),
+ RecipientListProducerThrowsInDoTryTest::throwsFailure, CAUGHT);
+ }
+
+ @Test
+ void producerThrowsShareUnitOfWork() throws Exception {
+ assertSameAsSettingTheException(r -> r.shareUnitOfWork(),
RecipientListProducerThrowsInDoTryTest::throwsFailure,
+ CAUGHT);
+ }
+
+ @Test
+ void producerCallsBackThenThrows() throws Exception {
+ // the callback came first and continued the exchange: the exception
is only logged, as before (control: same
+ // on main)
+ assertSameAsSettingTheException(r -> {
+ }, (exchange, callback) -> {
+ exchange.setException(new IllegalStateException("failed"));
+ callback.done(true);
+ throw new IllegalStateException("thrown after the callback");
+ }, CAUGHT);
+ }
+
+ @Test
+ void producerHandsOnCallbackThenThrows() throws Exception {
+ // the callback was queued on the reactive executor before the
producer threw: it comes first and continues the
+ // exchange once, the takeover is only logged (control: same on main;
a takeover without the guard would
+ // continue the exchange twice)
+ assertSameAsSettingTheException(r -> {
+ }, (exchange, callback) -> {
+ exchange.setException(new IllegalStateException("failed"));
+
exchange.getContext().getCamelContextExtension().getReactiveExecutor().schedule(()
-> callback.done(false));
+ throw new IllegalStateException("thrown after handing on the
callback");
+ }, CAUGHT);
+ }
+
+ @Test
+ void sedaQueueFull() throws Exception {
+ // a real producer that throws from process(exchange, callback): the
seda producer throws IllegalStateException
+ // "Queue full" when its bounded queue is full (the default, neither
blockWhenFull nor discardWhenFull)
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:seda")
+ .doTry()
+ .recipientList(constant("seda:full?size=1"))
+ .end()
+ .to("mock:seda-result")
+ .endDoTry()
+ .doCatch(IllegalStateException.class)
+ .to("mock:seda-caught")
+ .end();
+ }
+ });
+ context.start();
+ // fills the queue (no consumer)
+ template.sendBody("seda:full?size=1", "first");
+
+ Exchange out = template.asyncSend("direct:seda", e ->
e.getMessage().setBody("Hello")).get(10, TimeUnit.SECONDS);
+ assertEquals("exception=none caught=IllegalStateException:Queue full
result=0 caughtMock=1", outcome(out, "seda"));
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 3e0e95612b3f..41a3b204fe36 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -479,8 +479,8 @@ option brings back the redelivery of a whole routing slip
or dynamic router.
=== camel-core - an AsyncProcessor that throws in a step without an error
handler fails the exchange
A step that has no error handler of its own (the steps of `doTry`, `doCatch`,
`doFinally` and `onException`, also when
-nested in another EIP inside them, the outputs of a multicast, the steps of a
circuit breaker, and a step with
-`inheritErrorHandler(false)`) now handles a processor that throws from
`process(exchange, callback)` instead of
+nested in another EIP inside them, the outputs of a multicast, the recipients
of a recipient list inside a `doTry`,
+the steps of a circuit breaker, and a step with `inheritErrorHandler(false)`)
now handles a processor that throws from `process(exchange, callback)` instead
of
setting the exception on the exchange as if it had set it: the exchange fails
with that exception, so the `doCatch`
catches it, and the output of a multicast is redelivered by the error handler
of the route as for any other failure.
Before, the exchange hung.
