This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24134 in repository https://gitbox.apache.org/repos/asf/camel.git
commit 8217fc055abff410484a30d5944f505b15534453 Author: Claus Ibsen <[email protected]> AuthorDate: Thu Jul 16 14:53:36 2026 +0200 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 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../faulttolerance/FaultToleranceProcessor.java | 37 ++++++---- .../FaultToleranceTimeoutWriteBackRaceTest.java | 83 ++++++++++++++++++++++ .../resilience4j/ResilienceProcessor.java | 43 +++++++---- .../ResilienceTimeoutWriteBackRaceTest.java | 83 ++++++++++++++++++++++ 4 files changed, 222 insertions(+), 24 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 018ebe5c9bb8..2b7cb3a65833 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.ArrayList; 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 class FaultToleranceProcessor extends BaseProcessorSupport // 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 class FaultToleranceProcessor extends BaseProcessorSupport 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 @@ public class FaultToleranceProcessor extends BaseProcessorSupport 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 class FaultToleranceProcessor extends BaseProcessorSupport @Override public void reset() { this.exchange = null; + this.exchangeWriteGuard = null; } @Override @@ -436,21 +445,25 @@ public class FaultToleranceProcessor extends BaseProcessorSupport // 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 000000000000..90b7c450591e --- /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 6183d405c9d1..0e5a8b929212 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.Callable; 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 class ResilienceProcessor extends BaseProcessorSupport 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 @@ public class ResilienceProcessor extends BaseProcessorSupport 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 @@ public class ResilienceProcessor extends BaseProcessorSupport // 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 @@ public class ResilienceProcessor extends BaseProcessorSupport private final class CircuitBreakerTask implements PooledExchangeTask, Callable<Exchange>, Supplier<Exchange> { private Exchange exchange; + private AtomicBoolean exchangeWriteGuard; @Override public void prepare(Exchange exchange, AsyncCallback callback) { @@ -636,6 +647,7 @@ public class ResilienceProcessor extends BaseProcessorSupport @Override public void reset() { this.exchange = null; + this.exchangeWriteGuard = null; } @Override @@ -647,20 +659,21 @@ public class ResilienceProcessor extends BaseProcessorSupport 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) { @@ -671,6 +684,7 @@ public class ResilienceProcessor extends BaseProcessorSupport @Override public void reset() { this.exchange = null; + this.exchangeWriteGuard = null; } @Override @@ -680,6 +694,11 @@ public class ResilienceProcessor extends BaseProcessorSupport @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 000000000000..15f59538f636 --- /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"); + } + }; + } +}
