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 930d4d31dbcb CAMEL-26422: camel-resilience4j,
camel-microprofile-fault-tolerance - a circuit breaker no longer ends the
enclosing doTry (#28780)
930d4d31dbcb is described below
commit 930d4d31dbcbc608c4d66d58780b9ee8eeb964d1
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Nov 1 15:18:22 2026 +0530
CAMEL-26422: camel-resilience4j, camel-microprofile-fault-tolerance - a
circuit breaker no longer ends the enclosing doTry (#28780)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-Authored-By: GPT-6 <[email protected]>
---
.../faulttolerance/FaultToleranceProcessor.java | 17 ++-
.../FaultToleranceOuterTryScopeTest.java | 128 +++++++++++++++++
.../resilience4j/ResilienceProcessor.java | 26 +++-
.../resilience4j/ResilienceOuterTryScopeTest.java | 152 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 11 ++
5 files changed, 328 insertions(+), 6 deletions(-)
diff --git
a/components/camel-microprofile/camel-microprofile-fault-tolerance/src/main/java/org/apache/camel/component/microprofile/faulttolerance/FaultToleranceProcessor.java
b/components/camel-microprofile/camel-microprofile-fault-tolerance/src/main/java/org/apache/camel/component/microprofile/faulttolerance/FaultToleranceProcessor.java
index 31fb84bc7213..6044525d355e 100644
---
a/components/camel-microprofile/camel-microprofile-fault-tolerance/src/main/java/org/apache/camel/component/microprofile/faulttolerance/FaultToleranceProcessor.java
+++
b/components/camel-microprofile/camel-microprofile-fault-tolerance/src/main/java/org/apache/camel/component/microprofile/faulttolerance/FaultToleranceProcessor.java
@@ -281,6 +281,8 @@ public class FaultToleranceProcessor extends
BaseProcessorSupport
@SuppressWarnings("unchecked")
public boolean process(Exchange exchange, AsyncCallback callback) {
// run this as if we run inside try / catch so there is no regular
Camel error handler
+ Object previousTryRouteBlock =
exchange.getProperty(ExchangePropertyKey.TRY_ROUTE_BLOCK);
+ Object previousHandled =
exchange.removeProperty(ExchangePropertyKey.EXCEPTION_HANDLED);
exchange.setProperty(ExchangePropertyKey.TRY_ROUTE_BLOCK, true);
CircuitBreakerTask task = (CircuitBreakerTask)
taskFactory.acquire(exchange, callback);
// guard to prevent the worker thread from writing results back to the
original exchange
@@ -342,11 +344,24 @@ public class FaultToleranceProcessor extends
BaseProcessorSupport
}
}
- exchange.removeProperty(ExchangePropertyKey.TRY_ROUTE_BLOCK);
+ restoreScopeProperty(exchange, ExchangePropertyKey.TRY_ROUTE_BLOCK,
previousTryRouteBlock);
+ restoreScopeProperty(exchange, ExchangePropertyKey.EXCEPTION_HANDLED,
previousHandled);
callback.done(true);
return true;
}
+ /**
+ * Sets the property back to the value of the enclosing scope, or removes
it when that scope had none (as
+ * {@code TryProcessor} does for a nested doTry).
+ */
+ private static void restoreScopeProperty(Exchange exchange,
ExchangePropertyKey key, Object value) {
+ if (value != null) {
+ exchange.setProperty(key, value);
+ } else {
+ exchange.removeProperty(key);
+ }
+ }
+
@Override
protected void doBuild() throws Exception {
ObjectHelper.notNull(camelContext, "CamelContext", this);
diff --git
a/components/camel-microprofile/camel-microprofile-fault-tolerance/src/test/java/org/apache/camel/component/microprofile/faulttolerance/FaultToleranceOuterTryScopeTest.java
b/components/camel-microprofile/camel-microprofile-fault-tolerance/src/test/java/org/apache/camel/component/microprofile/faulttolerance/FaultToleranceOuterTryScopeTest.java
new file mode 100644
index 000000000000..4267fb19afd8
--- /dev/null
+++
b/components/camel-microprofile/camel-microprofile-fault-tolerance/src/test/java/org/apache/camel/component/microprofile/faulttolerance/FaultToleranceOuterTryScopeTest.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.microprofile.faulttolerance;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+class FaultToleranceOuterTryScopeTest extends CamelTestSupport {
+ private final AtomicInteger attempts = new AtomicInteger();
+
+ @Test
+ void synchronousCircuitBreakerPreservesOuterTry() throws Exception {
+ check("direct:sync");
+ }
+
+ @Test
+ void fallbackPreservesOuterTry() throws Exception {
+ check("direct:fallback");
+ }
+
+ @Test
+ void laterFailureAfterRecoveredFallbackIsCaught() throws Exception {
+ getMockEndpoint("mock:caught").expectedMessageCount(1);
+ Exchange out = template.send("direct:afterFallback", e ->
e.getMessage().setBody("Hello"));
+ MockEndpoint.assertIsSatisfied(context);
+ assertNull(out.getException(), "the doCatch must catch the failure
after the fallback");
+ }
+
+ @Test
+ void failingFallbackIsCaught() throws Exception {
+ // the fallback itself fails: the doCatch catches it, as for a circuit
breaker without a fallback
+ getMockEndpoint("mock:caught").expectedMessageCount(1);
+ Exchange out = template.send("direct:failingFallback", e ->
e.getMessage().setBody("Hello"));
+ MockEndpoint.assertIsSatisfied(context);
+ assertNull(out.getException(), "the doCatch must catch the failure of
the fallback");
+ }
+
+ @Test
+ void circuitBreakerInDoCatchKeepsTheFailureHandled() throws Exception {
+ // a failure after the circuit breaker, inside the doCatch, is not
caught by a later doCatch of the same doTry
+ getMockEndpoint("mock:first").expectedMessageCount(1);
+ getMockEndpoint("mock:second").expectedMessageCount(0);
+ Exchange out = template.send("direct:inCatch", e ->
e.getMessage().setBody("Hello"));
+ MockEndpoint.assertIsSatisfied(context);
+ assertInstanceOf(IllegalArgumentException.class, out.getException());
+ }
+
+ @Test
+ void outsideTryUsesGlobalErrorHandler() throws Exception {
+ getMockEndpoint("mock:dead").expectedMessageCount(1);
+ template.sendBody("direct:outside", "Hello");
+ MockEndpoint.assertIsSatisfied(context);
+ assertEquals(3, attempts.get());
+ assertEquals(0, getMockEndpoint("mock:caught").getReceivedCounter());
+ }
+
+ private void check(String uri) throws Exception {
+ getMockEndpoint("mock:caught").expectedMessageCount(1);
+ getMockEndpoint("mock:dead").expectedMessageCount(0);
+ template.sendBody(uri, "Hello");
+ MockEndpoint.assertIsSatisfied(context);
+ assertEquals(1, attempts.get(), "The outer doTry owns the recipient
list failure");
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
errorHandler(deadLetterChannel("mock:dead").maximumRedeliveries(2).redeliveryDelay(0));
+ from("direct:fail").errorHandler(noErrorHandler()).process(e
-> {
+ attempts.incrementAndGet();
+ throw new IllegalStateException("Failure after circuit
breaker");
+ });
+ from("direct:outside").circuitBreaker().to("mock:inside").end()
+ .recipientList(constant("direct:fail"));
+ from("direct:sync").doTry()
+ .circuitBreaker().to("mock:inside").end()
+ .recipientList(constant("direct:fail")).endDoTry()
+
.doCatch(IllegalStateException.class).to("mock:caught").end();
+ from("direct:fallback").doTry()
+ .circuitBreaker().throwException(new
IllegalArgumentException("Recover me"))
+ .onFallback().setBody(constant("Recovered")).end()
+ .recipientList(constant("direct:fail")).endDoTry()
+
.doCatch(IllegalStateException.class).to("mock:caught").end();
+ from("direct:afterFallback").doTry()
+ .circuitBreaker().throwException(new
IllegalArgumentException("Recover me"))
+ .onFallback().setBody(constant("Recovered")).end()
+ .throwException(new IllegalStateException("Failure after
the fallback")).endDoTry()
+
.doCatch(IllegalStateException.class).to("mock:caught").end();
+ from("direct:failingFallback").doTry()
+ .circuitBreaker().throwException(new
IllegalArgumentException("Recover me"))
+ .onFallback().throwException(new
IllegalStateException("The fallback fails")).end()
+ .endDoTry()
+
.doCatch(IllegalStateException.class).to("mock:caught").end();
+ from("direct:inCatch").doTry()
+ .throwException(new IllegalStateException("First failure"))
+ .doCatch(IllegalStateException.class).to("mock:first")
+ .circuitBreaker().to("mock:inside").end()
+ .throwException(new IllegalArgumentException("Failure
in the doCatch")).endDoTry()
+ .doCatch(Exception.class).to("mock:second").end();
+ }
+ };
+ }
+}
diff --git
a/components/camel-resilience4j/src/main/java/org/apache/camel/component/resilience4j/ResilienceProcessor.java
b/components/camel-resilience4j/src/main/java/org/apache/camel/component/resilience4j/ResilienceProcessor.java
index 792964d56106..c8dcdc38ff72 100644
---
a/components/camel-resilience4j/src/main/java/org/apache/camel/component/resilience4j/ResilienceProcessor.java
+++
b/components/camel-resilience4j/src/main/java/org/apache/camel/component/resilience4j/ResilienceProcessor.java
@@ -566,12 +566,31 @@ public class ResilienceProcessor extends
BaseProcessorSupport
public boolean process(Exchange exchange, AsyncCallback callback) {
// run this as if we run inside try .. catch so there is no regular
// Camel error handler
+ Object previousTryRouteBlock =
exchange.getProperty(ExchangePropertyKey.TRY_ROUTE_BLOCK);
+ Object previousHandled =
exchange.removeProperty(ExchangePropertyKey.EXCEPTION_HANDLED);
exchange.setProperty(ExchangePropertyKey.TRY_ROUTE_BLOCK, true);
+ AsyncCallback restoreScope = doneSync -> {
+ restoreScopeProperty(exchange,
ExchangePropertyKey.TRY_ROUTE_BLOCK, previousTryRouteBlock);
+ restoreScopeProperty(exchange,
ExchangePropertyKey.EXCEPTION_HANDLED, previousHandled);
+ callback.done(doneSync);
+ };
if (asynchronous) {
- return processAsync(exchange, callback);
+ return processAsync(exchange, restoreScope);
+ }
+ return processSync(exchange, restoreScope);
+ }
+
+ /**
+ * Sets the property back to the value of the enclosing scope, or removes
it when that scope had none (as
+ * {@code TryProcessor} does for a nested doTry).
+ */
+ private static void restoreScopeProperty(Exchange exchange,
ExchangePropertyKey key, Object value) {
+ if (value != null) {
+ exchange.setProperty(key, value);
+ } else {
+ exchange.removeProperty(key);
}
- return processSync(exchange, callback);
}
private boolean processSync(Exchange exchange, AsyncCallback callback) {
@@ -632,7 +651,6 @@ public class ResilienceProcessor extends
BaseProcessorSupport
failed);
}
- exchange.removeProperty(ExchangePropertyKey.TRY_ROUTE_BLOCK);
callback.done(true);
return true;
}
@@ -696,7 +714,6 @@ public class ResilienceProcessor extends
BaseProcessorSupport
exchange.getExchangeId(), id, failed);
}
- exchange.removeProperty(ExchangePropertyKey.TRY_ROUTE_BLOCK);
callback.done(false);
});
return false;
@@ -706,7 +723,6 @@ public class ResilienceProcessor extends
BaseProcessorSupport
if (fallbackTask != null) {
fallbackTaskFactory.release(fallbackTask);
}
- exchange.removeProperty(ExchangePropertyKey.TRY_ROUTE_BLOCK);
callback.done(true);
return true;
}
diff --git
a/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceOuterTryScopeTest.java
b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceOuterTryScopeTest.java
new file mode 100644
index 000000000000..c139bc573084
--- /dev/null
+++
b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceOuterTryScopeTest.java
@@ -0,0 +1,152 @@
+/*
+ * 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.component.mock.MockEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+class ResilienceOuterTryScopeTest extends CamelTestSupport {
+ private final AtomicInteger attempts = new AtomicInteger();
+
+ @Test
+ void synchronousCircuitBreakerPreservesOuterTry() throws Exception {
+ check("direct:sync");
+ }
+
+ @Test
+ void asynchronousCircuitBreakerPreservesOuterTry() throws Exception {
+ check("direct:async");
+ }
+
+ @Test
+ void fallbackPreservesOuterTry() throws Exception {
+ check("direct:fallback");
+ }
+
+ @Test
+ void laterFailureAfterRecoveredFallbackIsCaught() throws Exception {
+ getMockEndpoint("mock:caught").expectedMessageCount(1);
+ Exchange out = template.send("direct:afterFallback", e ->
e.getMessage().setBody("Hello"));
+ MockEndpoint.assertIsSatisfied(context);
+ assertNull(out.getException(), "the doCatch must catch the failure
after the fallback");
+ }
+
+ @Test
+ void asynchronousLaterFailureAfterRecoveredFallbackIsCaught() throws
Exception {
+ getMockEndpoint("mock:caught").expectedMessageCount(1);
+ Exchange out = template.send("direct:asyncAfterFallback", e ->
e.getMessage().setBody("Hello"));
+ MockEndpoint.assertIsSatisfied(context);
+ assertNull(out.getException(), "the doCatch must catch the failure
after the asynchronous fallback");
+ }
+
+ @Test
+ void failingFallbackIsCaught() throws Exception {
+ // the fallback itself fails: the doCatch catches it, as for a circuit
breaker without a fallback
+ getMockEndpoint("mock:caught").expectedMessageCount(1);
+ Exchange out = template.send("direct:failingFallback", e ->
e.getMessage().setBody("Hello"));
+ MockEndpoint.assertIsSatisfied(context);
+ assertNull(out.getException(), "the doCatch must catch the failure of
the fallback");
+ }
+
+ @Test
+ void circuitBreakerInDoCatchKeepsTheFailureHandled() throws Exception {
+ // a failure after the circuit breaker, inside the doCatch, is not
caught by a later doCatch of the same doTry
+ getMockEndpoint("mock:first").expectedMessageCount(1);
+ getMockEndpoint("mock:second").expectedMessageCount(0);
+ Exchange out = template.send("direct:inCatch", e ->
e.getMessage().setBody("Hello"));
+ MockEndpoint.assertIsSatisfied(context);
+ assertInstanceOf(IllegalArgumentException.class, out.getException());
+ }
+
+ @Test
+ void outsideTryUsesGlobalErrorHandler() throws Exception {
+ getMockEndpoint("mock:dead").expectedMessageCount(1);
+ template.sendBody("direct:outside", "Hello");
+ MockEndpoint.assertIsSatisfied(context);
+ assertEquals(3, attempts.get());
+ assertEquals(0, getMockEndpoint("mock:caught").getReceivedCounter());
+ }
+
+ private void check(String uri) throws Exception {
+ getMockEndpoint("mock:caught").expectedMessageCount(1);
+ getMockEndpoint("mock:dead").expectedMessageCount(0);
+ template.sendBody(uri, "Hello");
+ MockEndpoint.assertIsSatisfied(context);
+ assertEquals(1, attempts.get(), "The outer doTry owns the recipient
list failure");
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
errorHandler(deadLetterChannel("mock:dead").maximumRedeliveries(2).redeliveryDelay(0));
+ from("direct:fail").errorHandler(noErrorHandler()).process(e
-> {
+ attempts.incrementAndGet();
+ throw new IllegalStateException("Failure after circuit
breaker");
+ });
+ from("direct:outside").circuitBreaker().to("mock:inside").end()
+ .recipientList(constant("direct:fail"));
+ from("direct:sync").doTry()
+ .circuitBreaker().to("mock:inside").end()
+ .recipientList(constant("direct:fail")).endDoTry()
+
.doCatch(IllegalStateException.class).to("mock:caught").end();
+ from("direct:async").doTry()
+
.circuitBreaker().resilience4jConfiguration().asynchronous(true).end()
+ .to("mock:inside").end()
+ .recipientList(constant("direct:fail")).endDoTry()
+
.doCatch(IllegalStateException.class).to("mock:caught").end();
+ from("direct:fallback").doTry()
+ .circuitBreaker().throwException(new
IllegalArgumentException("Recover me"))
+ .onFallback().setBody(constant("Recovered")).end()
+ .recipientList(constant("direct:fail")).endDoTry()
+
.doCatch(IllegalStateException.class).to("mock:caught").end();
+ from("direct:afterFallback").doTry()
+ .circuitBreaker().throwException(new
IllegalArgumentException("Recover me"))
+ .onFallback().setBody(constant("Recovered")).end()
+ .throwException(new IllegalStateException("Failure after
the fallback")).endDoTry()
+
.doCatch(IllegalStateException.class).to("mock:caught").end();
+ from("direct:asyncAfterFallback").doTry()
+
.circuitBreaker().resilience4jConfiguration().asynchronous(true).end()
+ .throwException(new IllegalArgumentException("Recover
me"))
+ .onFallback().setBody(constant("Recovered")).end()
+ .throwException(new IllegalStateException("Failure after
the fallback")).endDoTry()
+
.doCatch(IllegalStateException.class).to("mock:caught").end();
+ from("direct:failingFallback").doTry()
+ .circuitBreaker().throwException(new
IllegalArgumentException("Recover me"))
+ .onFallback().throwException(new
IllegalStateException("The fallback fails")).end()
+ .endDoTry()
+
.doCatch(IllegalStateException.class).to("mock:caught").end();
+ from("direct:inCatch").doTry()
+ .throwException(new IllegalStateException("First failure"))
+ .doCatch(IllegalStateException.class).to("mock:first")
+ .circuitBreaker().to("mock:inside").end()
+ .throwException(new IllegalArgumentException("Failure
in the doCatch")).endDoTry()
+ .doCatch(Exception.class).to("mock:second").end();
+ }
+ };
+ }
+}
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 a827b2f9e442..e62114bd364f 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
@@ -489,6 +489,17 @@ which embeds its own relocated copy of Vert.x. The `vertx`
option of `camel-aran
See the
https://github.com/fabric8io/kubernetes-client/blob/main/doc/MIGRATION-v8.md[Kubernetes
client 8 migration guide]
for the other changes.
+=== camel-resilience4j, camel-microprofile-fault-tolerance - a circuit breaker
in a doTry
+
+A circuit breaker in the steps of a `doTry` no longer ends that `doTry` for
the steps after it. Before, when the
+circuit breaker was done, a later routing slip, dynamic router, multicast or
recipient list of the same `doTry` used
+the error handler of the route (redelivery and the dead letter channel)
instead of the `doCatch`. And after the
+fallback of the circuit breaker had recovered from a failure, no `doCatch` of
that `doTry` caught a later failure:
+the failure was thrown from the `doTry`. Now such failures go to the
`doCatch`, as for the other steps of the `doTry`.
+A failure of the fallback itself is now also caught by the `doCatch` (before,
it was thrown from the `doTry`).
+
+To have those steps use the error handler of the route, move them out of the
`doTry` block.
+
=== camel-aws2-kinesis - each endpoint uses its own client
Each `aws2-kinesis` endpoint now uses the Kinesis client of its own
configuration (`region`, credentials,
