CAMEL-24134: Circuit Breaker EIP - prevent timed-out worker from writing back to exchange

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

@@ -21,6 +21,7 @@
 import java.util.List;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import io.smallrye.faulttolerance.api.CircuitBreakerMaintenance;
 import io.smallrye.faulttolerance.api.CircuitBreakerState;
@@ -231,13 +232,19 @@ public boolean process(Exchange exchange, AsyncCallback callback) {
         // run this as if we run inside try / catch so there is no regular Camel error handler
         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
+        // after a timeout has triggered fallback processing on the caller thread
+        AtomicBoolean exchangeWriteGuard = new AtomicBoolean(false);
+        task.exchangeWriteGuard = exchangeWriteGuard;
         CircuitBreakerFallbackTask fallbackTask = null;
 
         try {
             // Run the fault-tolerant task within the guard
             try {
                 typedGuard.call(task);
             } catch (Exception e) {
+                // prevent the worker thread from writing results back to the exchange
+                exchangeWriteGuard.set(true);
                 // Do fallback if applicable. Note that a fallback handler is not configured on the TypedGuard builder
                 // and is instead invoked manually here since we need access to the message exchange on each FaultToleranceProcessor.process call
                 if (fallbackProcessor != null) {
@@ -387,6 +394,7 @@ protected void doShutdown() throws Exception {
     private final class CircuitBreakerTask implements PooledExchangeTask, Callable<Exchange> {
 
         private Exchange exchange;
+        private AtomicBoolean exchangeWriteGuard;
 
         @Override
         public void prepare(Exchange exchange, AsyncCallback callback) {
@@ -397,6 +405,7 @@ public void prepare(Exchange exchange, AsyncCallback callback) {
         @Override
         public void reset() {
             this.exchange = null;
+            this.exchangeWriteGuard = null;
         }
 
         @Override
@@ -436,30 +445,36 @@ public Exchange call() throws Exception {
                 // process the processor until its fully done
                 processor.process(copy);
 
-                // handle the processing result
-                if (copy.getException() != null) {
-                    exchange.setException(copy.getException());
-                } else {
-                    // copy the result as it's regarded as success
-                    ExchangeHelper.copyResults(exchange, copy);
-                    exchange.setProperty(ExchangePropertyKey.CIRCUIT_BREAKER_RESPONSE_SUCCESSFUL_EXECUTION, true);
-                    exchange.setProperty(ExchangePropertyKey.CIRCUIT_BREAKER_RESPONSE_FROM_FALLBACK, false);
-                    String state = getCircuitBreakerState();
-                    if (state != null) {
-                        exchange.setProperty(ExchangePropertyKey.CIRCUIT_BREAKER_RESPONSE_STATE, state);
+                // handle the processing result, but only write back if the fallback has not taken over
+                if (exchangeWriteGuard == null || exchangeWriteGuard.compareAndSet(false, true)) {
+                    if (copy.getException() != null) {
+                        exchange.setException(copy.getException());
+                    } else {
+                        // copy the result as it's regarded as success
+                        ExchangeHelper.copyResults(exchange, copy);
+                        exchange.setProperty(ExchangePropertyKey.CIRCUIT_BREAKER_RESPONSE_SUCCESSFUL_EXECUTION, true);
+                        exchange.setProperty(ExchangePropertyKey.CIRCUIT_BREAKER_RESPONSE_FROM_FALLBACK, false);
+                        String state = getCircuitBreakerState();
+                        if (state != null) {
+                            exchange.setProperty(ExchangePropertyKey.CIRCUIT_BREAKER_RESPONSE_STATE, state);
+                        }
                     }
                 }
             } catch (Exception e) {
-                exchange.setException(e);
+                if (exchangeWriteGuard == null || exchangeWriteGuard.compareAndSet(false, true)) {
+                    exchange.setException(e);
+                }
             } finally {
                 // must done uow
                 UnitOfWorkHelper.doneUow(uow, copy);
                 // remember any thrown exception
                 cause = exchange.getException();
             }
 
-            // and release exchange back in pool
-            processorExchangeFactory.release(exchange);
+            // and release correlated copy back in pool
+            if (copy != null) {
+                processorExchangeFactory.release(copy);
+            }
 
             if (cause != null) {
                 // throw exception so fault tolerance knows it was a failure

components/camel-microprofile/camel-microprofile-fault-tolerance/src/test/java/org/apache/camel/component/microprofile/faulttolerance/FaultTolerancePooledProcessorExchangeFactoryTest.java

@@ -0,0 +1,67 @@
+/*
+ * 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.TimeUnit;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.engine.PooledExchangeFactory;
+import org.apache.camel.impl.engine.PooledProcessorExchangeFactory;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+public class FaultTolerancePooledProcessorExchangeFactoryTest extends CamelTestSupport {
+
+    @Override
+    protected CamelContext createCamelContext() throws Exception {
+        CamelContext context = super.createCamelContext();
+        context.getCamelContextExtension().setExchangeFactory(new PooledExchangeFactory());
+        context.getCamelContextExtension().setProcessorExchangeFactory(new PooledProcessorExchangeFactory());
+        return context;
+    }
+
+    @Test
+    public void testBodySurvivesCircuitBreaker() throws Exception {
+        getMockEndpoint("mock:result").expectedMinimumMessageCount(1);
+        getMockEndpoint("mock:result").allMessages().body().isEqualTo("Bye World");
+
+        context.getRouteController().startAllRoutes();
+
+        MockEndpoint.assertIsSatisfied(context, 30, TimeUnit.SECONDS);
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() {
+        return new RouteBuilder() {
+            @Override
+            public void configure() {
+                from("timer:foo?repeatCount=1").autoStartup(false)
+                        .setBody(constant("Hello World"))
+                        .circuitBreaker()
+                            .to("direct:foo")
+                        .end()
+                        .to("log:result")
+                        .to("mock:result");
+
+                from("direct:foo")
+                        .transform().constant("Bye World");
+            }
+        };
+    }
+}

components/camel-microprofile/camel-microprofile-fault-tolerance/src/test/java/org/apache/camel/component/microprofile/faulttolerance/FaultToleranceTimeoutWriteBackRaceTest.java

@@ -0,0 +1,83 @@
+/*
+ * 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.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.RoutesBuilder;
+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.assertTrue;
+
+/**
+ * Test that a timed-out worker thread does not write its late results back to the original exchange, racing with
+ * fallback processing on the caller thread.
+ */
+public class FaultToleranceTimeoutWriteBackRaceTest extends CamelTestSupport {
+
+    private final CountDownLatch workerDone = new CountDownLatch(1);
+
+    @Test
+    public void testTimeoutWorkerDoesNotOverwriteFallback() throws Exception {
+        getMockEndpoint("mock:result").expectedBodiesReceived("Fallback result");
+
+        template.sendBody("direct:start", "Hello");
+
+        MockEndpoint.assertIsSatisfied(context);
+
+        // wait for the worker thread to fully complete its delayed processing
+        assertTrue(workerDone.await(10, TimeUnit.SECONDS), "Worker thread should complete");
+
+        // after the worker has completed, verify the exchange was not corrupted by a late write-back
+        Exchange received = getMockEndpoint("mock:result").getReceivedExchanges().get(0);
+        assertEquals("Fallback result", received.getIn().getBody(String.class));
+    }
+
+    @Override
+    protected RoutesBuilder createRouteBuilder() {
+        return new RouteBuilder() {
+            @Override
+            public void configure() {
+                from("direct:start")
+                        .circuitBreaker()
+                            .faultToleranceConfiguration()
+                                .timeoutEnabled(true)
+                                .timeoutDuration(500)
+                            .end()
+                            .process(exchange -> {
+                                try {
+                                    // simulate slow processing that outlasts the timeout
+                                    Thread.sleep(3000);
+                                exchange.getIn().setBody("Worker result");
+                            } finally {
+                                workerDone.countDown();
+                            }
+                        })
+                        .onFallback()
+                        .setBody(constant("Fallback result"))
+                        .end()
+                        .to("mock:result");
+            }
+        };
+    }
+}

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

@@ -23,6 +23,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
@@ -513,6 +514,11 @@ public boolean process(Exchange exchange, AsyncCallback callback) {
         try {
             fallbackTask = (CircuitBreakerFallbackTask) fallbackTaskFactory.acquire(exchange, callback);
             task = (CircuitBreakerTask) taskFactory.acquire(exchange, callback);
+            // guard to prevent the worker thread from writing results back to the original exchange
+            // after a timeout has triggered fallback processing on the caller thread
+            AtomicBoolean exchangeWriteGuard = new AtomicBoolean(false);
+            task.exchangeWriteGuard = exchangeWriteGuard;
+            fallbackTask.exchangeWriteGuard = exchangeWriteGuard;
             final CircuitBreakerTask ftask = task; // annoying final java thingy!
             Callable<Exchange> callable;
 
@@ -562,7 +568,7 @@ private void successState(Exchange exchange) {
         exchange.setProperty(ExchangePropertyKey.CIRCUIT_BREAKER_RESPONSE_STATE, circuitBreaker.getState().name());
     }
 
-    private Exchange processTask(Exchange exchange) {
+    private Exchange processTask(Exchange exchange, AtomicBoolean exchangeWriteGuard) {
         String state = circuitBreaker.getState().name();
 
         Exchange copy = null;
@@ -593,26 +599,32 @@ private Exchange processTask(Exchange exchange) {
             // process the processor until its fully done
             processor.process(copy);
 
-            // handle the processing result
-            if (copy.getException() != null) {
-                exchange.setException(copy.getException());
-            } else {
-                // copy the result as its regarded as success
-                ExchangeHelper.copyResults(exchange, copy);
-                exchange.setProperty(ExchangePropertyKey.CIRCUIT_BREAKER_RESPONSE_SUCCESSFUL_EXECUTION, true);
-                exchange.setProperty(ExchangePropertyKey.CIRCUIT_BREAKER_RESPONSE_FROM_FALLBACK, false);
+            // handle the processing result, but only write back if the fallback has not taken over
+            if (exchangeWriteGuard.compareAndSet(false, true)) {
+                if (copy.getException() != null) {
+                    exchange.setException(copy.getException());
+                } else {
+                    // copy the result as its regarded as success
+                    ExchangeHelper.copyResults(exchange, copy);
+                    exchange.setProperty(ExchangePropertyKey.CIRCUIT_BREAKER_RESPONSE_SUCCESSFUL_EXECUTION, true);
+                    exchange.setProperty(ExchangePropertyKey.CIRCUIT_BREAKER_RESPONSE_FROM_FALLBACK, false);
+                }
             }
         } catch (Exception e) {
-            exchange.setException(e);
+            if (exchangeWriteGuard.compareAndSet(false, true)) {
+                exchange.setException(e);
+            }
         } finally {
             // must done uow
             UnitOfWorkHelper.doneUow(uow, copy);
             // remember any thrown exception
             cause = exchange.getException();
         }
 
-        // and release exchange back in pool
-        processorExchangeFactory.release(exchange);
+        // and release correlated copy back in pool
+        if (copy != null) {
+            processorExchangeFactory.release(copy);
+        }
 
         if (cause != null) {
             // throw exception so resilient4j know it was a failure
@@ -624,6 +636,7 @@ private Exchange processTask(Exchange exchange) {
     private final class CircuitBreakerTask implements PooledExchangeTask, Callable<Exchange>, Supplier<Exchange> {
 
         private Exchange exchange;
+        private AtomicBoolean exchangeWriteGuard;
 
         @Override
         public void prepare(Exchange exchange, AsyncCallback callback) {
@@ -634,6 +647,7 @@ public void prepare(Exchange exchange, AsyncCallback callback) {
         @Override
         public void reset() {
             this.exchange = null;
+            this.exchangeWriteGuard = null;
         }
 
         @Override
@@ -645,20 +659,21 @@ public void run() {
         public Exchange call() throws Exception {
             // this task is either use as callable or supplier
             // therefore we must call process task before returning the response
-            return processTask(exchange);
+            return processTask(exchange, exchangeWriteGuard);
         }
 
         @Override
         public Exchange get() {
             // this task is either use as callable or supplier
             // therefore we must call process task before returning the response
-            return processTask(exchange);
+            return processTask(exchange, exchangeWriteGuard);
         }
     }
 
     private final class CircuitBreakerFallbackTask implements PooledExchangeTask, Function<Throwable, Exchange> {
 
         private Exchange exchange;
+        private AtomicBoolean exchangeWriteGuard;
 
         @Override
         public void prepare(Exchange exchange, AsyncCallback callback) {
@@ -669,6 +684,7 @@ public void prepare(Exchange exchange, AsyncCallback callback) {
         @Override
         public void reset() {
             this.exchange = null;
+            this.exchangeWriteGuard = null;
         }
 
         @Override
@@ -678,6 +694,11 @@ public void run() {
 
         @Override
         public Exchange apply(Throwable throwable) {
+            // prevent the worker thread from writing results back to the original exchange
+            if (exchangeWriteGuard != null) {
+                exchangeWriteGuard.set(true);
+            }
+
             String state = circuitBreaker.getState().name();
             exchange.setProperty(ExchangePropertyKey.CIRCUIT_BREAKER_RESPONSE_STATE, state);