You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2026/07/22 07:13:29 UTC
(camel) branch main updated: CAMEL-24209: camel-resilience4j - Add async (non-blocking) processing
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 c5ab46f01248 CAMEL-24209: camel-resilience4j - Add async (non-blocking) processing
c5ab46f01248 is described below
commit c5ab46f01248bb69fd49523ee35457c30b92a3cd
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Wed Jul 22 09:13:21 2026 +0200
CAMEL-24209: camel-resilience4j - Add async (non-blocking) processing
Adds a new 'asynchronous' option (default false) to the Resilience4j circuit
breaker. When enabled, uses Resilience4j's CompletionStage-based decorators
for true non-blocking processing — the caller thread is released immediately
and processing completes asynchronously. Sync and async code paths are in
separate methods for clarity. When async mode is used with timeout, the
timeoutExecutorService must be a ScheduledExecutorService, validated at
startup. Includes 6 new tests covering async success, timeout, fallback,
write-back race, bulkhead, and invalid executor scenarios.
Closes #24977
Co-Authored-By: Claude Opus 4.6 <no...@anthropic.com>
---
.../main/camel-main-configuration-metadata.json | 1 +
.../catalog/models/resilience4jConfiguration.json | 15 +-
.../apache/camel/catalog/schemas/camel-spring.xsd | 12 ++
.../apache/camel/catalog/schemas/camel-xml-io.xsd | 12 ++
.../resilience4j/ResilienceProcessor.java | 169 +++++++++++++++++++++
.../component/resilience4j/ResilienceReifier.java | 16 +-
.../resilience4j/ResilienceAsyncBulkheadTest.java | 70 +++++++++
.../ResilienceAsyncInvalidExecutorTest.java | 86 +++++++++++
.../resilience4j/ResilienceAsyncRouteOkTest.java | 70 +++++++++
.../resilience4j/ResilienceAsyncTimeoutTest.java | 80 ++++++++++
.../ResilienceAsyncTimeoutWithFallbackTest.java | 68 +++++++++
.../ResilienceAsyncTimeoutWriteBackRaceTest.java | 80 ++++++++++
.../docs/modules/eips/pages/resilience4j-eip.adoc | 81 ++++++++++
...ilience4jConfigurationDefinitionConfigurer.java | 4 +
.../camel/model/resilience4jConfiguration.json | 15 +-
.../model/Resilience4jConfigurationCommon.java | 16 ++
.../model/Resilience4jConfigurationDefinition.java | 16 ++
.../apache/camel/java/out/JavaDslModelWriter.java | 1 +
...ilience4jConfigurationPropertiesConfigurer.java | 4 +
.../camel-main-configuration-metadata.json | 1 +
core/camel-main/src/main/docs/main.adoc | 3 +-
.../main/Resilience4jConfigurationProperties.java | 23 +++
.../java/org/apache/camel/xml/in/ModelParser.java | 1 +
.../java/org/apache/camel/xml/out/ModelWriter.java | 1 +
.../org/apache/camel/yaml/out/YamlModelWriter.java | 1 +
.../dsl/yaml/deserializers/ModelDeserializers.java | 6 +
.../resources/schema/camelYamlDsl-canonical.json | 6 +
.../generated/resources/schema/camelYamlDsl.json | 6 +
28 files changed, 848 insertions(+), 16 deletions(-)
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/main/camel-main-configuration-metadata.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/main/camel-main-configuration-metadata.json
index 1acde96827c6..ff6f31dbb39e 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/main/camel-main-configuration-metadata.json
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/main/camel-main-configuration-metadata.json
@@ -345,6 +345,7 @@
{ "name": "camel.opentelemetry2.includePatterns", "required": false, "description": "Sets include pattern(s) that will explicitly enable tracing for Camel processors that matches the pattern. Multiple patterns can be separated by comma. All processors included by default if nothing is specified.", "sourceType": "org.apache.camel.main.Otel2ConfigurationProperties", "type": "string", "javaType": "java.lang.String", "secret": false },
{ "name": "camel.opentelemetry2.instrumentationName", "required": true, "description": "A name uniquely identifying the instrumentation scope, such as the instrumentation library, package, or fully qualified class name. Must not be null.", "sourceType": "org.apache.camel.main.Otel2ConfigurationProperties", "type": "string", "javaType": "java.lang.String", "defaultValue": "camel", "secret": false },
{ "name": "camel.opentelemetry2.traceProcessors", "required": false, "description": "Setting this to true will create new telemetry spans for each Camel custom Processors. Use the excludePattern property to filter out Processors.", "sourceType": "org.apache.camel.main.Otel2ConfigurationProperties", "type": "boolean", "javaType": "boolean", "defaultValue": false, "secret": false },
+ { "name": "camel.resilience4j.asynchronous", "required": false, "description": "Whether to use asynchronous (non-blocking) processing using Resilience4j's CompletionStage-based decorators. When enabled, the caller thread is released immediately and processing completes asynchronously.", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "boolean", "javaType": "java.lang.Boolean", "defaultValue": false, "secret": false },
{ "name": "camel.resilience4j.automaticTransitionFromOpenToHalfOpenEnabled", "required": false, "description": "Enables automatic transition from OPEN to HALF_OPEN state once the waitDurationInOpenState has passed.", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "boolean", "javaType": "java.lang.Boolean", "defaultValue": false, "secret": false },
{ "name": "camel.resilience4j.bulkheadEnabled", "required": false, "description": "Whether bulkhead is enabled or not on the circuit breaker.", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "boolean", "javaType": "java.lang.Boolean", "defaultValue": false, "secret": false },
{ "name": "camel.resilience4j.bulkheadFairCallHandlingEnabled", "required": false, "description": "Configures whether the bulkhead uses a fair calling strategy. When enabled (default), a fair strategy guarantees the order of incoming requests (FIFO). When disabled, no ordering is guaranteed and may improve throughput.", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "boolean", "javaType": "java.lang.Boolean", "defaultValue": true, "secret": false },
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/resilience4jConfiguration.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/resilience4jConfiguration.json
index 9b1452808a39..0ac5b7b290fc 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/resilience4jConfiguration.json
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/resilience4jConfiguration.json
@@ -32,12 +32,13 @@
"bulkheadMaxConcurrentCalls": { "index": 17, "kind": "attribute", "displayName": "Bulkhead Max Concurrent Calls", "group": "common", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 25, "description": "Configures the max amount of concurrent calls the bulkhead will support." },
"bulkheadMaxWaitDuration": { "index": 18, "kind": "attribute", "displayName": "Bulkhead Max Wait Duration", "group": "advanced", "label": "advanced", "required": false, "type": "duration", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "0", "description": "Configures a maximum amount of time which the calling thread will wait to enter the bulkhead. The default is 0 (no waiting)." },
"bulkheadFairCallHandlingEnabled": { "index": 19, "kind": "attribute", "displayName": "Bulkhead Fair Call Handling Enabled", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Configures whether the bulkhead uses a fair calling strategy. When enabled (default), a fair strategy guarantees the order of incoming requests (FIFO). Wh [...]
- "timeoutEnabled": { "index": 20, "kind": "attribute", "displayName": "Timeout Enabled", "group": "common", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether timeout is enabled or not on the circuit breaker." },
- "timeoutExecutorService": { "index": 21, "kind": "attribute", "displayName": "Timeout Executor Service", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "java.util.concurrent.ExecutorService", "deprecated": false, "autowired": false, "secret": false, "description": "References to a custom thread pool to use when timeout is enabled (uses ForkJoinPool.commonPool() by default)." },
- "timeoutDuration": { "index": 22, "kind": "attribute", "displayName": "Timeout Duration", "group": "common", "required": false, "type": "duration", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "1000", "description": "Configures the thread execution timeout. Default value is 1 second." },
- "timeoutCancelRunningFuture": { "index": 23, "kind": "attribute", "displayName": "Timeout Cancel Running Future", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Configures whether cancel is called on the running future. Defaults to true." },
- "micrometerEnabled": { "index": 24, "kind": "attribute", "displayName": "Micrometer Enabled", "group": "common", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether to enable collecting statistics using Micrometer for all circuit breaker instances. This is a global setting (configure via camel.resilience4j.micrometerEnabled=true) and requires adding camel-resili [...]
- "recordException": { "index": 25, "kind": "element", "displayName": "Record Exception", "group": "advanced", "label": "advanced", "required": false, "type": "array", "javaType": "java.util.List<java.lang.String>", "deprecated": false, "autowired": false, "secret": false, "description": "Configure a list of exceptions that are recorded as a failure and thus increase the failure rate. Any exception matching or inheriting from one of the list counts as a failure, unless explicitly ignor [...]
- "ignoreException": { "index": 26, "kind": "element", "displayName": "Ignore Exception", "group": "advanced", "label": "advanced", "required": false, "type": "array", "javaType": "java.util.List<java.lang.String>", "deprecated": false, "autowired": false, "secret": false, "description": "Configure a list of exceptions that are ignored and neither count as a failure nor success. Any exception matching or inheriting from one of the list will not count as a failure nor success, even if t [...]
+ "asynchronous": { "index": 20, "kind": "attribute", "displayName": "Asynchronous", "group": "common", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether to use asynchronous (non-blocking) processing with CompletionStage-based circuit breaker decorators. When enabled, the circuit breaker releases the caller thread immediately and completes processing asynchronou [...]
+ "timeoutEnabled": { "index": 21, "kind": "attribute", "displayName": "Timeout Enabled", "group": "common", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether timeout is enabled or not on the circuit breaker." },
+ "timeoutExecutorService": { "index": 22, "kind": "attribute", "displayName": "Timeout Executor Service", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "java.util.concurrent.ExecutorService", "deprecated": false, "autowired": false, "secret": false, "description": "References to a custom thread pool to use when timeout is enabled (uses ForkJoinPool.commonPool() by default)." },
+ "timeoutDuration": { "index": 23, "kind": "attribute", "displayName": "Timeout Duration", "group": "common", "required": false, "type": "duration", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "1000", "description": "Configures the thread execution timeout. Default value is 1 second." },
+ "timeoutCancelRunningFuture": { "index": 24, "kind": "attribute", "displayName": "Timeout Cancel Running Future", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Configures whether cancel is called on the running future. Defaults to true." },
+ "micrometerEnabled": { "index": 25, "kind": "attribute", "displayName": "Micrometer Enabled", "group": "common", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether to enable collecting statistics using Micrometer for all circuit breaker instances. This is a global setting (configure via camel.resilience4j.micrometerEnabled=true) and requires adding camel-resili [...]
+ "recordException": { "index": 26, "kind": "element", "displayName": "Record Exception", "group": "advanced", "label": "advanced", "required": false, "type": "array", "javaType": "java.util.List<java.lang.String>", "deprecated": false, "autowired": false, "secret": false, "description": "Configure a list of exceptions that are recorded as a failure and thus increase the failure rate. Any exception matching or inheriting from one of the list counts as a failure, unless explicitly ignor [...]
+ "ignoreException": { "index": 27, "kind": "element", "displayName": "Ignore Exception", "group": "advanced", "label": "advanced", "required": false, "type": "array", "javaType": "java.util.List<java.lang.String>", "deprecated": false, "autowired": false, "secret": false, "description": "Configure a list of exceptions that are ignored and neither count as a failure nor success. Any exception matching or inheriting from one of the list will not count as a failure nor success, even if t [...]
}
}
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-spring.xsd b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-spring.xsd
index 6e46cb1ed12d..f59c75ce5420 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-spring.xsd
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-spring.xsd
@@ -4572,6 +4572,18 @@ waiting). Default value: 0
Configures whether the bulkhead uses a fair calling strategy. When enabled (default), a fair strategy guarantees the
order of incoming requests (FIFO). When disabled, no ordering is guaranteed and may improve throughput. Default value:
true
+]]>
+ </xs:documentation>
+ </xs:annotation>
+ </xs:attribute>
+ <xs:attribute name="asynchronous" type="xs:string">
+ <xs:annotation>
+ <xs:documentation xml:lang="en">
+<![CDATA[
+Whether to use asynchronous (non-blocking) processing with CompletionStage-based circuit breaker decorators. When
+enabled, the circuit breaker releases the caller thread immediately and completes processing asynchronously. This is
+most valuable when the downstream processor supports asynchronous processing (e.g. Netty HTTP, Kafka). When used with
+timeout, the timeoutExecutorService must be a ScheduledExecutorService. Default value: false
]]>
</xs:documentation>
</xs:annotation>
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-xml-io.xsd b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-xml-io.xsd
index 54833296f891..b0ae756bd4e0 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-xml-io.xsd
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-xml-io.xsd
@@ -3682,6 +3682,18 @@ waiting). Default value: 0
Configures whether the bulkhead uses a fair calling strategy. When enabled (default), a fair strategy guarantees the
order of incoming requests (FIFO). When disabled, no ordering is guaranteed and may improve throughput. Default value:
true
+]]>
+ </xs:documentation>
+ </xs:annotation>
+ </xs:attribute>
+ <xs:attribute name="asynchronous" type="xs:string">
+ <xs:annotation>
+ <xs:documentation xml:lang="en">
+<![CDATA[
+Whether to use asynchronous (non-blocking) processing with CompletionStage-based circuit breaker decorators. When
+enabled, the circuit breaker releases the caller thread immediately and completes processing asynchronously. This is
+most valuable when the downstream processor supports asynchronous processing (e.g. Netty HTTP, Kafka). When used with
+timeout, the timeoutExecutorService must be a ScheduledExecutorService. Default value: false
]]>
</xs:documentation>
</xs:annotation>
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 eade71731573..932aa5e6f054 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
@@ -21,7 +21,10 @@ import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+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.function.Function;
@@ -40,6 +43,7 @@ import io.github.resilience4j.timelimiter.TimeLimiter;
import io.github.resilience4j.timelimiter.TimeLimiterConfig;
import io.github.resilience4j.timelimiter.TimeLimiterRegistry;
import org.apache.camel.AsyncCallback;
+import org.apache.camel.AsyncProcessor;
import org.apache.camel.CamelContext;
import org.apache.camel.CamelContextAware;
import org.apache.camel.Exchange;
@@ -62,6 +66,7 @@ import org.apache.camel.spi.IdAware;
import org.apache.camel.spi.ProcessorExchangeFactory;
import org.apache.camel.spi.RouteIdAware;
import org.apache.camel.spi.UnitOfWork;
+import org.apache.camel.support.AsyncProcessorConverterHelper;
import org.apache.camel.support.ExchangeHelper;
import org.apache.camel.support.PluginHelper;
import org.apache.camel.support.UnitOfWorkHelper;
@@ -97,8 +102,10 @@ public class ResilienceProcessor extends BaseProcessorSupport
private final boolean throwExceptionWhenHalfOpenOrOpenState;
private final Predicate<Throwable> recordPredicate;
private final Predicate<Throwable> ignorePredicate;
+ private boolean asynchronous;
private boolean shutdownExecutorService;
private ExecutorService executorService;
+ private ScheduledExecutorService scheduledExecutorService;
private ProcessorExchangeFactory processorExchangeFactory;
private PooledExchangeTaskFactory taskFactory;
private PooledExchangeTaskFactory fallbackTaskFactory;
@@ -290,6 +297,23 @@ public class ResilienceProcessor extends BaseProcessorSupport
this.executorService = executorService;
}
+ public ScheduledExecutorService getScheduledExecutorService() {
+ return scheduledExecutorService;
+ }
+
+ public void setScheduledExecutorService(ScheduledExecutorService scheduledExecutorService) {
+ this.scheduledExecutorService = scheduledExecutorService;
+ }
+
+ @ManagedAttribute(description = "Whether asynchronous (non-blocking) processing is enabled")
+ public boolean isAsynchronous() {
+ return asynchronous;
+ }
+
+ public void setAsynchronous(boolean asynchronous) {
+ this.asynchronous = asynchronous;
+ }
+
@Override
public String getTraceLabel() {
return "resilience4j";
@@ -519,6 +543,13 @@ public class ResilienceProcessor extends BaseProcessorSupport
// Camel error handler
exchange.setProperty(ExchangePropertyKey.TRY_ROUTE_BLOCK, true);
+ if (asynchronous) {
+ return processAsync(exchange, callback);
+ }
+ return processSync(exchange, callback);
+ }
+
+ private boolean processSync(Exchange exchange, AsyncCallback callback) {
CircuitBreakerFallbackTask fallbackTask = null;
CircuitBreakerTask task = null;
try {
@@ -575,6 +606,80 @@ public class ResilienceProcessor extends BaseProcessorSupport
return true;
}
+ private boolean processAsync(Exchange exchange, AsyncCallback callback) {
+ CircuitBreakerFallbackTask fallbackTask = null;
+ try {
+ fallbackTask = (CircuitBreakerFallbackTask) fallbackTaskFactory.acquire(exchange, callback);
+ AtomicBoolean exchangeWriteGuard = new AtomicBoolean(false);
+ fallbackTask.exchangeWriteGuard = exchangeWriteGuard;
+
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Processing exchange: {} using circuit breaker: {} (async)", exchange.getExchangeId(), id);
+ }
+
+ // build the CompletionStage supplier for the main task
+ Supplier<CompletionStage<Exchange>> supplier;
+ if (timeLimiter != null) {
+ // with timeout: submit work to executor so the supplier returns immediately
+ // and the TimeLimiter can schedule the timeout check
+ supplier = () -> {
+ CompletableFuture<Exchange> future = new CompletableFuture<>();
+ executorService.submit(() -> processTaskAsync(exchange, exchangeWriteGuard, future));
+ return future;
+ };
+ } else {
+ // without timeout: run directly on the caller thread
+ // if the downstream processor is async, it returns immediately
+ supplier = () -> processTaskAsync(exchange, exchangeWriteGuard, new CompletableFuture<>());
+ }
+
+ // decorate with resilience4j CompletionStage decorators
+ if (timeLimiter != null) {
+ supplier = TimeLimiter.decorateCompletionStage(timeLimiter, scheduledExecutorService, supplier);
+ }
+ if (bulkhead != null) {
+ supplier = Bulkhead.decorateCompletionStage(bulkhead, supplier);
+ }
+ supplier = CircuitBreaker.decorateCompletionStage(circuitBreaker, supplier);
+
+ // trigger the chain and handle completion
+ final CircuitBreakerFallbackTask fFallbackTask = fallbackTask;
+ supplier.get().whenComplete((result, throwable) -> {
+ try {
+ if (throwable != null) {
+ Throwable cause = throwable instanceof CompletionException ? throwable.getCause() : throwable;
+ fFallbackTask.apply(cause);
+ } else {
+ successState(result);
+ }
+ } catch (Exception e) {
+ exchange.setException(e);
+ } finally {
+ fallbackTaskFactory.release(fFallbackTask);
+ }
+
+ if (LOG.isTraceEnabled()) {
+ boolean failed = exchange.isFailed();
+ LOG.trace("Processing exchange: {} using circuit breaker: {} complete (async, failed: {})",
+ exchange.getExchangeId(), id, failed);
+ }
+
+ exchange.removeProperty(ExchangePropertyKey.TRY_ROUTE_BLOCK);
+ callback.done(false);
+ });
+ return false;
+ } catch (Exception e) {
+ // setup failure before the CompletionStage was created
+ exchange.setException(e);
+ if (fallbackTask != null) {
+ fallbackTaskFactory.release(fallbackTask);
+ }
+ exchange.removeProperty(ExchangePropertyKey.TRY_ROUTE_BLOCK);
+ callback.done(true);
+ return true;
+ }
+ }
+
private void successState(Exchange exchange) {
exchange.setProperty(ExchangePropertyKey.CIRCUIT_BREAKER_RESPONSE_SUCCESSFUL_EXECUTION, true);
exchange.setProperty(ExchangePropertyKey.CIRCUIT_BREAKER_RESPONSE_STATE, circuitBreaker.getState().name());
@@ -645,6 +750,70 @@ public class ResilienceProcessor extends BaseProcessorSupport
return exchange;
}
+ private CompletableFuture<Exchange> processTaskAsync(
+ Exchange exchange, AtomicBoolean exchangeWriteGuard, CompletableFuture<Exchange> future) {
+ String state = circuitBreaker.getState().name();
+
+ Exchange copy = null;
+ UnitOfWork uow = null;
+ try {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Processing exchange: {} using circuit breaker ({}):{} with processor: {} (async)",
+ exchange.getExchangeId(), state, id, processor);
+ }
+ copy = processorExchangeFactory.createCorrelatedCopy(exchange, false);
+ if (copy.getUnitOfWork() != null) {
+ uow = copy.getUnitOfWork();
+ } else {
+ uow = PluginHelper.getUnitOfWorkFactory(copy.getContext()).createUnitOfWork(copy);
+ copy.getExchangeExtension().setUnitOfWork(uow);
+ Route route = ExchangeHelper.getRoute(exchange);
+ if (route != null) {
+ uow.pushRoute(route);
+ }
+ }
+
+ final Exchange fCopy = copy;
+ final UnitOfWork fUow = uow;
+ 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);
+ }
+ }
+ } catch (Exception e) {
+ if (exchangeWriteGuard.compareAndSet(false, true)) {
+ exchange.setException(e);
+ }
+ } finally {
+ UnitOfWorkHelper.doneUow(fUow, fCopy);
+ processorExchangeFactory.release(fCopy);
+ }
+
+ Throwable cause = exchange.getException();
+ if (cause != null) {
+ future.completeExceptionally(cause);
+ } else {
+ future.complete(exchange);
+ }
+ });
+ } catch (Exception e) {
+ // cleanup on failure to set up async processing
+ if (copy != null) {
+ UnitOfWorkHelper.doneUow(uow, copy);
+ processorExchangeFactory.release(copy);
+ }
+ future.completeExceptionally(e);
+ }
+ return future;
+ }
+
private final class CircuitBreakerTask implements PooledExchangeTask, Callable<Exchange>, Supplier<Exchange> {
private Exchange exchange;
diff --git a/components/camel-resilience4j/src/main/java/org/apache/camel/component/resilience4j/ResilienceReifier.java b/components/camel-resilience4j/src/main/java/org/apache/camel/component/resilience4j/ResilienceReifier.java
index f198261372d2..d0b05fa9d175 100644
--- a/components/camel-resilience4j/src/main/java/org/apache/camel/component/resilience4j/ResilienceReifier.java
+++ b/components/camel-resilience4j/src/main/java/org/apache/camel/component/resilience4j/ResilienceReifier.java
@@ -23,6 +23,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.function.Predicate;
import io.github.resilience4j.bulkhead.BulkheadConfig;
@@ -90,6 +91,7 @@ public class ResilienceReifier extends ProcessorReifier<CircuitBreakerDefinition
cbConfig, bhConfig, tlConfig, processor, fallback, throwExceptionWhenHalfOpenOrOpenState, recordPredicate,
ignorePredicate);
answer.setDisabled(isDisabled(camelContext, definition));
+ answer.setAsynchronous(parseBoolean(config.getAsynchronous(), false));
configureTimeoutExecutorService(answer, config);
// using any existing circuit breakers?
if (config.getCircuitBreaker() != null) {
@@ -224,6 +226,8 @@ public class ResilienceReifier extends ProcessorReifier<CircuitBreakerDefinition
return;
}
+ boolean isAsync = parseBoolean(config.getAsynchronous(), false);
+
ExecutorService executorService;
boolean shutdownThreadPool = false;
@@ -234,8 +238,15 @@ public class ResilienceReifier extends ProcessorReifier<CircuitBreakerDefinition
executorService = lookupExecutorServiceRef("CircuitBreaker", definition, ref);
shutdownThreadPool = true;
}
+ if (isAsync && !(executorService instanceof ScheduledExecutorService)) {
+ throw new IllegalArgumentException(
+ "timeoutExecutorService must be a ScheduledExecutorService when asynchronous mode is enabled");
+ }
+ } else if (isAsync) {
+ executorService = camelContext.getExecutorServiceManager()
+ .newDefaultScheduledThreadPool(this, "CircuitBreaker");
+ shutdownThreadPool = true;
} else {
- // A default thread pool if none is provided.
executorService = camelContext.getExecutorServiceManager()
.newThreadPool(this, "CircuitBreaker",
camelContext.getExecutorServiceManager().getDefaultThreadPoolProfile());
@@ -243,6 +254,9 @@ public class ResilienceReifier extends ProcessorReifier<CircuitBreakerDefinition
}
processor.setExecutorService(executorService);
+ if (isAsync && executorService instanceof ScheduledExecutorService ses) {
+ processor.setScheduledExecutorService(ses);
+ }
processor.setShutdownExecutorService(shutdownThreadPool);
}
diff --git a/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceAsyncBulkheadTest.java b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceAsyncBulkheadTest.java
new file mode 100644
index 000000000000..644311f272e1
--- /dev/null
+++ b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceAsyncBulkheadTest.java
@@ -0,0 +1,70 @@
+/*
+ * 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 org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.CircuitBreakerConstants;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+public class ResilienceAsyncBulkheadTest extends CamelTestSupport {
+
+ @Test
+ public void testResilience() throws Exception {
+ test("direct:start");
+ }
+
+ @Test
+ public void testResilienceWithTimeOut() throws Exception {
+ test("direct:start.with.timeout.enabled");
+ }
+
+ private void test(String endPointUri) throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceived("Bye World");
+ getMockEndpoint("mock:result").expectedPropertyReceived(CircuitBreakerConstants.RESPONSE_SUCCESSFUL_EXECUTION, true);
+ getMockEndpoint("mock:result").expectedPropertyReceived(CircuitBreakerConstants.RESPONSE_FROM_FALLBACK, false);
+
+ template.sendBody(endPointUri, "Hello World");
+
+ MockEndpoint.assertIsSatisfied(context);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start").circuitBreaker()
+ .resilience4jConfiguration().asynchronous(true).bulkheadEnabled(true).end()
+ .to("direct:foo").to("log:foo")
+ .onFallback().transform().constant("Fallback message").end()
+ .to("log:result").to("mock:result");
+
+ from("direct:start.with.timeout.enabled").circuitBreaker()
+ .resilience4jConfiguration().asynchronous(true).bulkheadEnabled(true)
+ .timeoutEnabled(true).timeoutDuration(2000).end()
+ .to("direct:foo").to("log:foo")
+ .onFallback().transform().constant("Fallback message").end()
+ .to("log:result").to("mock:result");
+
+ from("direct:foo").transform().constant("Bye World");
+ }
+ };
+ }
+
+}
diff --git a/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceAsyncInvalidExecutorTest.java b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceAsyncInvalidExecutorTest.java
new file mode 100644
index 000000000000..90131512c191
--- /dev/null
+++ b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceAsyncInvalidExecutorTest.java
@@ -0,0 +1,86 @@
+/*
+ * 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.ExecutorService;
+import java.util.concurrent.Executors;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.FailedToCreateRouteException;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.apache.camel.test.junit6.TestSupport.assertIsInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Test that using a non-ScheduledExecutorService with async + timeout throws at startup.
+ */
+public class ResilienceAsyncInvalidExecutorTest extends CamelTestSupport {
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext ctx = super.createCamelContext();
+ ExecutorService plainExecutor = Executors.newFixedThreadPool(2);
+ ctx.getRegistry().bind("myExecutor", plainExecutor);
+ return ctx;
+ }
+
+ @Test
+ public void testInvalidExecutorThrows() {
+ Exception exception = assertThrows(Exception.class, () -> {
+ context.addRoutes(createTestRouteBuilder());
+ context.start();
+ });
+ assertIsInstanceOf(FailedToCreateRouteException.class, exception);
+ assertIsInstanceOf(IllegalArgumentException.class, exception.getCause());
+ assertTrue(exception.getCause().getMessage().contains("ScheduledExecutorService"),
+ "Should mention ScheduledExecutorService");
+ assertTrue(exception.getCause().getMessage().contains("asynchronous"),
+ "Should mention asynchronous");
+ }
+
+ private RoutesBuilder createTestRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .circuitBreaker()
+ .resilience4jConfiguration()
+ .asynchronous(true)
+ .timeoutEnabled(true)
+ .timeoutDuration(2000)
+ .timeoutExecutorService("myExecutor")
+ .end()
+ .to("direct:foo")
+ .end()
+ .to("mock:result");
+
+ from("direct:foo").transform().constant("Bye World");
+ }
+ };
+ }
+
+}
diff --git a/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceAsyncRouteOkTest.java b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceAsyncRouteOkTest.java
new file mode 100644
index 000000000000..c2c6a3fb06ef
--- /dev/null
+++ b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceAsyncRouteOkTest.java
@@ -0,0 +1,70 @@
+/*
+ * 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 org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.CircuitBreakerConstants;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+public class ResilienceAsyncRouteOkTest extends CamelTestSupport {
+
+ @Test
+ public void testResilience() throws Exception {
+ test("direct:start");
+ }
+
+ @Test
+ public void testResilienceWithTimeOut() throws Exception {
+ test("direct:start.with.timeout.enabled");
+ }
+
+ private void test(String endPointUri) throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceived("Bye World");
+ getMockEndpoint("mock:result").expectedPropertyReceived(CircuitBreakerConstants.RESPONSE_SUCCESSFUL_EXECUTION, true);
+ getMockEndpoint("mock:result").expectedPropertyReceived(CircuitBreakerConstants.RESPONSE_FROM_FALLBACK, false);
+ getMockEndpoint("mock:result").expectedPropertyReceived(CircuitBreakerConstants.RESPONSE_STATE, "CLOSED");
+
+ template.sendBody(endPointUri, "Hello World");
+
+ MockEndpoint.assertIsSatisfied(context);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start").circuitBreaker()
+ .resilience4jConfiguration().asynchronous(true).end()
+ .to("direct:foo").to("log:foo")
+ .onFallback().transform().constant("Fallback message").end()
+ .to("log:result").to("mock:result");
+
+ from("direct:start.with.timeout.enabled").circuitBreaker()
+ .resilience4jConfiguration().asynchronous(true).timeoutEnabled(true).timeoutDuration(2000).end()
+ .to("direct:foo").to("log:foo")
+ .onFallback().transform().constant("Fallback message").end()
+ .to("log:result").to("mock:result");
+
+ from("direct:foo").transform().constant("Bye World");
+ }
+ };
+ }
+
+}
diff --git a/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceAsyncTimeoutTest.java b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceAsyncTimeoutTest.java
new file mode 100644
index 000000000000..f88b9e5071aa
--- /dev/null
+++ b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceAsyncTimeoutTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.TimeoutException;
+
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.apache.camel.test.junit6.TestSupport.assertIsInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/**
+ * Resilience using async mode with timeout
+ */
+public class ResilienceAsyncTimeoutTest extends CamelTestSupport {
+
+ @Test
+ public void testFast() {
+ Object out = template.requestBody("direct:start", "fast");
+ assertEquals("Fast response", out);
+ }
+
+ @Test
+ public void testSlow() {
+ Exception exception = assertThrows(Exception.class,
+ () -> template.requestBody("direct:start", "slow"),
+ "Should fail due to timeout");
+ assertIsInstanceOf(TimeoutException.class, exception.getCause());
+ }
+
+ @Test
+ public void testSlowLoop() {
+ for (int i = 0; i < 10; i++) {
+ Exception exception = assertThrows(Exception.class,
+ () -> template.requestBody("direct:start", "slow"),
+ "Should fail due to timeout");
+ assertIsInstanceOf(TimeoutException.class, exception.getCause());
+ }
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start").circuitBreaker()
+ .resilience4jConfiguration().asynchronous(true).timeoutEnabled(true).timeoutDuration(2000).end()
+ .log("Resilience processing start: ${threadName}").toD("direct:${body}")
+ .log("Resilience processing end: ${threadName}").end().log("After Resilience ${body}");
+
+ from("direct:fast")
+ .log("Fast processing start: ${threadName}").delay(1000).transform().constant("Fast response")
+ .log("Fast processing end: ${threadName}");
+
+ from("direct:slow")
+ .log("Slow processing start: ${threadName}").delay(3000).transform().constant("Slow response")
+ .log("Slow processing end: ${threadName}");
+ }
+ };
+ }
+
+}
diff --git a/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceAsyncTimeoutWithFallbackTest.java b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceAsyncTimeoutWithFallbackTest.java
new file mode 100644
index 000000000000..0cec837631ea
--- /dev/null
+++ b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceAsyncTimeoutWithFallbackTest.java
@@ -0,0 +1,68 @@
+/*
+ * 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 org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Resilience using async mode with timeout and fallback
+ */
+public class ResilienceAsyncTimeoutWithFallbackTest extends CamelTestSupport {
+
+ @Test
+ public void testFast() {
+ Object out = template.requestBody("direct:start", "fast");
+ assertEquals("LAST CHANGE", out);
+ }
+
+ @Test
+ public void testSlow() {
+ Object out = template.requestBody("direct:start", "slow");
+ assertEquals("LAST CHANGE", out);
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start").circuitBreaker()
+ .resilience4jConfiguration().asynchronous(true).timeoutEnabled(true).timeoutDuration(2000).end()
+ .log("Resilience processing start: ${threadName}").toD("direct:${body}")
+ .log("Resilience processing end: ${threadName}").onFallback()
+ .log("Resilience fallback start: ${threadName}").transform().constant("Fallback response")
+ .log("Resilience fallback end: ${threadName}").end()
+ .log("After Resilience ${body}").transform(simple("A CHANGE")).transform(simple("LAST CHANGE"))
+ .log("End ${body}");
+
+ from("direct:fast")
+ .log("Fast processing start: ${threadName}").delay(1000).transform().constant("Fast response")
+ .log("Fast processing end: ${threadName}");
+
+ from("direct:slow")
+ .log("Slow processing start: ${threadName}").delay(3000).transform().constant("Slow response")
+ .log("Slow processing end: ${threadName}");
+ }
+ };
+ }
+
+}
diff --git a/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceAsyncTimeoutWriteBackRaceTest.java b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceAsyncTimeoutWriteBackRaceTest.java
new file mode 100644
index 000000000000..628fef240c5a
--- /dev/null
+++ b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceAsyncTimeoutWriteBackRaceTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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 in async mode.
+ */
+public class ResilienceAsyncTimeoutWriteBackRaceTest 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);
+
+ assertTrue(workerDone.await(10, TimeUnit.SECONDS), "Worker thread should complete");
+
+ 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()
+ .asynchronous(true)
+ .timeoutEnabled(true)
+ .timeoutDuration(500)
+ .end()
+ .process(exchange -> {
+ try {
+ Thread.sleep(3000);
+ exchange.getIn().setBody("Worker result");
+ } finally {
+ workerDone.countDown();
+ }
+ })
+ .onFallback()
+ .setBody(constant("Fallback result"))
+ .end()
+ .to("mock:result");
+ }
+ };
+ }
+}
diff --git a/core/camel-core-engine/src/main/docs/modules/eips/pages/resilience4j-eip.adoc b/core/camel-core-engine/src/main/docs/modules/eips/pages/resilience4j-eip.adoc
index aa84fc1eb53c..f7e90244e5ea 100644
--- a/core/camel-core-engine/src/main/docs/modules/eips/pages/resilience4j-eip.adoc
+++ b/core/camel-core-engine/src/main/docs/modules/eips/pages/resilience4j-eip.adoc
@@ -154,6 +154,87 @@ YAML::
In this example if calling the downstream service does not return a response within 2 seconds,
a timeout is triggered, and the exchange will fail with a `TimeoutException`.
+=== Asynchronous (non-blocking) processing
+
+By default, the Resilience4j circuit breaker processes exchanges synchronously,
+blocking the caller thread until the protected operation completes.
+When `asynchronous` is enabled, the circuit breaker uses Resilience4j's CompletionStage-based
+decorators, which release the caller thread immediately and complete processing asynchronously.
+
+This is most valuable when the downstream processor supports asynchronous processing
+(e.g., Netty HTTP, Kafka), as it avoids blocking threads while waiting for responses.
+
+[tabs]
+====
+Java::
++
+[source,java]
+----
+from("direct:start")
+ .circuitBreaker()
+ .resilience4jConfiguration()
+ .asynchronous(true)
+ .timeoutEnabled(true)
+ .timeoutDuration(2000)
+ .end()
+ .to("http://fooservice.com/faulty")
+ .onFallback()
+ .transform().constant("Fallback message")
+ .end()
+ .to("mock:result");
+----
+
+XML::
++
+[source,xml]
+----
+<route>
+ <from uri="direct:start"/>
+ <circuitBreaker>
+ <resilience4jConfiguration asynchronous="true" timeoutEnabled="true" timeoutDuration="2000"/>
+ <to uri="http://fooservice.com/faulty"/>
+ <onFallback>
+ <transform>
+ <constant>Fallback message</constant>
+ </transform>
+ </onFallback>
+ </circuitBreaker>
+ <to uri="mock:result"/>
+</route>
+----
+
+YAML::
++
+[source,yaml]
+----
+- route:
+ from:
+ uri: direct:start
+ steps:
+ - circuitBreaker:
+ steps:
+ - resilience4jConfiguration:
+ asynchronous: "true"
+ timeoutDuration: 2000
+ timeoutEnabled: "true"
+ - to:
+ uri: http://fooservice.com/faulty
+ - onFallback:
+ steps:
+ - transform:
+ expression:
+ constant:
+ expression: Fallback message
+ - to:
+ uri: mock:result
+----
+====
+
+NOTE: When using asynchronous mode with timeout, the `timeoutExecutorService` must be
+a `ScheduledExecutorService`. If not provided, Camel will automatically create one.
+If a custom executor is provided that is not a `ScheduledExecutorService`, the route will
+fail to start with an `IllegalArgumentException`.
+
=== Camel's Error Handler and Circuit Breaker EIP
By default, the xref:circuitBreaker-eip.adoc[Circuit Breaker] EIP handles errors by itself.
diff --git a/core/camel-core-model/src/generated/java/org/apache/camel/model/Resilience4jConfigurationDefinitionConfigurer.java b/core/camel-core-model/src/generated/java/org/apache/camel/model/Resilience4jConfigurationDefinitionConfigurer.java
index 3504155b89f5..382b2e6347d4 100644
--- a/core/camel-core-model/src/generated/java/org/apache/camel/model/Resilience4jConfigurationDefinitionConfigurer.java
+++ b/core/camel-core-model/src/generated/java/org/apache/camel/model/Resilience4jConfigurationDefinitionConfigurer.java
@@ -22,6 +22,7 @@ public class Resilience4jConfigurationDefinitionConfigurer extends org.apache.ca
private static final Map<String, Object> ALL_OPTIONS;
static {
Map<String, Object> map = new CaseInsensitiveMap();
+ map.put("Asynchronous", java.lang.String.class);
map.put("AutomaticTransitionFromOpenToHalfOpenEnabled", java.lang.String.class);
map.put("BulkheadEnabled", java.lang.String.class);
map.put("BulkheadFairCallHandlingEnabled", java.lang.String.class);
@@ -56,6 +57,7 @@ public class Resilience4jConfigurationDefinitionConfigurer extends org.apache.ca
public boolean configure(CamelContext camelContext, Object obj, String name, Object value, boolean ignoreCase) {
org.apache.camel.model.Resilience4jConfigurationDefinition target = (org.apache.camel.model.Resilience4jConfigurationDefinition) obj;
switch (ignoreCase ? name.toLowerCase() : name) {
+ case "asynchronous": target.setAsynchronous(property(camelContext, java.lang.String.class, value)); return true;
case "automatictransitionfromopentohalfopenenabled":
case "automaticTransitionFromOpenToHalfOpenEnabled": target.setAutomaticTransitionFromOpenToHalfOpenEnabled(property(camelContext, java.lang.String.class, value)); return true;
case "bulkheadenabled":
@@ -120,6 +122,7 @@ public class Resilience4jConfigurationDefinitionConfigurer extends org.apache.ca
@Override
public Class<?> getOptionType(String name, boolean ignoreCase) {
switch (ignoreCase ? name.toLowerCase() : name) {
+ case "asynchronous": return java.lang.String.class;
case "automatictransitionfromopentohalfopenenabled":
case "automaticTransitionFromOpenToHalfOpenEnabled": return java.lang.String.class;
case "bulkheadenabled":
@@ -180,6 +183,7 @@ public class Resilience4jConfigurationDefinitionConfigurer extends org.apache.ca
public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
org.apache.camel.model.Resilience4jConfigurationDefinition target = (org.apache.camel.model.Resilience4jConfigurationDefinition) obj;
switch (ignoreCase ? name.toLowerCase() : name) {
+ case "asynchronous": return target.getAsynchronous();
case "automatictransitionfromopentohalfopenenabled":
case "automaticTransitionFromOpenToHalfOpenEnabled": return target.getAutomaticTransitionFromOpenToHalfOpenEnabled();
case "bulkheadenabled":
diff --git a/core/camel-core-model/src/generated/resources/META-INF/org/apache/camel/model/resilience4jConfiguration.json b/core/camel-core-model/src/generated/resources/META-INF/org/apache/camel/model/resilience4jConfiguration.json
index 9b1452808a39..0ac5b7b290fc 100644
--- a/core/camel-core-model/src/generated/resources/META-INF/org/apache/camel/model/resilience4jConfiguration.json
+++ b/core/camel-core-model/src/generated/resources/META-INF/org/apache/camel/model/resilience4jConfiguration.json
@@ -32,12 +32,13 @@
"bulkheadMaxConcurrentCalls": { "index": 17, "kind": "attribute", "displayName": "Bulkhead Max Concurrent Calls", "group": "common", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 25, "description": "Configures the max amount of concurrent calls the bulkhead will support." },
"bulkheadMaxWaitDuration": { "index": 18, "kind": "attribute", "displayName": "Bulkhead Max Wait Duration", "group": "advanced", "label": "advanced", "required": false, "type": "duration", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "0", "description": "Configures a maximum amount of time which the calling thread will wait to enter the bulkhead. The default is 0 (no waiting)." },
"bulkheadFairCallHandlingEnabled": { "index": 19, "kind": "attribute", "displayName": "Bulkhead Fair Call Handling Enabled", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Configures whether the bulkhead uses a fair calling strategy. When enabled (default), a fair strategy guarantees the order of incoming requests (FIFO). Wh [...]
- "timeoutEnabled": { "index": 20, "kind": "attribute", "displayName": "Timeout Enabled", "group": "common", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether timeout is enabled or not on the circuit breaker." },
- "timeoutExecutorService": { "index": 21, "kind": "attribute", "displayName": "Timeout Executor Service", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "java.util.concurrent.ExecutorService", "deprecated": false, "autowired": false, "secret": false, "description": "References to a custom thread pool to use when timeout is enabled (uses ForkJoinPool.commonPool() by default)." },
- "timeoutDuration": { "index": 22, "kind": "attribute", "displayName": "Timeout Duration", "group": "common", "required": false, "type": "duration", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "1000", "description": "Configures the thread execution timeout. Default value is 1 second." },
- "timeoutCancelRunningFuture": { "index": 23, "kind": "attribute", "displayName": "Timeout Cancel Running Future", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Configures whether cancel is called on the running future. Defaults to true." },
- "micrometerEnabled": { "index": 24, "kind": "attribute", "displayName": "Micrometer Enabled", "group": "common", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether to enable collecting statistics using Micrometer for all circuit breaker instances. This is a global setting (configure via camel.resilience4j.micrometerEnabled=true) and requires adding camel-resili [...]
- "recordException": { "index": 25, "kind": "element", "displayName": "Record Exception", "group": "advanced", "label": "advanced", "required": false, "type": "array", "javaType": "java.util.List<java.lang.String>", "deprecated": false, "autowired": false, "secret": false, "description": "Configure a list of exceptions that are recorded as a failure and thus increase the failure rate. Any exception matching or inheriting from one of the list counts as a failure, unless explicitly ignor [...]
- "ignoreException": { "index": 26, "kind": "element", "displayName": "Ignore Exception", "group": "advanced", "label": "advanced", "required": false, "type": "array", "javaType": "java.util.List<java.lang.String>", "deprecated": false, "autowired": false, "secret": false, "description": "Configure a list of exceptions that are ignored and neither count as a failure nor success. Any exception matching or inheriting from one of the list will not count as a failure nor success, even if t [...]
+ "asynchronous": { "index": 20, "kind": "attribute", "displayName": "Asynchronous", "group": "common", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether to use asynchronous (non-blocking) processing with CompletionStage-based circuit breaker decorators. When enabled, the circuit breaker releases the caller thread immediately and completes processing asynchronou [...]
+ "timeoutEnabled": { "index": 21, "kind": "attribute", "displayName": "Timeout Enabled", "group": "common", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether timeout is enabled or not on the circuit breaker." },
+ "timeoutExecutorService": { "index": 22, "kind": "attribute", "displayName": "Timeout Executor Service", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "java.util.concurrent.ExecutorService", "deprecated": false, "autowired": false, "secret": false, "description": "References to a custom thread pool to use when timeout is enabled (uses ForkJoinPool.commonPool() by default)." },
+ "timeoutDuration": { "index": 23, "kind": "attribute", "displayName": "Timeout Duration", "group": "common", "required": false, "type": "duration", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "1000", "description": "Configures the thread execution timeout. Default value is 1 second." },
+ "timeoutCancelRunningFuture": { "index": 24, "kind": "attribute", "displayName": "Timeout Cancel Running Future", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Configures whether cancel is called on the running future. Defaults to true." },
+ "micrometerEnabled": { "index": 25, "kind": "attribute", "displayName": "Micrometer Enabled", "group": "common", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether to enable collecting statistics using Micrometer for all circuit breaker instances. This is a global setting (configure via camel.resilience4j.micrometerEnabled=true) and requires adding camel-resili [...]
+ "recordException": { "index": 26, "kind": "element", "displayName": "Record Exception", "group": "advanced", "label": "advanced", "required": false, "type": "array", "javaType": "java.util.List<java.lang.String>", "deprecated": false, "autowired": false, "secret": false, "description": "Configure a list of exceptions that are recorded as a failure and thus increase the failure rate. Any exception matching or inheriting from one of the list counts as a failure, unless explicitly ignor [...]
+ "ignoreException": { "index": 27, "kind": "element", "displayName": "Ignore Exception", "group": "advanced", "label": "advanced", "required": false, "type": "array", "javaType": "java.util.List<java.lang.String>", "deprecated": false, "autowired": false, "secret": false, "description": "Configure a list of exceptions that are ignored and neither count as a failure nor success. Any exception matching or inheriting from one of the list will not count as a failure nor success, even if t [...]
}
}
diff --git a/core/camel-core-model/src/main/java/org/apache/camel/model/Resilience4jConfigurationCommon.java b/core/camel-core-model/src/main/java/org/apache/camel/model/Resilience4jConfigurationCommon.java
index 0724a98c8f7b..45a47ed98885 100644
--- a/core/camel-core-model/src/main/java/org/apache/camel/model/Resilience4jConfigurationCommon.java
+++ b/core/camel-core-model/src/main/java/org/apache/camel/model/Resilience4jConfigurationCommon.java
@@ -122,6 +122,13 @@ public class Resilience4jConfigurationCommon extends IdentifiedType {
+ " When disabled, no ordering is guaranteed and may improve throughput.")
private String bulkheadFairCallHandlingEnabled;
@XmlAttribute
+ @Metadata(defaultValue = "false", javaType = "java.lang.Boolean",
+ description = "Whether to use asynchronous (non-blocking) processing with CompletionStage-based circuit breaker decorators."
+ + " When enabled, the circuit breaker releases the caller thread immediately and completes processing asynchronously."
+ + " This is most valuable when the downstream processor supports asynchronous processing (e.g. Netty HTTP, Kafka)."
+ + " When used with timeout, the timeoutExecutorService must be a ScheduledExecutorService.")
+ private String asynchronous;
+ @XmlAttribute
@Metadata(defaultValue = "false", javaType = "java.lang.Boolean",
description = "Whether timeout is enabled or not on the circuit breaker.")
private String timeoutEnabled;
@@ -177,6 +184,7 @@ public class Resilience4jConfigurationCommon extends IdentifiedType {
this.bulkheadMaxConcurrentCalls = source.bulkheadMaxConcurrentCalls;
this.bulkheadMaxWaitDuration = source.bulkheadMaxWaitDuration;
this.bulkheadFairCallHandlingEnabled = source.bulkheadFairCallHandlingEnabled;
+ this.asynchronous = source.asynchronous;
this.timeoutEnabled = source.timeoutEnabled;
this.micrometerEnabled = source.micrometerEnabled;
this.timeoutExecutorService = source.timeoutExecutorService;
@@ -345,6 +353,14 @@ public class Resilience4jConfigurationCommon extends IdentifiedType {
this.bulkheadFairCallHandlingEnabled = bulkheadFairCallHandlingEnabled;
}
+ public String getAsynchronous() {
+ return asynchronous;
+ }
+
+ public void setAsynchronous(String asynchronous) {
+ this.asynchronous = asynchronous;
+ }
+
public String getTimeoutEnabled() {
return timeoutEnabled;
}
diff --git a/core/camel-core-model/src/main/java/org/apache/camel/model/Resilience4jConfigurationDefinition.java b/core/camel-core-model/src/main/java/org/apache/camel/model/Resilience4jConfigurationDefinition.java
index 00e8540d2500..d0941bf1e52d 100644
--- a/core/camel-core-model/src/main/java/org/apache/camel/model/Resilience4jConfigurationDefinition.java
+++ b/core/camel-core-model/src/main/java/org/apache/camel/model/Resilience4jConfigurationDefinition.java
@@ -407,6 +407,22 @@ public class Resilience4jConfigurationDefinition extends Resilience4jConfigurati
return this;
}
+ /**
+ * Whether to use asynchronous (non-blocking) processing. Default is false.
+ */
+ public Resilience4jConfigurationDefinition asynchronous(boolean asynchronous) {
+ setAsynchronous(Boolean.toString(asynchronous));
+ return this;
+ }
+
+ /**
+ * Whether to use asynchronous (non-blocking) processing. Supports property placeholders.
+ */
+ public Resilience4jConfigurationDefinition asynchronous(String asynchronous) {
+ setAsynchronous(asynchronous);
+ return this;
+ }
+
/**
* Whether timeout is enabled or not on the circuit breaker. Default is false.
*/
diff --git a/core/camel-java-io/src/generated/java/org/apache/camel/java/out/JavaDslModelWriter.java b/core/camel-java-io/src/generated/java/org/apache/camel/java/out/JavaDslModelWriter.java
index d9a3e8a1233d..43fe1e447fea 100644
--- a/core/camel-java-io/src/generated/java/org/apache/camel/java/out/JavaDslModelWriter.java
+++ b/core/camel-java-io/src/generated/java/org/apache/camel/java/out/JavaDslModelWriter.java
@@ -2330,6 +2330,7 @@ public class JavaDslModelWriter extends JavaDslModelWriterSupport {
doWriteAttribute(sb, "bulkheadMaxConcurrentCalls", def.getBulkheadMaxConcurrentCalls(), "25");
doWriteAttribute(sb, "bulkheadMaxWaitDuration", def.getBulkheadMaxWaitDuration(), "0");
doWriteAttribute(sb, "bulkheadFairCallHandlingEnabled", def.getBulkheadFairCallHandlingEnabled(), "true");
+ doWriteAttribute(sb, "asynchronous", def.getAsynchronous(), "false");
doWriteAttribute(sb, "timeoutEnabled", def.getTimeoutEnabled(), "false");
doWriteAttribute(sb, "timeoutExecutorService", def.getTimeoutExecutorService(), null);
doWriteAttribute(sb, "timeoutDuration", def.getTimeoutDuration(), "1000");
diff --git a/core/camel-main/src/generated/java/org/apache/camel/main/Resilience4jConfigurationPropertiesConfigurer.java b/core/camel-main/src/generated/java/org/apache/camel/main/Resilience4jConfigurationPropertiesConfigurer.java
index 3061525bb70e..7847b1ef7cf9 100644
--- a/core/camel-main/src/generated/java/org/apache/camel/main/Resilience4jConfigurationPropertiesConfigurer.java
+++ b/core/camel-main/src/generated/java/org/apache/camel/main/Resilience4jConfigurationPropertiesConfigurer.java
@@ -22,6 +22,7 @@ public class Resilience4jConfigurationPropertiesConfigurer extends org.apache.ca
private static final Map<String, Object> ALL_OPTIONS;
static {
Map<String, Object> map = new CaseInsensitiveMap();
+ map.put("Asynchronous", java.lang.Boolean.class);
map.put("AutomaticTransitionFromOpenToHalfOpenEnabled", java.lang.Boolean.class);
map.put("BulkheadEnabled", java.lang.Boolean.class);
map.put("BulkheadFairCallHandlingEnabled", java.lang.Boolean.class);
@@ -53,6 +54,7 @@ public class Resilience4jConfigurationPropertiesConfigurer extends org.apache.ca
public boolean configure(CamelContext camelContext, Object obj, String name, Object value, boolean ignoreCase) {
org.apache.camel.main.Resilience4jConfigurationProperties target = (org.apache.camel.main.Resilience4jConfigurationProperties) obj;
switch (ignoreCase ? name.toLowerCase() : name) {
+ case "asynchronous": target.setAsynchronous(property(camelContext, java.lang.Boolean.class, value)); return true;
case "automatictransitionfromopentohalfopenenabled":
case "automaticTransitionFromOpenToHalfOpenEnabled": target.setAutomaticTransitionFromOpenToHalfOpenEnabled(property(camelContext, java.lang.Boolean.class, value)); return true;
case "bulkheadenabled":
@@ -112,6 +114,7 @@ public class Resilience4jConfigurationPropertiesConfigurer extends org.apache.ca
@Override
public Class<?> getOptionType(String name, boolean ignoreCase) {
switch (ignoreCase ? name.toLowerCase() : name) {
+ case "asynchronous": return java.lang.Boolean.class;
case "automatictransitionfromopentohalfopenenabled":
case "automaticTransitionFromOpenToHalfOpenEnabled": return java.lang.Boolean.class;
case "bulkheadenabled":
@@ -167,6 +170,7 @@ public class Resilience4jConfigurationPropertiesConfigurer extends org.apache.ca
public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
org.apache.camel.main.Resilience4jConfigurationProperties target = (org.apache.camel.main.Resilience4jConfigurationProperties) obj;
switch (ignoreCase ? name.toLowerCase() : name) {
+ case "asynchronous": return target.getAsynchronous();
case "automatictransitionfromopentohalfopenenabled":
case "automaticTransitionFromOpenToHalfOpenEnabled": return target.getAutomaticTransitionFromOpenToHalfOpenEnabled();
case "bulkheadenabled":
diff --git a/core/camel-main/src/generated/resources/META-INF/camel-main-configuration-metadata.json b/core/camel-main/src/generated/resources/META-INF/camel-main-configuration-metadata.json
index 1acde96827c6..ff6f31dbb39e 100644
--- a/core/camel-main/src/generated/resources/META-INF/camel-main-configuration-metadata.json
+++ b/core/camel-main/src/generated/resources/META-INF/camel-main-configuration-metadata.json
@@ -345,6 +345,7 @@
{ "name": "camel.opentelemetry2.includePatterns", "required": false, "description": "Sets include pattern(s) that will explicitly enable tracing for Camel processors that matches the pattern. Multiple patterns can be separated by comma. All processors included by default if nothing is specified.", "sourceType": "org.apache.camel.main.Otel2ConfigurationProperties", "type": "string", "javaType": "java.lang.String", "secret": false },
{ "name": "camel.opentelemetry2.instrumentationName", "required": true, "description": "A name uniquely identifying the instrumentation scope, such as the instrumentation library, package, or fully qualified class name. Must not be null.", "sourceType": "org.apache.camel.main.Otel2ConfigurationProperties", "type": "string", "javaType": "java.lang.String", "defaultValue": "camel", "secret": false },
{ "name": "camel.opentelemetry2.traceProcessors", "required": false, "description": "Setting this to true will create new telemetry spans for each Camel custom Processors. Use the excludePattern property to filter out Processors.", "sourceType": "org.apache.camel.main.Otel2ConfigurationProperties", "type": "boolean", "javaType": "boolean", "defaultValue": false, "secret": false },
+ { "name": "camel.resilience4j.asynchronous", "required": false, "description": "Whether to use asynchronous (non-blocking) processing using Resilience4j's CompletionStage-based decorators. When enabled, the caller thread is released immediately and processing completes asynchronously.", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "boolean", "javaType": "java.lang.Boolean", "defaultValue": false, "secret": false },
{ "name": "camel.resilience4j.automaticTransitionFromOpenToHalfOpenEnabled", "required": false, "description": "Enables automatic transition from OPEN to HALF_OPEN state once the waitDurationInOpenState has passed.", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "boolean", "javaType": "java.lang.Boolean", "defaultValue": false, "secret": false },
{ "name": "camel.resilience4j.bulkheadEnabled", "required": false, "description": "Whether bulkhead is enabled or not on the circuit breaker.", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "boolean", "javaType": "java.lang.Boolean", "defaultValue": false, "secret": false },
{ "name": "camel.resilience4j.bulkheadFairCallHandlingEnabled", "required": false, "description": "Configures whether the bulkhead uses a fair calling strategy. When enabled (default), a fair strategy guarantees the order of incoming requests (FIFO). When disabled, no ordering is guaranteed and may improve throughput.", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "boolean", "javaType": "java.lang.Boolean", "defaultValue": true, "secret": false },
diff --git a/core/camel-main/src/main/docs/main.adoc b/core/camel-main/src/main/docs/main.adoc
index 3cd4bcc174eb..cfd34759d89d 100644
--- a/core/camel-main/src/main/docs/main.adoc
+++ b/core/camel-main/src/main/docs/main.adoc
@@ -678,11 +678,12 @@ The camel.faulttolerance supports 12 options, which are listed below.
=== Resilience4j EIP Circuit Breaker configurations
-The camel.resilience4j supports 24 options, which are listed below.
+The camel.resilience4j supports 25 options, which are listed below.
[width="100%",cols="2,5,^1,2",options="header"]
|===
| Name | Description | Default | Type
+| *camel.resilience4j.asynchronous* | Whether to use asynchronous (non-blocking) processing using Resilience4j's CompletionStage-based decorators. When enabled, the caller thread is released immediately and processing completes asynchronously. | false | Boolean
| *camel.resilience4j.automaticTransitionFromOpenToHalfOpenEnabled* | Enables automatic transition from OPEN to HALF_OPEN state once the waitDurationInOpenState has passed. | false | Boolean
| *camel.resilience4j.bulkheadEnabled* | Whether bulkhead is enabled or not on the circuit breaker. | false | Boolean
| *camel.resilience4j.bulkheadFairCallHandlingEnabled* | Configures whether the bulkhead uses a fair calling strategy. When enabled (default), a fair strategy guarantees the order of incoming requests (FIFO). When disabled, no ordering is guaranteed and may improve throughput. | true | Boolean
diff --git a/core/camel-main/src/main/java/org/apache/camel/main/Resilience4jConfigurationProperties.java b/core/camel-main/src/main/java/org/apache/camel/main/Resilience4jConfigurationProperties.java
index 3c79d34b7f91..5dda9f41a610 100644
--- a/core/camel-main/src/main/java/org/apache/camel/main/Resilience4jConfigurationProperties.java
+++ b/core/camel-main/src/main/java/org/apache/camel/main/Resilience4jConfigurationProperties.java
@@ -71,6 +71,8 @@ public class Resilience4jConfigurationProperties implements BootstrapCloseable {
private String maxWaitDurationInHalfOpenState;
@Metadata(defaultValue = "true")
private Boolean bulkheadFairCallHandlingEnabled;
+ @Metadata(defaultValue = "false")
+ private Boolean asynchronous;
public Resilience4jConfigurationProperties(MainConfigurationProperties parent) {
this.parent = parent;
@@ -401,6 +403,18 @@ public class Resilience4jConfigurationProperties implements BootstrapCloseable {
this.bulkheadFairCallHandlingEnabled = bulkheadFairCallHandlingEnabled;
}
+ public Boolean getAsynchronous() {
+ return asynchronous;
+ }
+
+ /**
+ * Whether to use asynchronous (non-blocking) processing using Resilience4j's CompletionStage-based decorators. When
+ * enabled, the caller thread is released immediately and processing completes asynchronously.
+ */
+ public void setAsynchronous(Boolean asynchronous) {
+ this.asynchronous = asynchronous;
+ }
+
/**
* Refers to an existing io.github.resilience4j.circuitbreaker.CircuitBreaker instance to lookup and use from the
* registry. When using this, then any other circuit breaker options are not in use.
@@ -645,4 +659,13 @@ public class Resilience4jConfigurationProperties implements BootstrapCloseable {
return this;
}
+ /**
+ * Whether to use asynchronous (non-blocking) processing using Resilience4j's CompletionStage-based decorators. When
+ * enabled, the caller thread is released immediately and processing completes asynchronously.
+ */
+ public Resilience4jConfigurationProperties withAsynchronous(Boolean asynchronous) {
+ this.asynchronous = asynchronous;
+ return this;
+ }
+
}
diff --git a/core/camel-xml-io/src/generated/java/org/apache/camel/xml/in/ModelParser.java b/core/camel-xml-io/src/generated/java/org/apache/camel/xml/in/ModelParser.java
index 4bd796100d35..9ff17d4dd39f 100644
--- a/core/camel-xml-io/src/generated/java/org/apache/camel/xml/in/ModelParser.java
+++ b/core/camel-xml-io/src/generated/java/org/apache/camel/xml/in/ModelParser.java
@@ -794,6 +794,7 @@ public class ModelParser extends BaseParser {
}
protected <T extends Resilience4jConfigurationCommon> AttributeHandler<T> resilience4jConfigurationCommonAttributeHandler() {
return (def, key, val) -> switch (key) {
+ case "asynchronous": def.setAsynchronous(val); yield true;
case "automaticTransitionFromOpenToHalfOpenEnabled": def.setAutomaticTransitionFromOpenToHalfOpenEnabled(val); yield true;
case "bulkheadEnabled": def.setBulkheadEnabled(val); yield true;
case "bulkheadFairCallHandlingEnabled": def.setBulkheadFairCallHandlingEnabled(val); yield true;
diff --git a/core/camel-xml-io/src/generated/java/org/apache/camel/xml/out/ModelWriter.java b/core/camel-xml-io/src/generated/java/org/apache/camel/xml/out/ModelWriter.java
index f60eb2643f49..860b5815c8aa 100644
--- a/core/camel-xml-io/src/generated/java/org/apache/camel/xml/out/ModelWriter.java
+++ b/core/camel-xml-io/src/generated/java/org/apache/camel/xml/out/ModelWriter.java
@@ -1524,6 +1524,7 @@ public class ModelWriter extends BaseWriter {
doWriteAttribute("bulkheadMaxConcurrentCalls", def.getBulkheadMaxConcurrentCalls(), "25");
doWriteAttribute("bulkheadMaxWaitDuration", def.getBulkheadMaxWaitDuration(), "0");
doWriteAttribute("bulkheadFairCallHandlingEnabled", def.getBulkheadFairCallHandlingEnabled(), "true");
+ doWriteAttribute("asynchronous", def.getAsynchronous(), "false");
doWriteAttribute("timeoutEnabled", def.getTimeoutEnabled(), "false");
doWriteAttribute("timeoutExecutorService", def.getTimeoutExecutorService(), null);
doWriteAttribute("timeoutDuration", def.getTimeoutDuration(), "1000");
diff --git a/core/camel-yaml-io/src/generated/java/org/apache/camel/yaml/out/YamlModelWriter.java b/core/camel-yaml-io/src/generated/java/org/apache/camel/yaml/out/YamlModelWriter.java
index dd039a8c9a70..d39107a559f6 100644
--- a/core/camel-yaml-io/src/generated/java/org/apache/camel/yaml/out/YamlModelWriter.java
+++ b/core/camel-yaml-io/src/generated/java/org/apache/camel/yaml/out/YamlModelWriter.java
@@ -1519,6 +1519,7 @@ public class YamlModelWriter extends YamlModelWriterSupport {
doWriteAttribute(jo, "bulkheadMaxConcurrentCalls", def.getBulkheadMaxConcurrentCalls(), "25");
doWriteAttribute(jo, "bulkheadMaxWaitDuration", def.getBulkheadMaxWaitDuration(), "0");
doWriteAttribute(jo, "bulkheadFairCallHandlingEnabled", def.getBulkheadFairCallHandlingEnabled(), "true");
+ doWriteAttribute(jo, "asynchronous", def.getAsynchronous(), "false");
doWriteAttribute(jo, "timeoutEnabled", def.getTimeoutEnabled(), "false");
doWriteAttribute(jo, "timeoutExecutorService", def.getTimeoutExecutorService(), null);
doWriteAttribute(jo, "timeoutDuration", def.getTimeoutDuration(), "1000");
diff --git a/dsl/camel-yaml-dsl/camel-yaml-dsl-deserializers/src/generated/java/org/apache/camel/dsl/yaml/deserializers/ModelDeserializers.java b/dsl/camel-yaml-dsl/camel-yaml-dsl-deserializers/src/generated/java/org/apache/camel/dsl/yaml/deserializers/ModelDeserializers.java
index 41b495835e32..3b56f259aec9 100644
--- a/dsl/camel-yaml-dsl/camel-yaml-dsl-deserializers/src/generated/java/org/apache/camel/dsl/yaml/deserializers/ModelDeserializers.java
+++ b/dsl/camel-yaml-dsl/camel-yaml-dsl-deserializers/src/generated/java/org/apache/camel/dsl/yaml/deserializers/ModelDeserializers.java
@@ -13884,6 +13884,7 @@ public final class ModelDeserializers extends YamlDeserializerSupport {
description = "Configures Resilience4j settings for the Circuit Breaker EIP, such as failure rate threshold, wait duration, and sliding window parameters",
deprecated = false,
properties = {
+ @YamlProperty(name = "asynchronous", type = "boolean", defaultValue = "false", description = "Whether to use asynchronous (non-blocking) processing with CompletionStage-based circuit breaker decorators. When enabled, the circuit breaker releases the caller thread immediately and completes processing asynchronously. This is most valuable when the downstream processor supports asynchronous processing (e.g. Netty HTTP, Kafka). When used with timeout, the timeoutExecutorS [...]
@YamlProperty(name = "automaticTransitionFromOpenToHalfOpenEnabled", type = "boolean", defaultValue = "false", description = "Enables automatic transition from OPEN to HALF_OPEN state once the waitDurationInOpenState has passed.", displayName = "Automatic Transition From Open To Half Open Enabled"),
@YamlProperty(name = "bulkheadEnabled", type = "boolean", defaultValue = "false", description = "Whether bulkhead is enabled or not on the circuit breaker.", displayName = "Bulkhead Enabled"),
@YamlProperty(name = "bulkheadFairCallHandlingEnabled", type = "boolean", defaultValue = "true", description = "Configures whether the bulkhead uses a fair calling strategy. When enabled (default), a fair strategy guarantees the order of incoming requests (FIFO). When disabled, no ordering is guaranteed and may improve throughput.", displayName = "Bulkhead Fair Call Handling Enabled"),
@@ -13928,6 +13929,11 @@ public final class ModelDeserializers extends YamlDeserializerSupport {
String propertyKey, String propertyName, Node node) {
propertyKey = org.apache.camel.util.StringHelper.dashToCamelCase(propertyKey);
switch(propertyKey) {
+ case "asynchronous": {
+ String val = asText(node);
+ target.setAsynchronous(val);
+ break;
+ }
case "automaticTransitionFromOpenToHalfOpenEnabled": {
String val = asText(node);
target.setAutomaticTransitionFromOpenToHalfOpenEnabled(val);
diff --git a/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl-canonical.json b/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl-canonical.json
index 8d2f166078e8..66dd259d9786 100644
--- a/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl-canonical.json
+++ b/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl-canonical.json
@@ -3608,6 +3608,12 @@
"type" : "object",
"additionalProperties" : false,
"properties" : {
+ "asynchronous" : {
+ "type" : "boolean",
+ "title" : "Asynchronous",
+ "description" : "Whether to use asynchronous (non-blocking) processing with CompletionStage-based circuit breaker decorators. When enabled, the circuit breaker releases the caller thread immediately and completes processing asynchronously. This is most valuable when the downstream processor supports asynchronous processing (e.g. Netty HTTP, Kafka). When used with timeout, the timeoutExecutorService must be a ScheduledExecutorService.",
+ "default" : false
+ },
"automaticTransitionFromOpenToHalfOpenEnabled" : {
"type" : "boolean",
"title" : "Automatic Transition From Open To Half Open Enabled",
diff --git a/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl.json b/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl.json
index 61a26525fb4f..db094f3b49cc 100644
--- a/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl.json
+++ b/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl.json
@@ -5497,6 +5497,12 @@
"type" : "object",
"additionalProperties" : false,
"properties" : {
+ "asynchronous" : {
+ "type" : "boolean",
+ "title" : "Asynchronous",
+ "description" : "Whether to use asynchronous (non-blocking) processing with CompletionStage-based circuit breaker decorators. When enabled, the circuit breaker releases the caller thread immediately and completes processing asynchronously. This is most valuable when the downstream processor supports asynchronous processing (e.g. Netty HTTP, Kafka). When used with timeout, the timeoutExecutorService must be a ScheduledExecutorService.",
+ "default" : false
+ },
"automaticTransitionFromOpenToHalfOpenEnabled" : {
"type" : "boolean",
"title" : "Automatic Transition From Open To Half Open Enabled",