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 31fb84bc7213c..6044525d355e4 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 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); 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 0000000000000..4267fb19afd87 --- /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 792964d56106d..c8dcdc38ff724 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 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; } 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 0000000000000..c139bc5730845 --- /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 70a0b28507010..0c46d7f2ebdd1 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 @@ -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,