This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 24cd63f10fb2 CAMEL-25682: camel-resilience4j - a timed-out call waits
for a write-back of the task that is in progress
24cd63f10fb2 is described below
commit 24cd63f10fb22a520b1a83fba00a47459422943b
Author: allthingssecurity <[email protected]>
AuthorDate: Fri Oct 16 20:29:21 2026 +0530
CAMEL-25682: camel-resilience4j - a timed-out call waits for a write-back
of the task that is in progress
[CAMEL-25682](https://issues.apache.org/jira/browse/CAMEL-25682)
CAMEL-24134 added a write guard so that the task of the circuit breaker
does not write its result into the exchange after a timeout handed the exchange
to the fallback. The task claims the guard with `compareAndSet(false, true)`
and then copies its result (`ResilienceProcessor.processTask`, and
`processTaskAsync` in asynchronous mode). The fallback task only did
`exchangeWriteGuard.set(true)` before writing the exchange. When the timeout
fires after the task has won the guard and befo [...]
- with `onFallback`, the body of the fallback is replaced by the body of
the task, or `CamelCircuitBreakerResponseFromFallback` is set back to `false`
(and `CamelCircuitBreakerResponseSuccessfulExecution` to `true`) after the
fallback set them, while the exchange already continues in the route;
- without fallback, `ExchangeHelper.copyResults` ends with
`setException(null)` and clears the `TimeoutException` that the caller has just
set;
- headers and properties are written by two threads.
The caller does not wait for the task here (the task runs in
`CompletableFuture.supplyAsync` behind `TimeLimiter.decorateFutureSupplier`, or
`decorateCompletionStage` in async mode), so the window is the time the task
needs to copy its result; it needs a task that ends at about the moment of the
timeout.
Closes #27921
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../resilience4j/ResilienceProcessor.java | 121 ++++++++---
.../ResilienceTimeoutDuringWriteBackTest.java | 239 +++++++++++++++++++++
2 files changed, 332 insertions(+), 28 deletions(-)
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 c35f096ee343..792964d56106 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
@@ -26,8 +26,9 @@ import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.Supplier;
@@ -581,7 +582,7 @@ public class ResilienceProcessor extends
BaseProcessorSupport
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);
+ ExchangeWriteGuard exchangeWriteGuard = new ExchangeWriteGuard();
task.exchangeWriteGuard = exchangeWriteGuard;
fallbackTask.exchangeWriteGuard = exchangeWriteGuard;
final CircuitBreakerTask ftask = task; // annoying final java
thingy!
@@ -640,7 +641,7 @@ public class ResilienceProcessor extends
BaseProcessorSupport
CircuitBreakerFallbackTask fallbackTask = null;
try {
fallbackTask = (CircuitBreakerFallbackTask)
fallbackTaskFactory.acquire(exchange, callback);
- AtomicBoolean exchangeWriteGuard = new AtomicBoolean(false);
+ ExchangeWriteGuard exchangeWriteGuard = new ExchangeWriteGuard();
fallbackTask.exchangeWriteGuard = exchangeWriteGuard;
if (LOG.isTraceEnabled()) {
@@ -716,7 +717,7 @@ public class ResilienceProcessor extends
BaseProcessorSupport
exchange.setProperty(ExchangePropertyKey.CIRCUIT_BREAKER_RESPONSE_STATE,
circuitBreaker.getState().name());
}
- private Exchange processTask(Exchange exchange, AtomicBoolean
exchangeWriteGuard) {
+ private Exchange processTask(Exchange exchange, ExchangeWriteGuard
exchangeWriteGuard) {
String state = circuitBreaker.getState().name();
Exchange copy = null;
@@ -748,19 +749,27 @@ public class ResilienceProcessor extends
BaseProcessorSupport
processor.process(copy);
// 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);
+ if (exchangeWriteGuard.tryClaimForWriteBack()) {
+ try {
+ 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);
+ }
+ } finally {
+ exchangeWriteGuard.writeBackDone();
}
}
} catch (Exception e) {
- if (exchangeWriteGuard.compareAndSet(false, true)) {
- exchange.setException(e);
+ if (exchangeWriteGuard.tryClaimForWriteBack()) {
+ try {
+ exchange.setException(e);
+ } finally {
+ exchangeWriteGuard.writeBackDone();
+ }
}
} finally {
// must done uow
@@ -782,7 +791,7 @@ public class ResilienceProcessor extends
BaseProcessorSupport
}
private CompletableFuture<Exchange> processTaskAsync(
- Exchange exchange, AtomicBoolean exchangeWriteGuard,
CompletableFuture<Exchange> future) {
+ Exchange exchange, ExchangeWriteGuard exchangeWriteGuard,
CompletableFuture<Exchange> future) {
String state = circuitBreaker.getState().name();
Exchange copy = null;
@@ -809,18 +818,27 @@ public class ResilienceProcessor extends
BaseProcessorSupport
AsyncProcessor asyncProcessor =
AsyncProcessorConverterHelper.convert(processor);
asyncProcessor.process(copy, doneSync -> {
try {
- if (exchangeWriteGuard.compareAndSet(false, true)) {
- if (fCopy.getException() != null) {
- exchange.setException(fCopy.getException());
- } else {
- ExchangeHelper.copyResults(exchange, fCopy);
-
exchange.setProperty(ExchangePropertyKey.CIRCUIT_BREAKER_RESPONSE_SUCCESSFUL_EXECUTION,
true);
-
exchange.setProperty(ExchangePropertyKey.CIRCUIT_BREAKER_RESPONSE_FROM_FALLBACK,
false);
+ if (exchangeWriteGuard.tryClaimForWriteBack()) {
+ try {
+ if (fCopy.getException() != null) {
+ exchange.setException(fCopy.getException());
+ } else {
+ ExchangeHelper.copyResults(exchange, fCopy);
+
exchange.setProperty(ExchangePropertyKey.CIRCUIT_BREAKER_RESPONSE_SUCCESSFUL_EXECUTION,
+ true);
+
exchange.setProperty(ExchangePropertyKey.CIRCUIT_BREAKER_RESPONSE_FROM_FALLBACK,
false);
+ }
+ } finally {
+ exchangeWriteGuard.writeBackDone();
}
}
} catch (Exception e) {
- if (exchangeWriteGuard.compareAndSet(false, true)) {
- exchange.setException(e);
+ if (exchangeWriteGuard.tryClaimForWriteBack()) {
+ try {
+ exchange.setException(e);
+ } finally {
+ exchangeWriteGuard.writeBackDone();
+ }
}
} finally {
UnitOfWorkHelper.doneUow(fUow, fCopy);
@@ -845,10 +863,56 @@ public class ResilienceProcessor extends
BaseProcessorSupport
return future;
}
+ /**
+ * Makes the write-back of the task and the takeover by the caller (the
fallback, or the timeout exception) mutually
+ * exclusive, for one call. The task writes its result only if it claims
the exchange before the caller. The caller
+ * always writes: if it takes over while the write-back of the task is in
progress, it waits until that is done, and
+ * then works on the exchange that holds the result of the task.
+ */
+ private static final class ExchangeWriteGuard {
+ private final Lock lock = new ReentrantLock();
+ private boolean claimed;
+
+ /**
+ * Claims the exchange for the write-back of the task. When this
returns true, the lock is held until
+ * {@link #writeBackDone()} is called.
+ */
+ boolean tryClaimForWriteBack() {
+ lock.lock();
+ if (claimed) {
+ lock.unlock();
+ return false;
+ }
+ claimed = true;
+ return true;
+ }
+
+ /**
+ * Ends the write-back of the task, after {@link
#tryClaimForWriteBack()} returned true. Must be called in a
+ * {@code finally} block, as the lock stays held until then.
+ */
+ void writeBackDone() {
+ lock.unlock();
+ }
+
+ /**
+ * Claims the exchange for the caller. This always succeeds; it waits
for a write-back of the task that is in
+ * progress, and stops the task from writing afterwards.
+ */
+ void claim() {
+ lock.lock();
+ try {
+ claimed = true;
+ } finally {
+ lock.unlock();
+ }
+ }
+ }
+
private final class CircuitBreakerTask implements PooledExchangeTask,
Callable<Exchange>, Supplier<Exchange> {
private Exchange exchange;
- private AtomicBoolean exchangeWriteGuard;
+ private ExchangeWriteGuard exchangeWriteGuard;
@Override
public void prepare(Exchange exchange, AsyncCallback callback) {
@@ -885,7 +949,7 @@ public class ResilienceProcessor extends
BaseProcessorSupport
private final class CircuitBreakerFallbackTask implements
PooledExchangeTask, Function<Throwable, Exchange> {
private Exchange exchange;
- private AtomicBoolean exchangeWriteGuard;
+ private ExchangeWriteGuard exchangeWriteGuard;
@Override
public void prepare(Exchange exchange, AsyncCallback callback) {
@@ -906,9 +970,10 @@ public class ResilienceProcessor extends
BaseProcessorSupport
@Override
public Exchange apply(Throwable throwable) {
- // prevent the worker thread from writing results back to the
original exchange
+ // prevent the worker thread from writing results back to the
original exchange (waits for a
+ // write-back that the worker has already started)
if (exchangeWriteGuard != null) {
- exchangeWriteGuard.set(true);
+ exchangeWriteGuard.claim();
}
String state = circuitBreaker.getState().name();
diff --git
a/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceTimeoutDuringWriteBackTest.java
b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceTimeoutDuringWriteBackTest.java
new file mode 100644
index 000000000000..a9a654f71095
--- /dev/null
+++
b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceTimeoutDuringWriteBackTest.java
@@ -0,0 +1,239 @@
+/*
+ * 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.Arrays;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePropertyKey;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.support.DefaultMessage;
+import org.apache.camel.support.ExchangeHelper;
+import org.apache.camel.support.SynchronizationAdapter;
+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.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The timeout fires while the task is copying its result into the exchange
(it has already won the write guard). The
+ * caller must not change the exchange at the same time (fallback or timeout
exception): it waits until the copy is
+ * done, then sets the timeout result or runs the fallback on the exchange,
which then holds the result of the task (as
+ * when the timeout fires just after the copy). The copy does not change the
exchange after the caller took it over.
+ * Sync and async ({@code asynchronous=true}) mode.
+ * <p>
+ * The task holds its copy at a gate inside {@code ExchangeHelper.copyResults}
(the first read of the body of its
+ * result). The gate is opened when the caller has returned (it did not wait
for the copy), or when the caller has not
+ * returned 2 seconds after the timeout (it waits for the copy).
+ */
+class ResilienceTimeoutDuringWriteBackTest extends CamelTestSupport {
+
+ private static final int TIMEOUT = 500;
+
+ private final Gate gate = new Gate();
+ private final CountDownLatch taskDone = new CountDownLatch(1);
+
+ @Test
+ void testFallbackIsNotOverwrittenByWriteBackInProgress() throws Exception {
+ assertFallbackResult(sendWhileTaskHoldsWriteBack("direct:fallback"));
+ }
+
+ @Test
+ void testTimeoutExceptionIsNotClearedByWriteBackInProgress() throws
Exception {
+ assertTimeoutResult(sendWhileTaskHoldsWriteBack("direct:nofallback"));
+ }
+
+ @Test
+ void testAsyncFallbackIsNotOverwrittenByWriteBackInProgress() throws
Exception {
+
assertFallbackResult(sendWhileTaskHoldsWriteBack("direct:asyncFallback"));
+ }
+
+ @Test
+ void testAsyncTimeoutExceptionIsNotClearedByWriteBackInProgress() throws
Exception {
+
assertTimeoutResult(sendWhileTaskHoldsWriteBack("direct:asyncNofallback"));
+ }
+
+ private static void assertFallbackResult(Exchange exchange) {
+ assertEquals("Fallback result",
exchange.getMessage().getBody(String.class));
+ // the fallback ran after the write-back of the task, on the exchange
that holds the result of the task
+ assertEquals("task", exchange.getMessage().getHeader("Written-By"));
+ assertEquals(Boolean.TRUE,
exchange.getProperty(ExchangePropertyKey.CIRCUIT_BREAKER_RESPONSE_FROM_FALLBACK));
+ assertEquals(Boolean.FALSE,
+
exchange.getProperty(ExchangePropertyKey.CIRCUIT_BREAKER_RESPONSE_SUCCESSFUL_EXECUTION));
+ }
+
+ private static void assertTimeoutResult(Exchange exchange) {
+ assertInstanceOf(TimeoutException.class, exchange.getException());
+ // the timeout was set after the write-back of the task, on the
exchange that holds the result of the task
+ assertEquals("task", exchange.getMessage().getHeader("Written-By"));
+ assertEquals(Boolean.TRUE,
exchange.getProperty(ExchangePropertyKey.CIRCUIT_BREAKER_RESPONSE_TIMED_OUT));
+ assertEquals(Boolean.FALSE,
+
exchange.getProperty(ExchangePropertyKey.CIRCUIT_BREAKER_RESPONSE_SUCCESSFUL_EXECUTION));
+ }
+
+ private Exchange sendWhileTaskHoldsWriteBack(String uri) throws Exception {
+ Future<Exchange> reply = template.asyncSend(uri, e ->
e.getIn().setBody("Hello"));
+
+ assertTrue(gate.entered.await(10, TimeUnit.SECONDS), "the task should
reach the write-back");
+ StackTraceElement[] where = gate.where.get();
+ assertTrue(Arrays.stream(where).anyMatch(f ->
f.getMethodName().equals("doCopyResults")),
+ "the gate should hold the task inside copyResults, not at " +
Arrays.toString(where));
+
+ // the caller times out while the task holds the gate: either it
returns without waiting for the task, or it
+ // waits for the task, then it has not returned 2 seconds after the
timeout
+ try {
+ reply.get(TIMEOUT + 2000, TimeUnit.MILLISECONDS);
+ } catch (TimeoutException e) {
+ // the caller waits for the write-back of the task
+ }
+ gate.open.countDown();
+
+ Exchange exchange = reply.get(10, TimeUnit.SECONDS);
+ assertNotNull(exchange);
+ assertTrue(taskDone.await(10, TimeUnit.SECONDS), "the task should be
done");
+ return exchange;
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:fallback")
+ .circuitBreaker()
+
.resilience4jConfiguration().timeoutEnabled(true).timeoutDuration(TIMEOUT).end()
+ .process(e -> holdWriteBack(e))
+ .onFallback()
+ .setBody(constant("Fallback result"))
+ .end();
+
+ from("direct:nofallback")
+ .circuitBreaker()
+
.resilience4jConfiguration().timeoutEnabled(true).timeoutDuration(TIMEOUT).end()
+ .process(e -> holdWriteBack(e))
+ .end();
+
+ from("direct:asyncFallback")
+ .circuitBreaker()
+
.resilience4jConfiguration().asynchronous(true).timeoutEnabled(true).timeoutDuration(TIMEOUT)
+ .end()
+ .process(e -> holdWriteBack(e))
+ .onFallback()
+ .setBody(constant("Fallback result"))
+ .end();
+
+ from("direct:asyncNofallback")
+ .circuitBreaker()
+
.resilience4jConfiguration().asynchronous(true).timeoutEnabled(true).timeoutDuration(TIMEOUT)
+ .end()
+ .process(e -> holdWriteBack(e))
+ .end();
+ }
+ };
+ }
+
+ private void holdWriteBack(Exchange copy) {
+ GatedMessage result = new GatedMessage(copy, gate);
+ result.setBody("Task result");
+ result.setHeader("Written-By", "task");
+ copy.setIn(result);
+ // done when the task has finished with its copy of the exchange
+ copy.getExchangeExtension().addOnCompletion(new
SynchronizationAdapter() {
+ @Override
+ public void onDone(Exchange exchange) {
+ taskDone.countDown();
+ }
+ });
+ result.armed = true;
+ }
+
+ private static final class Gate {
+ final CountDownLatch entered = new CountDownLatch(1);
+ final CountDownLatch open = new CountDownLatch(1);
+ final AtomicReference<StackTraceElement[]> where = new
AtomicReference<>();
+
+ void pass() {
+ where.set(Thread.currentThread().getStackTrace());
+ entered.countDown();
+ // keep waiting if the task thread is interrupted, bounded, and
keep the interrupt status
+ boolean interrupted = false;
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(20);
+ try {
+ while (true) {
+ long left = deadline - System.nanoTime();
+ if (left <= 0) {
+ return;
+ }
+ try {
+ if (open.await(left, TimeUnit.NANOSECONDS)) {
+ return;
+ }
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ }
+ } finally {
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+ }
+
+ /**
+ * The result message of the task: the first read of its body by {@code
ExchangeHelper.copyResults} after the task
+ * is done holds at the gate (other reads, such as the stream caching
advice, pass).
+ */
+ private static final class GatedMessage extends DefaultMessage {
+ private final Gate gate;
+ volatile boolean armed;
+
+ GatedMessage(Exchange exchange, Gate gate) {
+ super(exchange);
+ this.gate = gate;
+ }
+
+ @Override
+ public Object getBody() {
+ if (armed && calledByCopyResults()) {
+ armed = false;
+ gate.pass();
+ }
+ return super.getBody();
+ }
+
+ private static boolean calledByCopyResults() {
+ return StackWalker.getInstance()
+ .walk(frames -> frames.anyMatch(f ->
f.getClassName().equals(ExchangeHelper.class.getName())
+ && f.getMethodName().equals("doCopyResults")));
+ }
+
+ @Override
+ public DefaultMessage newInstance() {
+ return new DefaultMessage(getExchange());
+ }
+ }
+}
