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 <[email protected]>
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 <[email protected]>
---
.../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",
