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 ba044ff20a90 CAMEL-26333: camel-core - doFinally runs only its first
step after a called route's error handler handled a failure (#28756)
ba044ff20a90 is described below
commit ba044ff20a90ebd00b8166a87c0b6457aa63d201
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Nov 1 00:46:37 2026 +0530
CAMEL-26333: camel-core - doFinally runs only its first step after a called
route's error handler handled a failure (#28756)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/processor/FinallyProcessor.java | 41 ++++-
.../DoFinallyAfterHandledCalledRouteTest.java | 204 +++++++++++++++++++++
...utingSlipInDoFinallyAfterRecipientListTest.java | 127 +++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 9 +
4 files changed, 379 insertions(+), 2 deletions(-)
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/FinallyProcessor.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/FinallyProcessor.java
index 4cb237b9db6d..7b63ffc9fb62 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/FinallyProcessor.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/FinallyProcessor.java
@@ -18,6 +18,7 @@ package org.apache.camel.processor;
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
+import org.apache.camel.ExchangeExtension;
import org.apache.camel.ExchangePropertyKey;
import org.apache.camel.Processor;
import org.apache.camel.Traceable;
@@ -60,8 +61,26 @@ public class FinallyProcessor extends
BaseDelegateProcessorSupport
}
}
+ // a route called from the doTry (such as with direct) can have
handled a failure with its error handler (dead
+ // letter channel or a handled onException), which marks the exchange
as handled by the error handler and as
+ // failure handled. The route then does not continue after the doTry,
but the doFinally block must still run all
+ // its steps (a pipeline stops before its second step when the
exchange is handled by the error handler), and a
+ // failure in a route that the block calls must be handled by that
route's error handler as usual (which does not
+ // handle a new failure of an exchange that is failure handled). So
the marks are cleared while the block runs,
+ // and set again afterwards (as the onCompletion does)
+ final ExchangeExtension ext = exchange.getExchangeExtension();
+ // a primitive "set and true" snapshot, unlike the nullable Boolean of
OnCompletionProcessor: only a failure that
+ // an error handler handled needs clearing and restoring here
+ final boolean errorHandlerHandled = ext.isErrorHandlerHandledSet() &&
ext.isErrorHandlerHandled();
+ final boolean failureHandled = errorHandlerHandled &&
ext.isFailureHandled();
+ if (errorHandlerHandled) {
+ ext.setErrorHandlerHandled(null);
+ ext.setFailureHandled(false);
+ }
+
// continue processing
- return processor.process(exchange, new FinallyAsyncCallback(exchange,
callback, exception));
+ return processor.process(exchange,
+ new FinallyAsyncCallback(exchange, callback, exception,
errorHandlerHandled, failureHandled));
}
@Override
@@ -114,16 +133,34 @@ public class FinallyProcessor extends
BaseDelegateProcessorSupport
private final Exchange exchange;
private final AsyncCallback callback;
private final Exception exception;
+ private final boolean errorHandlerHandled;
+ private final boolean failureHandled;
- FinallyAsyncCallback(Exchange exchange, AsyncCallback callback,
Exception exception) {
+ FinallyAsyncCallback(Exchange exchange, AsyncCallback callback,
Exception exception,
+ boolean errorHandlerHandled, boolean
failureHandled) {
this.exchange = exchange;
this.callback = callback;
this.exception = exception;
+ this.errorHandlerHandled = errorHandlerHandled;
+ this.failureHandled = failureHandled;
}
@Override
public void done(boolean doneSync) {
try {
+ if (errorHandlerHandled) {
+ // the doFinally block has run, so the route does not
continue after the doTry, as before
+ // (unless the block marked it itself, such as a route it
called that failed)
+ final ExchangeExtension ext =
exchange.getExchangeExtension();
+ if (!ext.isErrorHandlerHandledSet()) {
+ ext.setErrorHandlerHandled(true);
+ }
+ // one-way restore, unlike OnCompletionProcessor: set
again but never cleared, so a failure that the
+ // block handled itself stays marked (errorHandlerHandled
without failureHandled does not occur)
+ if (failureHandled) {
+ ext.setFailureHandled(true);
+ }
+ }
if (exception == null) {
// only clear the failure details when the doFinally did
not fail
// (the doTry restores the failure details of an earlier
failure)
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/DoFinallyAfterHandledCalledRouteTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/DoFinallyAfterHandledCalledRouteTest.java
new file mode 100644
index 000000000000..5afe2e87bd71
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/DoFinallyAfterHandledCalledRouteTest.java
@@ -0,0 +1,204 @@
+/*
+ * 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.time.Duration;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
+
+/**
+ * The doTry calls a route whose error handler handles a failure (dead letter
channel or a handled onException). The
+ * doFinally block must run all its steps. The caller does not continue
routing after the doTry, as before
+ * (CAMEL-19441).
+ */
+class DoFinallyAfterHandledCalledRouteTest extends ContextTestSupport {
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ private Exchange send(String errorHandler, boolean calleeFails) throws
Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ switch (errorHandler) {
+ case "dlc" ->
errorHandler(deadLetterChannel("mock:dead").redeliveryDelay(0));
+ case "dlcAsync" ->
errorHandler(deadLetterChannel("mock:dead").maximumRedeliveries(1)
+ .redeliveryDelay(10).asyncDelayedRedelivery());
+ default ->
onException(IllegalArgumentException.class).handled(true).to("mock:dead");
+ }
+
+ from("direct:start")
+ .process(e -> e.setProperty("startThread",
Thread.currentThread().getName()))
+ .doTry()
+ .to("direct:callee")
+ .to("mock:afterCallee")
+ .doCatch(Exception.class)
+ .to("mock:catch")
+ .doFinally()
+ .to("mock:f1")
+ .to("mock:f2")
+ .setHeader("f3", constant("yes"))
+ .process(e -> e.setProperty("finallyThread",
Thread.currentThread().getName()))
+ .to("mock:f4")
+ .end()
+ .to("mock:end");
+
+ if (calleeFails) {
+ from("direct:callee").setBody(constant("in callee"))
+ .throwException(new
IllegalArgumentException("Forced"));
+ } else {
+ from("direct:callee").setBody(constant("in callee"));
+ }
+ }
+ });
+ context.start();
+
+ return assertTimeoutPreemptively(Duration.ofSeconds(10),
+ () -> template.send("direct:start", e ->
e.getMessage().setBody("Hello")), "the send did not complete");
+ }
+
+ private void assertAllFinallySteps(Exchange out) {
+ assertEquals(1, getMockEndpoint("mock:f1").getReceivedCounter(),
"mock:f1");
+ assertEquals(1, getMockEndpoint("mock:f2").getReceivedCounter(),
"mock:f2");
+ assertEquals(1, getMockEndpoint("mock:f4").getReceivedCounter(),
"mock:f4");
+ assertEquals("yes", out.getMessage().getHeader("f3"), "the header set
by the doFinally");
+ }
+
+ private void assertHandledByCalledRoute(Exchange out) {
+ assertNull(out.getException(), "the called route handled the failure");
+ assertEquals(1, getMockEndpoint("mock:dead").getReceivedCounter(),
"mock:dead");
+ assertEquals(0,
getMockEndpoint("mock:afterCallee").getReceivedCounter(), "mock:afterCallee");
+ assertEquals(0, getMockEndpoint("mock:catch").getReceivedCounter(),
"there is no exception to catch");
+ // the caller does not continue routing after the doTry (CAMEL-19441)
+ assertEquals(0, getMockEndpoint("mock:end").getReceivedCounter(),
"mock:end");
+ }
+
+ @Test
+ void deadLetterChannel() throws Exception {
+ Exchange out = send("dlc", true);
+ assertHandledByCalledRoute(out);
+ assertAllFinallySteps(out);
+ }
+
+ @Test
+ void deadLetterChannelAsyncDelayedRedelivery() throws Exception {
+ Exchange out = send("dlcAsync", true);
+ assertHandledByCalledRoute(out);
+ assertAllFinallySteps(out);
+ // the redelivery, and so the doFinally block, ran on another thread
than the one that started the route
+ assertNotEquals(out.getProperty("startThread"),
out.getProperty("finallyThread"));
+ }
+
+ @Test
+ void handledOnException() throws Exception {
+ Exchange out = send("handled", true);
+ assertHandledByCalledRoute(out);
+ assertAllFinallySteps(out);
+ }
+
+ @Test
+ void controlCalledRouteDoesNotFail() throws Exception {
+ Exchange out = send("dlc", false);
+ assertNull(out.getException());
+ assertEquals(0, getMockEndpoint("mock:dead").getReceivedCounter(),
"mock:dead");
+ assertEquals(1,
getMockEndpoint("mock:afterCallee").getReceivedCounter(), "mock:afterCallee");
+ assertAllFinallySteps(out);
+ assertEquals(1, getMockEndpoint("mock:end").getReceivedCounter(),
"mock:end");
+ }
+
+ @Test
+ void routeCalledFromTheDoFinallyHandlesItsOwnFailure() throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
errorHandler(deadLetterChannel("mock:dead").redeliveryDelay(0));
+
+ from("direct:start")
+ .doTry()
+ .to("direct:callee")
+ .doFinally()
+ .to("mock:f1")
+ .to("direct:callee2")
+ .to("mock:f2")
+ .end()
+ .to("mock:end");
+
+ from("direct:callee").throwException(new
IllegalArgumentException("Forced"));
+ from("direct:callee2").throwException(new
IllegalStateException("Forced in doFinally"));
+ }
+ });
+ context.start();
+
+ Exchange out = assertTimeoutPreemptively(Duration.ofSeconds(10),
+ () -> template.send("direct:start", e ->
e.getMessage().setBody("Hello")), "the send did not complete");
+ // the dead letter channel of the route called from the doFinally
handles that failure too, as when the doTry
+ // did not fail before
+ assertNull(out.getException(), "the called routes handled their
failures");
+ assertEquals(2, getMockEndpoint("mock:dead").getReceivedCounter(),
"mock:dead");
+ assertEquals(1, getMockEndpoint("mock:f1").getReceivedCounter(),
"mock:f1");
+ // a handled failure of a called route stops the caller, also in the
doFinally block
+ assertEquals(0, getMockEndpoint("mock:f2").getReceivedCounter(),
"mock:f2");
+ assertEquals(0, getMockEndpoint("mock:end").getReceivedCounter(),
"mock:end");
+ }
+
+ @Test
+ void routeCalledFromTheDoFinallyContinuesItsOwnFailure() throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
errorHandler(deadLetterChannel("mock:dead").redeliveryDelay(0));
+ onException(NumberFormatException.class).continued(true);
+
+ from("direct:start")
+ .doTry()
+ .to("direct:callee")
+ .doFinally()
+ .to("mock:f1")
+ .to("direct:callee2")
+ .to("mock:f2")
+ .end()
+ .to("mock:end");
+
+ from("direct:callee").throwException(new
IllegalArgumentException("Forced"));
+ from("direct:callee2").throwException(new
NumberFormatException("Forced in doFinally"))
+ .to("mock:callee2End");
+ }
+ });
+ context.start();
+
+ Exchange out = assertTimeoutPreemptively(Duration.ofSeconds(10),
+ () -> template.send("direct:start", e ->
e.getMessage().setBody("Hello")), "the send did not complete");
+ assertNull(out.getException(), "the called routes handled or continued
their failures");
+ assertEquals(1, getMockEndpoint("mock:dead").getReceivedCounter(),
"mock:dead");
+ // the onException of the route called from the doFinally continues
its failure
+ assertEquals(1,
getMockEndpoint("mock:callee2End").getReceivedCounter(), "mock:callee2End");
+ assertEquals(1, getMockEndpoint("mock:f1").getReceivedCounter(),
"mock:f1");
+ assertEquals(1, getMockEndpoint("mock:f2").getReceivedCounter(),
"mock:f2");
+ // the caller does not continue routing after the doTry (CAMEL-19441)
+ assertEquals(0, getMockEndpoint("mock:end").getReceivedCounter(),
"mock:end");
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/routingslip/RoutingSlipInDoFinallyAfterRecipientListTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/routingslip/RoutingSlipInDoFinallyAfterRecipientListTest.java
new file mode 100644
index 000000000000..56e43c3340a4
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/routingslip/RoutingSlipInDoFinallyAfterRecipientListTest.java
@@ -0,0 +1,127 @@
+/*
+ * 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.routingslip;
+
+import java.io.IOException;
+import java.time.Duration;
+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 static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
+
+/**
+ * A recipient list is the first step of a doFinally after a called route's
error handler handled a failure, and a
+ * routing slip (or dynamic router) runs later on the same exchange. The
recipient list must send to its recipients, and
+ * the routing slip must route to its own endpoints, not to the value the
recipient list evaluated.
+ */
+class RoutingSlipInDoFinallyAfterRecipientListTest extends ContextTestSupport {
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ private void routes(boolean deadLetterChannel, boolean dynamicRouter,
boolean calleeFails) throws Exception {
+ AtomicInteger routerCalls = new AtomicInteger();
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ if (deadLetterChannel) {
+
errorHandler(deadLetterChannel("mock:dead").redeliveryDelay(0));
+ } else {
+ onException(Exception.class).handled(true).to("mock:dead");
+ }
+
+ var outerFinally =
from("direct:start").doTry().to("direct:middle").doFinally();
+ if (dynamicRouter) {
+ outerFinally.dynamicRouter(method(new Object() {
+ public String route() {
+ return routerCalls.getAndIncrement() == 0 ?
"mock:slip" : null;
+ }
+ }, "route"));
+ } else {
+ outerFinally.routingSlip(constant("mock:slip"));
+ }
+ outerFinally.end();
+
+
from("direct:middle").doTry().to("direct:callee").doFinally().recipientList(constant("mock:a,mock:b"))
+ .end();
+
+ if (calleeFails) {
+ from("direct:callee").throwException(new
IOException("Forced"));
+ } else {
+ from("direct:callee").to("mock:callee");
+ }
+ }
+ });
+ context.start();
+ }
+
+ private Exchange send() {
+ return assertTimeoutPreemptively(Duration.ofSeconds(10),
+ () -> template.send("direct:start", e ->
e.getMessage().setBody("Hello")), "the send did not complete");
+ }
+
+ private void assertEachDestinationOnce(Exchange out) {
+ assertNull(out.getException());
+ assertEquals(1, getMockEndpoint("mock:a").getReceivedCounter(),
"mock:a (recipient list)");
+ assertEquals(1, getMockEndpoint("mock:b").getReceivedCounter(),
"mock:b (recipient list)");
+ assertEquals(1, getMockEndpoint("mock:slip").getReceivedCounter(),
"mock:slip (the routing slip's own endpoint)");
+
assertFalse(out.getProperties().containsKey(Exchange.EVALUATE_EXPRESSION_RESULT),
+ "the recipient list removed the value it evaluated");
+ }
+
+ @Test
+ void routingSlipWithDeadLetterChannel() throws Exception {
+ routes(true, false, true);
+ Exchange out = send();
+ assertEquals(1, getMockEndpoint("mock:dead").getReceivedCounter(),
"mock:dead");
+ assertEachDestinationOnce(out);
+ }
+
+ @Test
+ void routingSlipWithHandledOnException() throws Exception {
+ routes(false, false, true);
+ Exchange out = send();
+ assertEquals(1, getMockEndpoint("mock:dead").getReceivedCounter(),
"mock:dead");
+ assertEachDestinationOnce(out);
+ }
+
+ @Test
+ void dynamicRouterWithDeadLetterChannel() throws Exception {
+ routes(true, true, true);
+ Exchange out = send();
+ assertEquals(1, getMockEndpoint("mock:dead").getReceivedCounter(),
"mock:dead");
+ assertEachDestinationOnce(out);
+ }
+
+ @Test
+ void controlCalledRouteDoesNotFail() throws Exception {
+ routes(true, false, false);
+ Exchange out = send();
+ assertEquals(0, getMockEndpoint("mock:dead").getReceivedCounter(),
"mock:dead");
+ assertEquals(1, getMockEndpoint("mock:callee").getReceivedCounter(),
"mock:callee");
+ assertEachDestinationOnce(out);
+ }
+}
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 3be645a14d24..33e287c98cab 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
@@ -142,6 +142,15 @@ or payload from the body after a timed out `pollEnrich`
must keep it before the
with the same message id, headers and attachments, so after a timeout it is no
longer of the message type of the
consumer (such as `GenericFileMessage` or `JmsMessage`).
+=== camel-core - doFinally runs all its steps after a called route handled a
failure
+
+When a `doTry` calls a route (such as with `direct`) whose error handler
handles a failure (a dead letter channel or a
+handled `onException`), the `doFinally` block now runs all its steps. Before,
in all Camel 4.x releases, it ran only its
+first step and skipped the rest, so a second or later step of the block (such
as a `to`, a recipient list or a routing
+slip) did not send its message. A route called from such a `doFinally` that
fails is now handled (or continued) by its
+own error handler, as anywhere else; before, its error handler left the
failure alone and the exchange failed with that
+exception. As before, the route does not continue after the `doTry` once a
called route handled the failure.
+
=== camel-core - ThrottlingExceptionRoutePolicy counts the failures within the
failure window
The circuit of `ThrottlingExceptionRoutePolicy` now opens when
`failureThreshold` failures occurred within the last
