From 9d1028dbe3a21500b2d27468bb53111209c0928d Mon Sep 17 00:00:00 2001 From: Claus Ibsen Date: Thu, 16 Jul 2026 14:41:58 +0200 Subject: [PATCH 1/2] CAMEL-24133: Circuit Breaker EIP - Release correlated copy instead of original exchange in pooled mode Co-Authored-By: Claude Opus 4.6 Signed-off-by: Claus Ibsen --- .../FaultToleranceProcessor.java | 6 +- ...ncePooledProcessorExchangeFactoryTest.java | 67 +++++++++++++++++++ .../resilience4j/ResilienceProcessor.java | 6 +- ...ncePooledProcessorExchangeFactoryTest.java | 67 +++++++++++++++++++ 4 files changed, 142 insertions(+), 4 deletions(-) create mode 100644 components/camel-microprofile/camel-microprofile-fault-tolerance/src/test/java/org/apache/camel/component/microprofile/faulttolerance/FaultTolerancePooledProcessorExchangeFactoryTest.java create mode 100644 components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResiliencePooledProcessorExchangeFactoryTest.java 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 072165aca2cb6..018ebe5c9bb8f 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 @@ -458,8 +458,10 @@ public Exchange call() throws 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 diff --git a/components/camel-microprofile/camel-microprofile-fault-tolerance/src/test/java/org/apache/camel/component/microprofile/faulttolerance/FaultTolerancePooledProcessorExchangeFactoryTest.java b/components/camel-microprofile/camel-microprofile-fault-tolerance/src/test/java/org/apache/camel/component/microprofile/faulttolerance/FaultTolerancePooledProcessorExchangeFactoryTest.java new file mode 100644 index 0000000000000..27406fe2d2981 --- /dev/null +++ b/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"); + } + }; + } +} 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 aa2e1c8dead54..6183d405c9d12 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 @@ -611,8 +611,10 @@ private Exchange processTask(Exchange exchange) { 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 diff --git a/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResiliencePooledProcessorExchangeFactoryTest.java b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResiliencePooledProcessorExchangeFactoryTest.java new file mode 100644 index 0000000000000..2fc30e0502f20 --- /dev/null +++ b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResiliencePooledProcessorExchangeFactoryTest.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.resilience4j; + +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 ResiliencePooledProcessorExchangeFactoryTest 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"); + } + }; + } +} From 8217fc055abff410484a30d5944f505b15534453 Mon Sep 17 00:00:00 2001 From: Claus Ibsen Date: Thu, 16 Jul 2026 14:53:36 +0200 Subject: [PATCH 2/2] CAMEL-24134: Circuit Breaker EIP - prevent timed-out worker from writing back to exchange Both ResilienceProcessor and FaultToleranceProcessor had a race condition where the worker thread could write results back to the original exchange after a timeout triggered fallback processing on the caller thread. This could overwrite fallback results, reinstate cleared exceptions, or corrupt the exchange through non-atomic interleaving of ExchangeHelper.copyResults. Fix: add a shared AtomicBoolean guard between the worker task and the fallback. The first to claim it (via CAS) gets to write to the exchange; the other skips the write-back entirely. Co-Authored-By: Claude Opus 4.6 Signed-off-by: Claus Ibsen --- .../FaultToleranceProcessor.java | 37 ++++++--- ...aultToleranceTimeoutWriteBackRaceTest.java | 83 +++++++++++++++++++ .../resilience4j/ResilienceProcessor.java | 43 +++++++--- .../ResilienceTimeoutWriteBackRaceTest.java | 83 +++++++++++++++++++ 4 files changed, 222 insertions(+), 24 deletions(-) create mode 100644 components/camel-microprofile/camel-microprofile-fault-tolerance/src/test/java/org/apache/camel/component/microprofile/faulttolerance/FaultToleranceTimeoutWriteBackRaceTest.java create mode 100644 components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceTimeoutWriteBackRaceTest.java 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 018ebe5c9bb8f..2b7cb3a65833a 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 @@ -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,6 +232,10 @@ 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 { @@ -238,6 +243,8 @@ public boolean process(Exchange exchange, AsyncCallback callback) { 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 { 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,21 +445,25 @@ 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); diff --git a/components/camel-microprofile/camel-microprofile-fault-tolerance/src/test/java/org/apache/camel/component/microprofile/faulttolerance/FaultToleranceTimeoutWriteBackRaceTest.java b/components/camel-microprofile/camel-microprofile-fault-tolerance/src/test/java/org/apache/camel/component/microprofile/faulttolerance/FaultToleranceTimeoutWriteBackRaceTest.java new file mode 100644 index 0000000000000..90b7c450591ea --- /dev/null +++ b/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"); + } + }; + } +} 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 6183d405c9d12..0e5a8b9292129 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 @@ -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 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,17 +599,21 @@ 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); @@ -626,6 +636,7 @@ private Exchange processTask(Exchange exchange) { private final class CircuitBreakerTask implements PooledExchangeTask, Callable, Supplier { private Exchange exchange; + private AtomicBoolean exchangeWriteGuard; @Override public void prepare(Exchange exchange, AsyncCallback callback) { @@ -636,6 +647,7 @@ public void prepare(Exchange exchange, AsyncCallback callback) { @Override public void reset() { this.exchange = null; + this.exchangeWriteGuard = null; } @Override @@ -647,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 { private Exchange exchange; + private AtomicBoolean exchangeWriteGuard; @Override public void prepare(Exchange exchange, AsyncCallback callback) { @@ -671,6 +684,7 @@ public void prepare(Exchange exchange, AsyncCallback callback) { @Override public void reset() { this.exchange = null; + this.exchangeWriteGuard = null; } @Override @@ -680,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); diff --git a/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceTimeoutWriteBackRaceTest.java b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceTimeoutWriteBackRaceTest.java new file mode 100644 index 0000000000000..15f59538f6369 --- /dev/null +++ b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceTimeoutWriteBackRaceTest.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.resilience4j; + +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 ResilienceTimeoutWriteBackRaceTest 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() + .resilience4jConfiguration() + .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"); + } + }; + } +}