This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch feature/CAMEL-24209-async-circuit-breaker in repository https://gitbox.apache.org/repos/asf/camel.git
commit 031b3154c29be30713590fbcbee6e6688348d410 Author: Claus Ibsen <[email protected]> AuthorDate: Tue Jul 21 12:14:44 2026 +0200 CAMEL-24209: camel-resilience4j - Add asynchronous (non-blocking) processing option Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../main/camel-main-configuration-metadata.json | 11 +- .../catalog/models/resilience4jConfiguration.json | 15 +- .../apache/camel/catalog/schemas/camel-spring.xsd | 51 +------ .../apache/camel/catalog/schemas/camel-xml-io.xsd | 51 +------ .../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 ++ .../dsl/yaml/deserializers/ModelDeserializers.java | 6 + .../resources/schema/camelYamlDsl-canonical.json | 6 + .../generated/resources/schema/camelYamlDsl.json | 6 + 20 files changed, 809 insertions(+), 108 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..144a7868822f 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 @@ -347,27 +347,24 @@ { "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.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 }, { "name": "camel.resilience4j.bulkheadMaxConcurrentCalls", "required": false, "description": "Configures the max amount of concurrent calls the bulkhead will support.", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "integer", "javaType": "java.lang.Integer", "secret": false }, - { "name": "camel.resilience4j.bulkheadMaxWaitDuration", "required": false, "description": "Configures a maximum amount of time which the calling thread will wait to enter the bulkhead. Accepts Camel duration expressions (e.g. 500ms, 1s) or millisecond values. Default is 0 (no waiting).", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "string", "javaType": "java.lang.String", "secret": false }, + { "name": "camel.resilience4j.bulkheadMaxWaitDuration", "required": false, "description": "Configures a maximum amount of time which the calling thread will wait to enter the bulkhead. If bulkhead has space available, entry is guaranteed and immediate. If bulkhead is full, calling threads will contest for space, if it becomes available. maxWaitDuration can be set to 0. Note: for threads running on an event-loop or equivalent (rx computation pool, etc), setting maxWaitDuration to 0 is [...] { "name": "camel.resilience4j.circuitBreaker", "required": false, "description": "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.", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "string", "javaType": "java.lang.String", "secret": false }, { "name": "camel.resilience4j.config", "required": false, "description": "Refers to an existing io.github.resilience4j.circuitbreaker.CircuitBreakerConfig instance to lookup and use from the registry.", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "string", "javaType": "java.lang.String", "secret": false }, { "name": "camel.resilience4j.failureRateThreshold", "required": false, "description": "Configures the failure rate threshold in percentage. If the failure rate is equal or greater than the threshold the CircuitBreaker transitions to open and starts short-circuiting calls. The threshold must be greater than 0 and not greater than 100. Default value is 50 percentage.", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "number", "javaType": "java.lang.F [...] - { "name": "camel.resilience4j.maxWaitDurationInHalfOpenState", "required": false, "description": "Configures the maximum wait duration which controls how long the CircuitBreaker should stay in Half Open state, before it switches to open. Value 0 means circuit breaker will wait in half open state until all permitted calls have been completed. Accepts Camel duration expressions (e.g. 10s, 1m) or millisecond values.", "sourceType": "org.apache.camel.main.Resilience4jConfigurationPropert [...] { "name": "camel.resilience4j.micrometerEnabled", "required": false, "description": "Whether to enable collecting statistics using Micrometer. This requires adding camel-resilience4j-micrometer JAR to the classpath.", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "boolean", "javaType": "java.lang.Boolean", "defaultValue": false, "secret": false }, { "name": "camel.resilience4j.minimumNumberOfCalls", "required": false, "description": "Configures configures the minimum number of calls which are required (per sliding window period) before the CircuitBreaker can calculate the error rate. For example, if minimumNumberOfCalls is 10, then at least 10 calls must be recorded, before the failure rate can be calculated. If only 9 calls have been recorded the CircuitBreaker will not transition to open even if all 9 calls have failed. Defa [...] { "name": "camel.resilience4j.permittedNumberOfCallsInHalfOpenState", "required": false, "description": "Configures the number of permitted calls when the CircuitBreaker is half open. The size must be greater than 0. Default size is 10.", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "integer", "javaType": "java.lang.Integer", "defaultValue": 10, "secret": false }, { "name": "camel.resilience4j.slidingWindowSize", "required": false, "description": "Configures the size of the sliding window which is used to record the outcome of calls when the CircuitBreaker is closed. slidingWindowSize configures the size of the sliding window. Sliding window can either be count-based or time-based. If slidingWindowType is COUNT_BASED, the last slidingWindowSize calls are recorded and aggregated. If slidingWindowType is TIME_BASED, the calls of the last sliding [...] - { "name": "camel.resilience4j.slidingWindowSynchronizationStrategy", "required": false, "description": "Configures the synchronization strategy for the sliding window. LOCK_FREE uses a CAS-based lock-free algorithm for better performance under high concurrency. SYNCHRONIZED uses blocking locks with lower memory allocation.", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "enum", "javaType": "java.lang.String", "defaultValue": "SYNCHRONIZED", "secre [...] { "name": "camel.resilience4j.slidingWindowType", "required": false, "description": "Configures the type of the sliding window which is used to record the outcome of calls when the CircuitBreaker is closed. Sliding window can either be count-based or time-based. If slidingWindowType is COUNT_BASED, the last slidingWindowSize calls are recorded and aggregated. If slidingWindowType is TIME_BASED, the calls of the last slidingWindowSize seconds are recorded and aggregated. Default slidi [...] - { "name": "camel.resilience4j.slowCallDurationThreshold", "required": false, "description": "Configures the duration threshold above which calls are considered as slow and increase the slow calls percentage. Accepts Camel duration expressions (e.g. 60s, 1m) or millisecond values. Default value is 60 seconds.", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "string", "javaType": "java.lang.String", "defaultValue": "60s", "secret": false }, + { "name": "camel.resilience4j.slowCallDurationThreshold", "required": false, "description": "Configures the duration threshold (seconds) above which calls are considered as slow and increase the slow calls percentage. Default value is 60 seconds.", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "integer", "javaType": "java.lang.Integer", "defaultValue": 60, "secret": false }, { "name": "camel.resilience4j.slowCallRateThreshold", "required": false, "description": "Configures a threshold in percentage. The CircuitBreaker considers a call as slow when the call duration is greater than slowCallDurationThreshold(Duration. When the percentage of slow calls is equal or greater the threshold, the CircuitBreaker transitions to open and starts short-circuiting calls. The threshold must be greater than 0 and not greater than 100. Default value is 100 percentage whic [...] { "name": "camel.resilience4j.throwExceptionWhenHalfOpenOrOpenState", "required": false, "description": "Whether to throw io.github.resilience4j.circuitbreaker.CallNotPermittedException when the call is rejected due circuit breaker is half open or open.", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "boolean", "javaType": "java.lang.Boolean", "defaultValue": false, "secret": false }, { "name": "camel.resilience4j.timeoutCancelRunningFuture", "required": false, "description": "Configures whether cancel is called on the running future. Defaults to true.", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "boolean", "javaType": "java.lang.Boolean", "defaultValue": true, "secret": false }, - { "name": "camel.resilience4j.timeoutDuration", "required": false, "description": "Configures the thread execution timeout. Accepts Camel duration expressions (e.g. 1s, 500ms) or millisecond values. Default value is 1000 millis (1 second).", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "string", "javaType": "java.lang.String", "defaultValue": "1000", "secret": false }, + { "name": "camel.resilience4j.timeoutDuration", "required": false, "description": "Configures the thread execution timeout (millis). Default value is 1000 millis (1 second).", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "integer", "javaType": "java.lang.Integer", "defaultValue": 1000, "secret": false }, { "name": "camel.resilience4j.timeoutEnabled", "required": false, "description": "Whether timeout is enabled or not on the circuit breaker. Default is false.", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "boolean", "javaType": "java.lang.Boolean", "defaultValue": false, "secret": false }, { "name": "camel.resilience4j.timeoutExecutorService", "required": false, "description": "References to a custom thread pool to use when timeout is enabled (uses ForkJoinPool#commonPool() by default)", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "string", "javaType": "java.lang.String", "secret": false }, - { "name": "camel.resilience4j.waitDurationInOpenState", "required": false, "description": "Configures the wait duration which specifies how long the CircuitBreaker should stay open, before it switches to half open. Accepts Camel duration expressions (e.g. 60s, 1m) or millisecond values. Default value is 60 seconds.", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "string", "javaType": "java.lang.String", "defaultValue": "60s", "secret": false }, + { "name": "camel.resilience4j.waitDurationInOpenState", "required": false, "description": "Configures the wait duration (in seconds) which specifies how long the CircuitBreaker should stay open, before it switches to half open. Default value is 60 seconds.", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "integer", "javaType": "java.lang.Integer", "defaultValue": 60, "secret": false }, { "name": "camel.resilience4j.writableStackTraceEnabled", "required": false, "description": "Enables writable stack traces. When set to false, Exception.getStackTrace returns a zero length array. This may be used to reduce log spam when the circuit breaker is open as the cause of the exceptions is already known (the circuit breaker is short-circuiting calls).", "sourceType": "org.apache.camel.main.Resilience4jConfigurationProperties", "type": "boolean", "javaType": "java.lang.Boolean [...] { "name": "camel.rest.apiComponent", "required": false, "description": "Sets the name of the Camel component to use as the REST API (such as swagger or openapi)", "sourceType": "org.apache.camel.spi.RestConfiguration", "type": "string", "javaType": "java.lang.String", "secret": false }, { "name": "camel.rest.apiContextPath", "required": false, "description": "Sets a leading API context-path the REST API services will be using. This can be used when using components such as camel-servlet where the deployed web application is deployed using a context-path.", "sourceType": "org.apache.camel.spi.RestConfiguration", "type": "string", "javaType": "java.lang.String", "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..a1b529fd3603 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 @@ -4450,17 +4450,6 @@ closed. Sliding window can either be count-based or time-based. Default value: 1 <![CDATA[ Configures the type of the sliding window which is used to record the outcome of calls when the CircuitBreaker is closed. Sliding window can either be count-based or time-based. Default value: COUNT_BASED -]]> - </xs:documentation> - </xs:annotation> - </xs:attribute> - <xs:attribute name="slidingWindowSynchronizationStrategy" type="xs:string"> - <xs:annotation> - <xs:documentation xml:lang="en"> -<![CDATA[ -Configures the synchronization strategy for the sliding window. LOCK_FREE uses a CAS-based lock-free algorithm for -better performance under high concurrency. SYNCHRONIZED uses blocking locks with lower memory allocation. Default value: -SYNCHRONIZED ]]> </xs:documentation> </xs:annotation> @@ -4489,8 +4478,8 @@ to reduce log spam when the circuit breaker is open. Default value: true <xs:annotation> <xs:documentation xml:lang="en"> <![CDATA[ -Configures the wait duration which specifies how long the CircuitBreaker should stay open, before it switches to half -open. The default is 60 seconds. Default value: 60000 +Configures the wait duration (in seconds) which specifies how long the CircuitBreaker should stay open, before it +switches to half open. Default value: 60 ]]> </xs:documentation> </xs:annotation> @@ -4501,17 +4490,6 @@ open. The default is 60 seconds. Default value: 60000 <![CDATA[ Enables automatic transition from OPEN to HALF_OPEN state once the waitDurationInOpenState has passed. Default value: false -]]> - </xs:documentation> - </xs:annotation> - </xs:attribute> - <xs:attribute name="maxWaitDurationInHalfOpenState" type="xs:string"> - <xs:annotation> - <xs:documentation xml:lang="en"> -<![CDATA[ -Configures the maximum wait duration which controls how long the CircuitBreaker should stay in Half Open state, before -it switches to open. Value 0 means circuit breaker will wait in half open state until all permitted calls have been -completed. Default value: 0 ]]> </xs:documentation> </xs:annotation> @@ -4531,8 +4509,8 @@ transitions to open and starts short-circuiting calls. Default value: 100 <xs:annotation> <xs:documentation xml:lang="en"> <![CDATA[ -Configures the duration threshold above which calls are considered as slow and increase the slow calls percentage. The -default is 60 seconds. Default value: 60000 +Configures the duration threshold (seconds) above which calls are considered as slow and increase the slow calls +percentage. Default value: 60 ]]> </xs:documentation> </xs:annotation> @@ -4559,19 +4537,7 @@ Configures the max amount of concurrent calls the bulkhead will support. Default <xs:annotation> <xs:documentation xml:lang="en"> <![CDATA[ -Configures a maximum amount of time which the calling thread will wait to enter the bulkhead. The default is 0 (no -waiting). Default value: 0 -]]> - </xs:documentation> - </xs:annotation> - </xs:attribute> - <xs:attribute name="bulkheadFairCallHandlingEnabled" type="xs:string"> - <xs:annotation> - <xs:documentation xml:lang="en"> -<![CDATA[ -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 +Configures a maximum amount of time which the calling thread will wait to enter the bulkhead. Default value: 0 ]]> </xs:documentation> </xs:annotation> @@ -4616,8 +4582,7 @@ Configures whether cancel is called on the running future. Defaults to true. Def <xs:annotation> <xs:documentation xml:lang="en"> <![CDATA[ -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-resilience4j-micrometer JAR to the +Whether to enable collecting statistics using Micrometer. This requires adding camel-resilience4j-micrometer JAR to the classpath. Default value: false ]]> </xs:documentation> @@ -4832,8 +4797,8 @@ References a custom thread pool to use when offloading a guarded action to anoth <xs:annotation> <xs:documentation xml:lang="en"> <![CDATA[ -Deprecated: Whether the fallback goes over the network. If so, the fallback is executed on a separate thread-pool to -avoid exhausting the main thread-pool. Default value: false +Whether the fallback goes over the network. If so, the fallback is executed on a separate thread-pool to avoid +exhausting the main thread-pool. 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..f4e0045b8144 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 @@ -3560,17 +3560,6 @@ closed. Sliding window can either be count-based or time-based. Default value: 1 <![CDATA[ Configures the type of the sliding window which is used to record the outcome of calls when the CircuitBreaker is closed. Sliding window can either be count-based or time-based. Default value: COUNT_BASED -]]> - </xs:documentation> - </xs:annotation> - </xs:attribute> - <xs:attribute name="slidingWindowSynchronizationStrategy" type="xs:string"> - <xs:annotation> - <xs:documentation xml:lang="en"> -<![CDATA[ -Configures the synchronization strategy for the sliding window. LOCK_FREE uses a CAS-based lock-free algorithm for -better performance under high concurrency. SYNCHRONIZED uses blocking locks with lower memory allocation. Default value: -SYNCHRONIZED ]]> </xs:documentation> </xs:annotation> @@ -3599,8 +3588,8 @@ to reduce log spam when the circuit breaker is open. Default value: true <xs:annotation> <xs:documentation xml:lang="en"> <![CDATA[ -Configures the wait duration which specifies how long the CircuitBreaker should stay open, before it switches to half -open. The default is 60 seconds. Default value: 60000 +Configures the wait duration (in seconds) which specifies how long the CircuitBreaker should stay open, before it +switches to half open. Default value: 60 ]]> </xs:documentation> </xs:annotation> @@ -3611,17 +3600,6 @@ open. The default is 60 seconds. Default value: 60000 <![CDATA[ Enables automatic transition from OPEN to HALF_OPEN state once the waitDurationInOpenState has passed. Default value: false -]]> - </xs:documentation> - </xs:annotation> - </xs:attribute> - <xs:attribute name="maxWaitDurationInHalfOpenState" type="xs:string"> - <xs:annotation> - <xs:documentation xml:lang="en"> -<![CDATA[ -Configures the maximum wait duration which controls how long the CircuitBreaker should stay in Half Open state, before -it switches to open. Value 0 means circuit breaker will wait in half open state until all permitted calls have been -completed. Default value: 0 ]]> </xs:documentation> </xs:annotation> @@ -3641,8 +3619,8 @@ transitions to open and starts short-circuiting calls. Default value: 100 <xs:annotation> <xs:documentation xml:lang="en"> <![CDATA[ -Configures the duration threshold above which calls are considered as slow and increase the slow calls percentage. The -default is 60 seconds. Default value: 60000 +Configures the duration threshold (seconds) above which calls are considered as slow and increase the slow calls +percentage. Default value: 60 ]]> </xs:documentation> </xs:annotation> @@ -3669,19 +3647,7 @@ Configures the max amount of concurrent calls the bulkhead will support. Default <xs:annotation> <xs:documentation xml:lang="en"> <![CDATA[ -Configures a maximum amount of time which the calling thread will wait to enter the bulkhead. The default is 0 (no -waiting). Default value: 0 -]]> - </xs:documentation> - </xs:annotation> - </xs:attribute> - <xs:attribute name="bulkheadFairCallHandlingEnabled" type="xs:string"> - <xs:annotation> - <xs:documentation xml:lang="en"> -<![CDATA[ -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 +Configures a maximum amount of time which the calling thread will wait to enter the bulkhead. Default value: 0 ]]> </xs:documentation> </xs:annotation> @@ -3726,8 +3692,7 @@ Configures whether cancel is called on the running future. Defaults to true. Def <xs:annotation> <xs:documentation xml:lang="en"> <![CDATA[ -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-resilience4j-micrometer JAR to the +Whether to enable collecting statistics using Micrometer. This requires adding camel-resilience4j-micrometer JAR to the classpath. Default value: false ]]> </xs:documentation> @@ -3954,8 +3919,8 @@ References a custom thread pool to use when offloading a guarded action to anoth <xs:annotation> <xs:documentation xml:lang="en"> <![CDATA[ -Deprecated: Whether the fallback goes over the network. If so, the fallback is executed on a separate thread-pool to -avoid exhausting the main thread-pool. Default value: false +Whether the fallback goes over the network. If so, the fallback is executed on a separate thread-pool to avoid +exhausting the main thread-pool. 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/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",
