CAMEL-26422: camel-resilience4j, camel-microprofile-fault-tolerance - a circuit breaker no longer ends the enclosing doTry

components/camel-microprofile/camel-microprofile-fault-tolerance/src/main/java/org/apache/camel/component/microprofile/faulttolerance/FaultToleranceProcessor.java

@@ -281,6 +281,8 @@ public boolean hasNext() {
     @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 boolean process(Exchange exchange, AsyncCallback callback) {
             }
         }
 
-        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);

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();
+            }
+        };
+    }
+}

components/camel-resilience4j/src/main/java/org/apache/camel/component/resilience4j/ResilienceProcessor.java

@@ -566,12 +566,31 @@ public boolean hasNext() {
     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 @@ private boolean processSync(Exchange exchange, AsyncCallback callback) {
                     failed);
         }
 
-        exchange.removeProperty(ExchangePropertyKey.TRY_ROUTE_BLOCK);
         callback.done(true);
         return true;
     }
@@ -696,7 +714,6 @@ private boolean processAsync(Exchange exchange, AsyncCallback callback) {
                             exchange.getExchangeId(), id, failed);
                 }
 
-                exchange.removeProperty(ExchangePropertyKey.TRY_ROUTE_BLOCK);
                 callback.done(false);
             });
             return false;
@@ -706,7 +723,6 @@ private boolean processAsync(Exchange exchange, AsyncCallback callback) {
             if (fallbackTask != null) {
                 fallbackTaskFactory.release(fallbackTask);
             }
-            exchange.removeProperty(ExchangePropertyKey.TRY_ROUTE_BLOCK);
             callback.done(true);
             return true;
         }

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();
+            }
+        };
+    }
+}

docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc

@@ -340,6 +340,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,