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 da2dee044359 CAMEL-26421: camel-core - a route called from a doTry
uses the error handler of its routing slip, multicast, recipient list and
dynamic router (#28779)
da2dee044359 is described below
commit da2dee044359116d7f33b69647b58771ed34e11e
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Nov 1 16:37:17 2026 +0530
CAMEL-26421: camel-core - a route called from a doTry uses the error
handler of its routing slip, multicast, recipient list and dynamic router
(#28779)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-Authored-By: GPT-6 <[email protected]>
---
.../ResilienceCalledRouteRoutingSlipTest.java | 63 +++++
.../camel/impl/engine/CamelInternalProcessor.java | 28 +-
.../DoTryCalledRouteErrorHandlerTest.java | 305 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 19 ++
4 files changed, 408 insertions(+), 7 deletions(-)
diff --git
a/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceCalledRouteRoutingSlipTest.java
b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceCalledRouteRoutingSlipTest.java
new file mode 100644
index 000000000000..297d77777b76
--- /dev/null
+++
b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceCalledRouteRoutingSlipTest.java
@@ -0,0 +1,63 @@
+/*
+ * 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.resilience4j;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+
+/**
+ * A route called from the steps of a circuit breaker uses the error handler
of its routing slip (CAMEL-26421).
+ */
+// a regression that hangs a send fails the test instead of blocking the build
+@Timeout(10)
+class ResilienceCalledRouteRoutingSlipTest extends CamelTestSupport {
+ private final AtomicInteger first = new AtomicInteger();
+ private final AtomicInteger failing = new AtomicInteger();
+
+ @Test
+ void calledRouteRedeliversTheFailingStepOfItsRoutingSlip() {
+ Exchange out = template.send("direct:start", e ->
e.getMessage().setBody("Hello"));
+ assertInstanceOf(IllegalStateException.class, out.getException());
+ assertEquals(1, first.get(), "the step before the failing one is
called once");
+ assertEquals(3, failing.get(), "the failing step is redelivered by the
error handler of the routing slip");
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:start").circuitBreaker().to("direct:called").end();
+
from("direct:called").errorHandler(defaultErrorHandler().maximumRedeliveries(2).redeliveryDelay(0))
+ .routingSlip(constant("direct:first,direct:failing"));
+ from("direct:first").errorHandler(noErrorHandler()).process(e
-> first.incrementAndGet());
+
from("direct:failing").errorHandler(noErrorHandler()).process(e -> {
+ failing.incrementAndGet();
+ throw new IllegalStateException("Failing step");
+ });
+ }
+ };
+ }
+}
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 eaded72ce636..a43b8ce41b79 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
@@ -489,7 +489,8 @@ public class CamelInternalProcessor extends
DelegateAsyncProcessor implements In
}
/**
- * Advice to invoke callbacks for before and after routing.
+ * Advice to invoke callbacks for before and after routing, and to keep
the doTry block of a calling route out of
+ * this route.
*/
public static class RouteLifecycleAdvice implements
CamelInternalProcessorAdvice<Object> {
@@ -505,20 +506,33 @@ public class CamelInternalProcessor extends
DelegateAsyncProcessor implements In
if (uow != null) {
uow.beforeRoute(exchange, route);
}
- return null;
+ // the try block marker of a doTry turns off the error handler per
output of the EIPs in the steps of
+ // that doTry (routing slip, dynamic router, multicast, recipient
list), but a route called from the doTry
+ // has its own error handler, so the marker of the caller is
removed here and set again when this route
+ // is done
+ return
exchange.removeProperty(ExchangePropertyKey.TRY_ROUTE_BLOCK);
}
@Override
- public void after(Exchange exchange, Object object) throws Exception {
- UnitOfWork uow = exchange.getUnitOfWork();
- if (uow != null) {
- uow.afterRoute(exchange, route);
+ public void after(Exchange exchange, Object tryRouteBlock) throws
Exception {
+ try {
+ UnitOfWork uow = exchange.getUnitOfWork();
+ if (uow != null) {
+ uow.afterRoute(exchange, route);
+ }
+ } finally {
+ // the caller is in its doTry again (also when this route
completed asynchronously)
+ if (tryRouteBlock != null) {
+ exchange.setProperty(ExchangePropertyKey.TRY_ROUTE_BLOCK,
tryRouteBlock);
+ } else {
+
exchange.removeProperty(ExchangePropertyKey.TRY_ROUTE_BLOCK);
+ }
}
}
@Override
public boolean hasState() {
- return false;
+ return true;
}
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/DoTryCalledRouteErrorHandlerTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/DoTryCalledRouteErrorHandlerTest.java
new file mode 100644
index 000000000000..6acd9bfad9cd
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/DoTryCalledRouteErrorHandlerTest.java
@@ -0,0 +1,305 @@
+/*
+ * 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.atomic.AtomicInteger;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+// a regression that hangs a send fails the test instead of blocking the build
+@Timeout(10)
+class DoTryCalledRouteErrorHandlerTest extends ContextTestSupport {
+
+ private final AtomicInteger failAttempts = new AtomicInteger();
+ private final AtomicInteger routerCalls = new AtomicInteger();
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ private void routes(boolean doTryCaller, String eip) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
onException(IllegalStateException.class).maximumRedeliveries(2).redeliveryDelay(0);
+ if (doTryCaller) {
+
from("direct:r0").doTry().to("direct:r1").doFinally().to("mock:fin").end();
+ } else {
+ from("direct:r0").to("direct:r1");
+ }
+ switch (eip) {
+ case "slip" ->
from("direct:r1").routingSlip(constant("mock:m1,direct:fail"));
+ case "asyncSlip" -> from("direct:r1").threads(1,
1).routingSlip(constant("mock:m1,direct:fail"));
+ case "multicast" ->
from("direct:r1").multicast().to("mock:m1", "direct:fail").end();
+ case "recipientList" ->
from("direct:r1").recipientList(constant("mock:m1,direct:fail"));
+ case "split" ->
from("direct:r1").split(constant("mock:m1,direct:fail")).toD("${body}").end();
+ case "router" ->
from("direct:r1").dynamicRouter(method(new Object() {
+ public String route() {
+ int n = routerCalls.getAndIncrement();
+ return n == 0 ? "mock:m1" : n == 1 ? "direct:fail"
: null;
+ }
+ }, "route"));
+ default ->
from("direct:r1").to("mock:m1").to("direct:fail");
+ }
+ failingRoute(this);
+ }
+ });
+ context.start();
+ }
+
+ private void check(boolean doTryCaller, String eip) throws Exception {
+ routes(doTryCaller, eip);
+ Exchange out = template.send("direct:r0", e ->
e.getMessage().setBody("b"));
+ assertInstanceOf(IllegalStateException.class, out.getException(), "the
failure must reach the caller");
+ assertEquals(1, getMockEndpoint("mock:m1").getReceivedCounter(),
"mock:m1 (no duplicates)");
+ assertEquals(3, failAttempts.get(), "attempts on the failing output (1
+ 2 redeliveries)");
+ }
+
+ @Test
+ void slipPlainCaller() throws Exception {
+ check(false, "slip");
+ }
+
+ @Test
+ void slipDoTryCaller() throws Exception {
+ check(true, "slip");
+ }
+
+ @Test
+ void multicastPlainCaller() throws Exception {
+ check(false, "multicast");
+ }
+
+ @Test
+ void multicastDoTryCaller() throws Exception {
+ check(true, "multicast");
+ }
+
+ @Test
+ void recipientListDoTryCaller() throws Exception {
+ check(true, "recipientList");
+ }
+
+ @Test
+ void splitPlainCaller() throws Exception {
+ check(false, "split");
+ }
+
+ // control: the steps of a split are channels with the error handler of
the route, so a split was not affected
+ @Test
+ void splitDoTryCaller() throws Exception {
+ check(true, "split");
+ }
+
+ @Test
+ void routerPlainCaller() throws Exception {
+ check(false, "router");
+ }
+
+ @Test
+ void routerDoTryCaller() throws Exception {
+ check(true, "router");
+ }
+
+ @Test
+ void plainToDoTryCaller() throws Exception {
+ check(true, "to");
+ }
+
+ @Test
+ void asynchronousCalledRouteUsesOwnErrorHandler() throws Exception {
+ check(true, "asyncSlip");
+ }
+
+ @Test
+ void returningFromCalledRouteRestoresOuterTry() throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
errorHandler(deadLetterChannel("mock:dead").maximumRedeliveries(2).redeliveryDelay(0));
+ from("direct:start").doTry()
+ .to("direct:sub")
+ .recipientList(constant("direct:fail")).end()
+ .doCatch(Exception.class).to("mock:caught").end();
+ from("direct:sub").threads(1, 1).to("mock:sub");
+ failingRoute(this);
+ }
+ });
+ context.start();
+ template.sendBody("direct:start", "body");
+ assertEquals(1, failAttempts.get());
+ assertEquals(1, getMockEndpoint("mock:caught").getReceivedCounter());
+ assertEquals(0, getMockEndpoint("mock:dead").getReceivedCounter());
+ }
+
+ @Test
+ void recursiveRouteInvocationHasItsOwnTryScope() throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
onException(IllegalStateException.class).maximumRedeliveries(2).redeliveryDelay(0);
+ from("direct:recursive").choice()
+ .when(header("nested").isNull())
+ .doTry().setHeader("nested",
constant(true)).to("direct:recursive")
+
.doCatch(Exception.class).to("mock:caught").end().endChoice()
+
.otherwise().routingSlip(constant("mock:m1,direct:fail")).end();
+ failingRoute(this);
+ }
+ });
+ context.start();
+ template.sendBody("direct:recursive", "body");
+ assertEquals(1, getMockEndpoint("mock:m1").getReceivedCounter());
+ assertEquals(3, failAttempts.get());
+ assertEquals(1, getMockEndpoint("mock:caught").getReceivedCounter());
+ }
+
+ @Test
+ void doCatchOfTheCallerCatchesAfterTheCalledRouteRedelivered() throws
Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
onException(IllegalStateException.class).maximumRedeliveries(2).redeliveryDelay(0);
+
from("direct:r0").doTry().to("direct:r1").doCatch(Exception.class).to("mock:caught").end();
+ from("direct:r1").routingSlip(constant("mock:m1,direct:fail"));
+ failingRoute(this);
+ }
+ });
+ context.start();
+ Exchange out = template.send("direct:r0", e ->
e.getMessage().setBody("b"));
+ assertNull(out.getException());
+ assertEquals(1, getMockEndpoint("mock:m1").getReceivedCounter());
+ assertEquals(3, failAttempts.get());
+ assertEquals(1, getMockEndpoint("mock:caught").getReceivedCounter());
+ }
+
+ @Test
+ void sedaCallFromDoTryUsesTheCalledRouteErrorHandler() throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
onException(IllegalStateException.class).maximumRedeliveries(2).redeliveryDelay(0);
+
from("direct:r0").doTry().to("seda:r1?waitForTaskToComplete=Always").doCatch(Exception.class)
+ .to("mock:caught").end();
+ from("seda:r1").routingSlip(constant("mock:m1,direct:fail"));
+ failingRoute(this);
+ }
+ });
+ context.start();
+ template.sendBody("direct:r0", "b");
+ assertEquals(1, getMockEndpoint("mock:m1").getReceivedCounter());
+ assertEquals(3, failAttempts.get());
+ assertEquals(1, getMockEndpoint("mock:caught").getReceivedCounter());
+ }
+
+ @Test
+ void enrichFromDoTryUsesTheCalledRouteErrorHandler() throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
onException(IllegalStateException.class).maximumRedeliveries(2).redeliveryDelay(0);
+
from("direct:r0").doTry().enrich("direct:r1").doCatch(Exception.class).to("mock:caught").end();
+ from("direct:r1").multicast().to("mock:m1",
"direct:fail").end();
+ failingRoute(this);
+ }
+ });
+ context.start();
+ template.sendBody("direct:r0", "b");
+ assertEquals(1, getMockEndpoint("mock:m1").getReceivedCounter());
+ assertEquals(3, failAttempts.get());
+ assertEquals(1, getMockEndpoint("mock:caught").getReceivedCounter());
+ }
+
+ @Test
+ void doTryInTheCalledRouteStillTurnsOffItsHandlerPerOutput() throws
Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
onException(IllegalStateException.class).maximumRedeliveries(2).redeliveryDelay(0);
+
from("direct:r0").doTry().to("direct:r1").doCatch(Exception.class).to("mock:outer").end();
+
from("direct:r1").doTry().recipientList(constant("direct:fail")).end()
+ .doCatch(Exception.class).to("mock:inner").end();
+ failingRoute(this);
+ }
+ });
+ context.start();
+ template.sendBody("direct:r0", "b");
+ assertEquals(1, failAttempts.get());
+ assertEquals(1, getMockEndpoint("mock:inner").getReceivedCounter());
+ assertEquals(0, getMockEndpoint("mock:outer").getReceivedCounter());
+ }
+
+ @Test
+ void returningFromACalledRouteThatContinuedAFailureRestoresOuterTry()
throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
errorHandler(deadLetterChannel("mock:dead").maximumRedeliveries(2).redeliveryDelay(0));
+ from("direct:r0").doTry()
+ .to("direct:r1")
+ .recipientList(constant("direct:fail")).end()
+ .doCatch(Exception.class).to("mock:caught").end();
+ from("direct:r1")
+
.onException(IllegalArgumentException.class).continued(true).end()
+ .throwException(new
IllegalArgumentException("continued"));
+ failingRoute(this);
+ }
+ });
+ context.start();
+ template.sendBody("direct:r0", "b");
+ // the recipient list after the call is in the try block of the caller
again (no error handler per output)
+ assertEquals(1, failAttempts.get());
+ assertEquals(1, getMockEndpoint("mock:caught").getReceivedCounter());
+ assertEquals(0, getMockEndpoint("mock:dead").getReceivedCounter());
+ }
+
+ @Test
+ void
multicastFailureHandledByTheCalledRouteNoLongerReachesTheCallersDoCatch()
throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:r0").doTry().to("direct:r1").doCatch(Exception.class).to("mock:caught").end();
+
from("direct:r1").errorHandler(deadLetterChannel("mock:dead").maximumRedeliveries(2).redeliveryDelay(0))
+ .multicast().to("mock:m1", "direct:fail").end();
+ failingRoute(this);
+ }
+ });
+ context.start();
+ Exchange out = template.send("direct:r0", e ->
e.getMessage().setBody("b"));
+ // the dead letter channel of the called route handles the failed
output after its redeliveries
+ assertNull(out.getException());
+ assertEquals(1, getMockEndpoint("mock:m1").getReceivedCounter());
+ assertEquals(3, failAttempts.get());
+ assertEquals(1, getMockEndpoint("mock:dead").getReceivedCounter());
+ assertEquals(0, getMockEndpoint("mock:caught").getReceivedCounter());
+ }
+
+ private void failingRoute(RouteBuilder builder) {
+
builder.from("direct:fail").errorHandler(builder.noErrorHandler()).process(e ->
{
+ failAttempts.incrementAndGet();
+ throw new IllegalStateException("boom");
+ });
+ }
+}
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 4b2f0584d565..20a743212e8f 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
@@ -300,6 +300,25 @@ space next to the comma, such as `+${replace(a,b )}+` for
the text `++b ++`, quo
Only the quotes around a whole argument are removed: a quote inside an
argument, as in `+a'b'c+`, is now kept (it was
removed).
+=== camel-core - a route called from a doTry uses the error handler of its EIPs
+
+A route that a `doTry` calls from its steps, or from its `doCatch` or
`doFinally` blocks (with `direct`, `seda`,
+`enrich` and so on), now uses the error handler per output of its routing
slip, dynamic router, multicast and recipient
+list EIPs, as it does when it is called from outside a `doTry`. Before, these
EIPs left out that error handler while the
+caller was in its `doTry`: a routing slip redelivered the whole slip (so an
endpoint that had already succeeded got the
+message again), a multicast or recipient list did not redeliver the failed
output, and a dynamic router that counts its
+calls could lose the failure. The same applies to a route called from the
steps of a circuit breaker. A redelivery now
+starts at the failed output, so such a route can make more delivery attempts
than before.
+
+When the called route handles failures (a dead letter channel, or an
`onException` with `handled(true)`), a failed
+output of its multicast is now handled there (after the redeliveries of that
error handler), and no longer reaches the
+`doCatch` of the caller.
+
+To turn off the error handler per output of such an EIP again (no redelivery
of its outputs, and the failure goes to the
+caller), put the EIP in a `doTry` with only a `doFinally` in the called route,
or turn off the error handler of the
+called route with `errorHandler(noErrorHandler())`, which turns it off for the
other steps of that route too. Neither
+option brings back the redelivery of a whole routing slip or dynamic router.
+
=== camel-util - FilterIterator.remove after hasNext throws
IllegalStateException
`FilterIterator.hasNext()` looks ahead in the iterator it wraps, so a
`remove()` after `next()` and then `hasNext()`
