diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/split.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/split.json index 889d6bc7dfec9..bcb4808a3b8cb 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/split.json +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/split.json @@ -31,11 +31,19 @@ "executorService": { "index": 15, "kind": "attribute", "displayName": "Executor Service", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "java.util.concurrent.ExecutorService", "deprecated": false, "autowired": false, "secret": false, "description": "Reference to a custom thread pool to use for parallel processing. Setting this option implies parallel processing." }, "onPrepare": { "index": 16, "kind": "attribute", "displayName": "On Prepare", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "org.apache.camel.Processor", "deprecated": false, "autowired": false, "secret": false, "description": "Reference to a processor for preparing the exchange to be sent. Can be used to deep-clone messages that should be sent." }, "shareUnitOfWork": { "index": 17, "kind": "attribute", "displayName": "Share Unit Of Work", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Shares the unit of work with the parent and each of the split messages. By default each split exchange has its own individual unit of work." }, - "outputs": { "index": 18, "kind": "element", "displayName": "Outputs", "group": "common", "required": true, "type": "array", "javaType": "java.util.List>", "oneOf": [ "a2aSubTask", "aggregate", "bean", "choice", "circuitBreaker", "claimCheck", "convertBodyTo", "convertHeaderTo", "convertVariableTo", "delay", "doCatch", "doFinally", "doTry", "dynamicRouter", "enrich", "filter", "idempotentConsumer", "intercept", "interceptFrom", "interceptSendToEndpoint", "kamelet", "loadBalance", "log", "loop", "marshal", "multicast", "onCompletion", "onException", "pausable", "pipeline", "policy", "poll", "pollEnrich", "process", "recipientList", "removeHeader", "removeHeaders", "removeProperties", "removeProperty", "removeVariable", "resequence", "resumable", "rollback", "routingSlip", "saga", "sample", "script", "setBody", "setExchangePattern", "setHeader", "setHeaders", "setProperty", "setVariable", "setVariables", "sort", "split", "step", "stop", "threads", "throttle", "throwException", "to", "toD", "tokenizer", "transacted", "transform", "transformDataType", "unmarshal", "validate", "wireTap" ], "deprecated": false, "autowired": false, "secret": false } + "group": { "index": 18, "kind": "attribute", "displayName": "Group", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "Groups N split messages into a single message with a java.util.List body. This allows processing items in chunks instead of one at a time." }, + "errorThreshold": { "index": 19, "kind": "attribute", "displayName": "Error Threshold", "group": "advanced", "label": "advanced", "required": false, "type": "number", "javaType": "java.lang.Double", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the error threshold as a fraction (0.0-1.0) of failed items before aborting the split operation. For example, 0.1 means abort if more than 10% of items fail. When the threshold is exceeded, a org.apache.camel.CamelExchangeException is thrown. This option is mutually exclusive with stopOnException . When set, individual item failures are tracked but processing continues until the threshold is exceeded. Note: When combined with parallelProcessing , the failure ratio may vary between runs because parallel items complete in non-deterministic order. For deterministic abort behavior with parallel processing, prefer maxFailedRecords (absolute count) over errorThreshold (ratio)." }, + "maxFailedRecords": { "index": 20, "kind": "attribute", "displayName": "Max Failed Records", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the maximum number of failed records before aborting the split operation. When the count is exceeded, a org.apache.camel.CamelExchangeException is thrown. This option is mutually exclusive with stopOnException . Can be combined with errorThreshold processing aborts when either threshold is exceeded." }, + "resumeStrategy": { "index": 21, "kind": "attribute", "displayName": "Resume Strategy", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "org.apache.camel.resume.ResumeStrategy", "deprecated": false, "autowired": false, "secret": false, "description": "Sets a ResumeStrategy for resume-from-last-position support. The watermark key must also be configured via watermarkKey(String) ." }, + "watermarkKey": { "index": 22, "kind": "attribute", "displayName": "Watermark Key", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the key to use in the watermark store." }, + "watermarkExpression": { "index": 23, "kind": "attribute", "displayName": "Watermark Expression", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Sets a Simple expression to evaluate on each completed sub-exchange to determine the new watermark value. When set, enables value-based watermarking instead of index-based. The expression is evaluated using the Simple language." }, + "outputs": { "index": 24, "kind": "element", "displayName": "Outputs", "group": "common", "required": true, "type": "array", "javaType": "java.util.List>", "oneOf": [ "a2aSubTask", "aggregate", "bean", "choice", "circuitBreaker", "claimCheck", "convertBodyTo", "convertHeaderTo", "convertVariableTo", "delay", "doCatch", "doFinally", "doTry", "dynamicRouter", "enrich", "filter", "idempotentConsumer", "intercept", "interceptFrom", "interceptSendToEndpoint", "kamelet", "loadBalance", "log", "loop", "marshal", "multicast", "onCompletion", "onException", "pausable", "pipeline", "policy", "poll", "pollEnrich", "process", "recipientList", "removeHeader", "removeHeaders", "removeProperties", "removeProperty", "removeVariable", "resequence", "resumable", "rollback", "routingSlip", "saga", "sample", "script", "setBody", "setExchangePattern", "setHeader", "setHeaders", "setProperty", "setVariable", "setVariables", "sort", "split", "step", "stop", "threads", "throttle", "throwException", "to", "toD", "tokenizer", "transacted", "transform", "transformDataType", "unmarshal", "validate", "wireTap" ], "deprecated": false, "autowired": false, "secret": false } }, "exchangeProperties": { "CamelSplitIndex": { "index": 0, "kind": "exchangeProperty", "displayName": "Split Index", "label": "producer", "required": false, "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "important": true, "description": "A split counter that increases for each Exchange being split. The counter starts from 0." }, "CamelSplitComplete": { "index": 1, "kind": "exchangeProperty", "displayName": "Split Complete", "label": "producer", "required": false, "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "description": "Whether this Exchange is the last." }, - "CamelSplitSize": { "index": 2, "kind": "exchangeProperty", "displayName": "Split Size", "label": "producer", "required": false, "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "important": true, "description": "The total number of Exchanges that was split. This property is not applied for stream based splitting, except for the very last message because then Camel knows the total size." } + "CamelSplitSize": { "index": 2, "kind": "exchangeProperty", "displayName": "Split Size", "label": "producer", "required": false, "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "important": true, "description": "The total number of Exchanges that was split. This property is not applied for stream based splitting, except for the very last message because then Camel knows the total size." }, + "CamelSplitResult": { "index": 3, "kind": "exchangeProperty", "displayName": "Split Result", "label": "producer", "required": false, "javaType": "org.apache.camel.SplitResult", "deprecated": false, "autowired": false, "secret": false, "description": "The result of a Splitter EIP operation with error thresholds, providing structured failure details." }, + "CamelSplitWatermark": { "index": 4, "kind": "exchangeProperty", "displayName": "Split Watermark", "label": "producer", "required": false, "javaType": "String", "deprecated": false, "autowired": false, "secret": false, "description": "The current watermark value from the watermark store, set before split processing begins." } } } 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 d3685f4cf0292..7e56fd14b7506 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 @@ -13262,6 +13262,70 @@ Reference to a processor for preparing the exchange to be sent. Can be used to d + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 540634bc00728..12ffe8606d961 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 @@ -12405,6 +12405,70 @@ Reference to a processor for preparing the exchange to be sent. Can be used to d + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/core/camel-api/src/generated/java/org/apache/camel/ExchangeConstantProvider.java b/core/camel-api/src/generated/java/org/apache/camel/ExchangeConstantProvider.java index b68d1400ffd47..efc177fdf5ff1 100644 --- a/core/camel-api/src/generated/java/org/apache/camel/ExchangeConstantProvider.java +++ b/core/camel-api/src/generated/java/org/apache/camel/ExchangeConstantProvider.java @@ -17,7 +17,7 @@ public class ExchangeConstantProvider { private static final Map MAP; static { - Map map = new HashMap<>(151); + Map map = new HashMap<>(153); map.put("AGGREGATED_COLLECTION_GUARD", "CamelAggregatedCollectionGuard"); map.put("AGGREGATED_COMPLETED_BY", "CamelAggregatedCompletedBy"); map.put("AGGREGATED_CORRELATION_KEY", "CamelAggregatedCorrelationKey"); @@ -148,7 +148,9 @@ public class ExchangeConstantProvider { map.put("SLIP_PRODUCER", "CamelSlipProducer"); map.put("SPLIT_COMPLETE", "CamelSplitComplete"); map.put("SPLIT_INDEX", "CamelSplitIndex"); + map.put("SPLIT_RESULT", "CamelSplitResult"); map.put("SPLIT_SIZE", "CamelSplitSize"); + map.put("SPLIT_WATERMARK", "CamelSplitWatermark"); map.put("STEP_ID", "CamelStepId"); map.put("STREAM_CACHE_UNIT_OF_WORK", "CamelStreamCacheUnitOfWork"); map.put("TIMER_COUNTER", "CamelTimerCounter"); diff --git a/core/camel-api/src/main/java/org/apache/camel/Exchange.java b/core/camel-api/src/main/java/org/apache/camel/Exchange.java index 4d1ab3e3e1d3c..27607438f0a86 100644 --- a/core/camel-api/src/main/java/org/apache/camel/Exchange.java +++ b/core/camel-api/src/main/java/org/apache/camel/Exchange.java @@ -276,6 +276,20 @@ public interface Exchange extends VariableAware { javaType = "int", important = true) String SPLIT_SIZE = "CamelSplitSize"; + /** + * @since 4.22 + */ + @Metadata(label = "split", + description = "The result of a Splitter EIP operation with error thresholds, providing structured failure details.", + javaType = "org.apache.camel.SplitResult") + String SPLIT_RESULT = "CamelSplitResult"; + /** + * @since 4.22 + */ + @Metadata(label = "split", + description = "The current watermark value from the watermark store, set before split processing begins.", + javaType = "String") + String SPLIT_WATERMARK = "CamelSplitWatermark"; @Metadata(label = "step", description = "The id of the Step EIP", javaType = "String") String STEP_ID = "CamelStepId"; diff --git a/core/camel-api/src/main/java/org/apache/camel/ExchangePropertyKey.java b/core/camel-api/src/main/java/org/apache/camel/ExchangePropertyKey.java index c106badb87833..8b4dd6415e0c4 100644 --- a/core/camel-api/src/main/java/org/apache/camel/ExchangePropertyKey.java +++ b/core/camel-api/src/main/java/org/apache/camel/ExchangePropertyKey.java @@ -88,7 +88,9 @@ public enum ExchangePropertyKey { SLIP_PRODUCER(Exchange.SLIP_PRODUCER), SPLIT_COMPLETE(Exchange.SPLIT_COMPLETE), SPLIT_INDEX(Exchange.SPLIT_INDEX), + SPLIT_RESULT(Exchange.SPLIT_RESULT), SPLIT_SIZE(Exchange.SPLIT_SIZE), + SPLIT_WATERMARK(Exchange.SPLIT_WATERMARK), STEP_ID(Exchange.STEP_ID), STREAM_CACHE_UNIT_OF_WORK(Exchange.STREAM_CACHE_UNIT_OF_WORK), TO_ENDPOINT(Exchange.TO_ENDPOINT), @@ -222,8 +224,12 @@ public String getName() { return SPLIT_COMPLETE; case Exchange.SPLIT_INDEX: return SPLIT_INDEX; + case Exchange.SPLIT_RESULT: + return SPLIT_RESULT; case Exchange.SPLIT_SIZE: return SPLIT_SIZE; + case Exchange.SPLIT_WATERMARK: + return SPLIT_WATERMARK; case Exchange.STEP_ID: return STEP_ID; case Exchange.STREAM_CACHE_UNIT_OF_WORK: diff --git a/core/camel-api/src/main/java/org/apache/camel/SplitResult.java b/core/camel-api/src/main/java/org/apache/camel/SplitResult.java new file mode 100644 index 0000000000000..6dbf371ef6bac --- /dev/null +++ b/core/camel-api/src/main/java/org/apache/camel/SplitResult.java @@ -0,0 +1,97 @@ +/* + * 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; + +import java.util.Collections; +import java.util.List; + +/** + * Result of a Splitter EIP operation that provides structured information about the outcome, including failure details + * when error thresholds ({@code errorThreshold} or {@code maxFailedRecords}) are configured. + *

+ * This result is available as an exchange property ({@link Exchange#SPLIT_RESULT}) after the split operation completes. + * + * @since 4.22 + */ +public final class SplitResult { + + /** + * Details of a single split item failure. + * + * @param index the 0-based index of the failed item in the split sequence + * @param exception the exception that caused the failure + * @since 4.22 + */ + public record Failure(int index, Exception exception) { + } + + private final int totalItems; + private final int failureCount; + private final List failures; + private final boolean aborted; + + public SplitResult(int totalItems, int failureCount, List failures, boolean aborted) { + this.totalItems = totalItems; + this.failureCount = failureCount; + this.failures = failures != null ? Collections.unmodifiableList(failures) : Collections.emptyList(); + this.aborted = aborted; + } + + /** + * The total number of items that were prepared for splitting. When {@code group()} is used, this counts the number + * of chunks (groups), not the number of individual elements within them. + */ + public int getTotalItems() { + return totalItems; + } + + /** + * The number of items that completed successfully. + */ + public int getSuccessCount() { + return totalItems - failureCount; + } + + /** + * The number of items that failed during processing. + */ + public int getFailureCount() { + return failureCount; + } + + /** + * The list of individual failures with their index and exception details. + */ + public List getFailures() { + return failures; + } + + /** + * Whether the split operation was aborted early because an error threshold was exceeded. + */ + public boolean isAborted() { + return aborted; + } + + @Override + public String toString() { + return "SplitResult[total=" + totalItems + + ", success=" + getSuccessCount() + + ", failures=" + failureCount + + ", aborted=" + aborted + "]"; + } +} diff --git a/core/camel-core-engine/src/main/docs/modules/eips/pages/split-eip.adoc b/core/camel-core-engine/src/main/docs/modules/eips/pages/split-eip.adoc index 54019e0b2b520..d69500c1e42ef 100644 --- a/core/camel-core-engine/src/main/docs/modules/eips/pages/split-eip.adoc +++ b/core/camel-core-engine/src/main/docs/modules/eips/pages/split-eip.adoc @@ -1078,6 +1078,359 @@ BuildCombinedResponse: (id=1,item=A);(id=2,item=B);(id=3,item=C) Response to caller: Response[(id=1,item=A);(id=2,item=B);(id=3,item=C)] ---- +=== Chunking with group + +The `group` option on the Split EIP allows grouping N split items together into a single message +with a `java.util.List` body. This is useful when you want to process items in batches rather than +one at a time. + +[tabs] +==== +Java:: ++ +[source,java] +---- +from("direct:start") + .split(body()).group(3) + .to("mock:batch"); +---- + +XML:: ++ +[source,xml] +---- + + + + ${body} + + + +---- + +YAML:: ++ +[source,yaml] +---- +- route: + from: + uri: direct:start + steps: + - split: + group: "3" + expression: + simple: + expression: "${body}" + steps: + - to: + uri: mock:batch +---- +==== + +If the input has 7 items, the route above produces 3 exchanges: one with items 1-3, one with items 4-6, +and one with item 7. + +NOTE: This `group` option is on the Split EIP definition itself and works with any expression. +It is different from the `group` option on the xref:components:languages:tokenize-language.adoc[Tokenize] language, +which groups tokenized text lines together. + +=== Error handling with maxFailedRecords and errorThreshold + +The `stopOnException` option is all-or-nothing: a single failure stops the entire split. +For more fine-grained control, the Splitter provides two error threshold options that let processing +continue through some failures while stopping when the error rate becomes unacceptable. + +==== maxFailedRecords + +The `maxFailedRecords` option sets the maximum number of failed split items before aborting. +Processing continues as long as the failure count stays below this threshold. When the threshold +is reached, the splitter stops and sets an exception on the exchange. + +[tabs] +==== +Java:: ++ +[source,java] +---- +from("direct:start") + .split(body()).maxFailedRecords(5) + .process(exchange -> { + // processing logic that may throw exceptions + }) + .to("mock:result"); +---- + +XML:: ++ +[source,xml] +---- + + + + ${body} + + + + +---- + +YAML:: ++ +[source,yaml] +---- +- route: + from: + uri: direct:start + steps: + - split: + maxFailedRecords: "5" + expression: + simple: + expression: "${body}" + steps: + - process: + ref: myProcessor + - to: + uri: mock:result +---- +==== + +In this example, the first 4 failures are tolerated and processing continues. +When the 5th failure occurs, the splitter stops and the exchange will have an exception set. + +==== errorThreshold + +The `errorThreshold` option sets the maximum allowed failure ratio as a fraction between 0.0 and 1.0. +After each failure, the ratio of failed items to total processed items is calculated. If this ratio +meets or exceeds the threshold, the splitter stops. + +[tabs] +==== +Java:: ++ +[source,java] +---- +from("direct:start") + .split(body()).errorThreshold(0.5) + .process(exchange -> { + // processing logic that may throw exceptions + }) + .to("mock:result"); +---- + +XML:: ++ +[source,xml] +---- + + + + ${body} + + + + +---- + +YAML:: ++ +[source,yaml] +---- +- route: + from: + uri: direct:start + steps: + - split: + errorThreshold: "0.5" + expression: + simple: + expression: "${body}" + steps: + - process: + ref: myProcessor + - to: + uri: mock:result +---- +==== + +In this example, if 50% or more of the processed items have failed, the splitter stops. + +Both `maxFailedRecords` and `errorThreshold` can be combined. The splitter stops when either +threshold is exceeded. + +IMPORTANT: The `stopOnException` option is mutually exclusive with `maxFailedRecords` and +`errorThreshold`. You cannot use `stopOnException` together with either of these options. + +NOTE: When error thresholds are configured, individual item exceptions are cleared from sub-exchanges +after being recorded in the `SplitResult`. This means a custom `AggregationStrategy` will not see +individual item exceptions — use the `SplitResult` exchange property to access failure details instead. + +NOTE: When using `errorThreshold` with `parallelProcessing`, the failure ratio may vary slightly +between runs because the ratio is calculated as failures are reported, and the order in which +parallel items complete is non-deterministic. For deterministic abort behavior with parallel +processing, prefer `maxFailedRecords` (absolute count) over `errorThreshold` (ratio). + +==== SplitResult + +When `maxFailedRecords` or `errorThreshold` is configured, the splitter makes a +`SplitResult` object available as an exchange property (`CamelSplitResult`) after the split completes. +This provides structured information about the outcome: + +[source,java] +---- +Exchange result = template.send("direct:start", + e -> e.getIn().setBody(myItems)); + +SplitResult splitResult = result.getProperty(Exchange.SPLIT_RESULT, SplitResult.class); +if (splitResult != null) { + int total = splitResult.getTotalItems(); // total items (or chunks when group() is used) + int success = splitResult.getSuccessCount(); // successful items + int failures = splitResult.getFailureCount(); // failed items + boolean aborted = splitResult.isAborted(); // true if a threshold was exceeded + + // inspect individual failures + for (SplitResult.Failure failure : splitResult.getFailures()) { + int index = failure.index(); // 0-based index of failed item + Exception ex = failure.exception(); // the exception that occurred + } +} +---- + +=== Watermark tracking + +The Splitter supports watermark tracking for incremental processing scenarios using Camel's +`ResumeStrategy` SPI. A watermark records how far processing has progressed, so subsequent +runs can skip already-processed items. + +This is useful when processing a data source repeatedly (e.g., polling a database or file) +where you want to resume from where the last run left off. You can use any `ResumeStrategy` +implementation for persistence — from a simple in-memory strategy for testing to a Kafka-backed +strategy for production use. + +==== Index-based watermark + +The simplest form uses the split index as the watermark. On each run, items up to and including +the stored watermark index are skipped. After successful processing, the watermark is updated +to the last processed index. + +[tabs] +==== +Java:: ++ +[source,java] +---- +ResumeStrategy strategy = ... // any ResumeStrategy implementation + +from("direct:start") + .split(body()).resumeStrategy(strategy, "myJob") + .to("mock:result"); +---- + +XML:: ++ +[source,xml] +---- + + + + ${body} + + + +---- + +YAML:: ++ +[source,yaml] +---- +- route: + from: + uri: direct:start + steps: + - split: + resumeStrategy: "#myStrategy" + watermarkKey: myJob + expression: + simple: + expression: "${body}" + steps: + - to: + uri: mock:result +---- +==== + +On the first run with 5 items, all are processed and the watermark is stored as `"4"` (the last 0-based index). +On the next run with the same 5 items, items 0 through 4 are skipped and nothing is processed. +If the data source grows to 8 items, only items 5, 6, and 7 are processed. + +==== Value-based watermark + +For more control, you can use a `watermarkExpression` to extract a watermark value from each processed item. +The value from the last successfully processed item (by index order) is stored. + +[tabs] +==== +Java:: ++ +[source,java] +---- +ResumeStrategy strategy = ... // any ResumeStrategy implementation + +from("direct:start") + .split(body()) + .resumeStrategy(strategy, "dateJob") + .watermarkExpression("${body}") + .to("mock:result"); +---- + +XML:: ++ +[source,xml] +---- + + + + ${body} + + + +---- + +YAML:: ++ +[source,yaml] +---- +- route: + from: + uri: direct:start + steps: + - split: + resumeStrategy: "#myStrategy" + watermarkKey: dateJob + watermarkExpression: "${body}" + expression: + simple: + expression: "${body}" + steps: + - to: + uri: mock:result +---- +==== + +With value-based watermarking, the previous watermark value is exposed as the exchange property +`CamelSplitWatermark` before split processing begins. You can use this to filter items in your +processing logic. + +NOTE: The `watermarkExpression` option uses the Simple language for expression evaluation. + +NOTE: The watermark is only updated when the split completes successfully. If the split is aborted +(e.g., due to exceeding `maxFailedRecords`), the watermark is not updated, which allows the +failed batch to be retried. + +IMPORTANT: Watermark tracking assumes sequential route invocations (e.g., batch jobs triggered by +a `timer` or `scheduler`). If multiple exchanges hit the same route concurrently, they will read +the same watermark and may process duplicate items. Use a single-consumer pattern for watermark-based routes. + === Stop processing in case of exception The Splitter will by default continue to process diff --git a/core/camel-core-model/src/generated/resources/META-INF/org/apache/camel/model/split.json b/core/camel-core-model/src/generated/resources/META-INF/org/apache/camel/model/split.json index 889d6bc7dfec9..bcb4808a3b8cb 100644 --- a/core/camel-core-model/src/generated/resources/META-INF/org/apache/camel/model/split.json +++ b/core/camel-core-model/src/generated/resources/META-INF/org/apache/camel/model/split.json @@ -31,11 +31,19 @@ "executorService": { "index": 15, "kind": "attribute", "displayName": "Executor Service", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "java.util.concurrent.ExecutorService", "deprecated": false, "autowired": false, "secret": false, "description": "Reference to a custom thread pool to use for parallel processing. Setting this option implies parallel processing." }, "onPrepare": { "index": 16, "kind": "attribute", "displayName": "On Prepare", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "org.apache.camel.Processor", "deprecated": false, "autowired": false, "secret": false, "description": "Reference to a processor for preparing the exchange to be sent. Can be used to deep-clone messages that should be sent." }, "shareUnitOfWork": { "index": 17, "kind": "attribute", "displayName": "Share Unit Of Work", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Shares the unit of work with the parent and each of the split messages. By default each split exchange has its own individual unit of work." }, - "outputs": { "index": 18, "kind": "element", "displayName": "Outputs", "group": "common", "required": true, "type": "array", "javaType": "java.util.List>", "oneOf": [ "a2aSubTask", "aggregate", "bean", "choice", "circuitBreaker", "claimCheck", "convertBodyTo", "convertHeaderTo", "convertVariableTo", "delay", "doCatch", "doFinally", "doTry", "dynamicRouter", "enrich", "filter", "idempotentConsumer", "intercept", "interceptFrom", "interceptSendToEndpoint", "kamelet", "loadBalance", "log", "loop", "marshal", "multicast", "onCompletion", "onException", "pausable", "pipeline", "policy", "poll", "pollEnrich", "process", "recipientList", "removeHeader", "removeHeaders", "removeProperties", "removeProperty", "removeVariable", "resequence", "resumable", "rollback", "routingSlip", "saga", "sample", "script", "setBody", "setExchangePattern", "setHeader", "setHeaders", "setProperty", "setVariable", "setVariables", "sort", "split", "step", "stop", "threads", "throttle", "throwException", "to", "toD", "tokenizer", "transacted", "transform", "transformDataType", "unmarshal", "validate", "wireTap" ], "deprecated": false, "autowired": false, "secret": false } + "group": { "index": 18, "kind": "attribute", "displayName": "Group", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "Groups N split messages into a single message with a java.util.List body. This allows processing items in chunks instead of one at a time." }, + "errorThreshold": { "index": 19, "kind": "attribute", "displayName": "Error Threshold", "group": "advanced", "label": "advanced", "required": false, "type": "number", "javaType": "java.lang.Double", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the error threshold as a fraction (0.0-1.0) of failed items before aborting the split operation. For example, 0.1 means abort if more than 10% of items fail. When the threshold is exceeded, a org.apache.camel.CamelExchangeException is thrown. This option is mutually exclusive with stopOnException . When set, individual item failures are tracked but processing continues until the threshold is exceeded. Note: When combined with parallelProcessing , the failure ratio may vary between runs because parallel items complete in non-deterministic order. For deterministic abort behavior with parallel processing, prefer maxFailedRecords (absolute count) over errorThreshold (ratio)." }, + "maxFailedRecords": { "index": 20, "kind": "attribute", "displayName": "Max Failed Records", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the maximum number of failed records before aborting the split operation. When the count is exceeded, a org.apache.camel.CamelExchangeException is thrown. This option is mutually exclusive with stopOnException . Can be combined with errorThreshold processing aborts when either threshold is exceeded." }, + "resumeStrategy": { "index": 21, "kind": "attribute", "displayName": "Resume Strategy", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "org.apache.camel.resume.ResumeStrategy", "deprecated": false, "autowired": false, "secret": false, "description": "Sets a ResumeStrategy for resume-from-last-position support. The watermark key must also be configured via watermarkKey(String) ." }, + "watermarkKey": { "index": 22, "kind": "attribute", "displayName": "Watermark Key", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the key to use in the watermark store." }, + "watermarkExpression": { "index": 23, "kind": "attribute", "displayName": "Watermark Expression", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Sets a Simple expression to evaluate on each completed sub-exchange to determine the new watermark value. When set, enables value-based watermarking instead of index-based. The expression is evaluated using the Simple language." }, + "outputs": { "index": 24, "kind": "element", "displayName": "Outputs", "group": "common", "required": true, "type": "array", "javaType": "java.util.List>", "oneOf": [ "a2aSubTask", "aggregate", "bean", "choice", "circuitBreaker", "claimCheck", "convertBodyTo", "convertHeaderTo", "convertVariableTo", "delay", "doCatch", "doFinally", "doTry", "dynamicRouter", "enrich", "filter", "idempotentConsumer", "intercept", "interceptFrom", "interceptSendToEndpoint", "kamelet", "loadBalance", "log", "loop", "marshal", "multicast", "onCompletion", "onException", "pausable", "pipeline", "policy", "poll", "pollEnrich", "process", "recipientList", "removeHeader", "removeHeaders", "removeProperties", "removeProperty", "removeVariable", "resequence", "resumable", "rollback", "routingSlip", "saga", "sample", "script", "setBody", "setExchangePattern", "setHeader", "setHeaders", "setProperty", "setVariable", "setVariables", "sort", "split", "step", "stop", "threads", "throttle", "throwException", "to", "toD", "tokenizer", "transacted", "transform", "transformDataType", "unmarshal", "validate", "wireTap" ], "deprecated": false, "autowired": false, "secret": false } }, "exchangeProperties": { "CamelSplitIndex": { "index": 0, "kind": "exchangeProperty", "displayName": "Split Index", "label": "producer", "required": false, "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "important": true, "description": "A split counter that increases for each Exchange being split. The counter starts from 0." }, "CamelSplitComplete": { "index": 1, "kind": "exchangeProperty", "displayName": "Split Complete", "label": "producer", "required": false, "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "description": "Whether this Exchange is the last." }, - "CamelSplitSize": { "index": 2, "kind": "exchangeProperty", "displayName": "Split Size", "label": "producer", "required": false, "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "important": true, "description": "The total number of Exchanges that was split. This property is not applied for stream based splitting, except for the very last message because then Camel knows the total size." } + "CamelSplitSize": { "index": 2, "kind": "exchangeProperty", "displayName": "Split Size", "label": "producer", "required": false, "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "important": true, "description": "The total number of Exchanges that was split. This property is not applied for stream based splitting, except for the very last message because then Camel knows the total size." }, + "CamelSplitResult": { "index": 3, "kind": "exchangeProperty", "displayName": "Split Result", "label": "producer", "required": false, "javaType": "org.apache.camel.SplitResult", "deprecated": false, "autowired": false, "secret": false, "description": "The result of a Splitter EIP operation with error thresholds, providing structured failure details." }, + "CamelSplitWatermark": { "index": 4, "kind": "exchangeProperty", "displayName": "Split Watermark", "label": "producer", "required": false, "javaType": "String", "deprecated": false, "autowired": false, "secret": false, "description": "The current watermark value from the watermark store, set before split processing begins." } } } diff --git a/core/camel-core-model/src/main/java/org/apache/camel/model/SplitDefinition.java b/core/camel-core-model/src/main/java/org/apache/camel/model/SplitDefinition.java index 4ff1356fd9086..c3c412a8aec4c 100644 --- a/core/camel-core-model/src/main/java/org/apache/camel/model/SplitDefinition.java +++ b/core/camel-core-model/src/main/java/org/apache/camel/model/SplitDefinition.java @@ -28,6 +28,7 @@ import org.apache.camel.Expression; import org.apache.camel.Processor; import org.apache.camel.model.language.ExpressionDefinition; +import org.apache.camel.resume.ResumeStrategy; import org.apache.camel.spi.Metadata; /** @@ -47,6 +48,8 @@ public class SplitDefinition extends OutputExpressionNode implements ExecutorSer private AggregationStrategy aggregationStrategyBean; @XmlTransient private Processor onPrepareProcessor; + @XmlTransient + private ResumeStrategy resumeStrategyBean; @XmlAttribute @Metadata(defaultValue = ",", @@ -115,6 +118,24 @@ public class SplitDefinition extends OutputExpressionNode implements ExecutorSer description = "Shares the unit of work with the parent and each of the split messages." + " By default each split exchange has its own individual unit of work.") private String shareUnitOfWork; + @XmlAttribute + @Metadata(label = "advanced", javaType = "java.lang.Integer") + private String group; + @XmlAttribute + @Metadata(label = "advanced", javaType = "java.lang.Double") + private String errorThreshold; + @XmlAttribute + @Metadata(label = "advanced", javaType = "java.lang.Integer") + private String maxFailedRecords; + @XmlAttribute + @Metadata(label = "advanced", javaType = "org.apache.camel.resume.ResumeStrategy") + private String resumeStrategy; + @XmlAttribute + @Metadata(label = "advanced") + private String watermarkKey; + @XmlAttribute + @Metadata(label = "advanced") + private String watermarkExpression; public SplitDefinition() { } @@ -137,6 +158,13 @@ public SplitDefinition(SplitDefinition source) { this.executorService = source.executorService; this.onPrepare = source.onPrepare; this.shareUnitOfWork = source.shareUnitOfWork; + this.group = source.group; + this.errorThreshold = source.errorThreshold; + this.maxFailedRecords = source.maxFailedRecords; + this.resumeStrategyBean = source.resumeStrategyBean; + this.resumeStrategy = source.resumeStrategy; + this.watermarkKey = source.watermarkKey; + this.watermarkExpression = source.watermarkExpression; } public SplitDefinition(Expression expression) { @@ -596,6 +624,161 @@ public SplitDefinition shareUnitOfWork(String shareUnitOfWork) { return this; } + /** + * Groups N split messages into a single message with a {@link java.util.List} body. This allows processing items in + * chunks instead of one at a time. + * + * @param group the number of items per group + * @return the builder + */ + public SplitDefinition group(int group) { + return group(Integer.toString(group)); + } + + /** + * Groups N split messages into a single message with a {@link java.util.List} body. This allows processing items in + * chunks instead of one at a time. + * + * @param group the number of items per group + * @return the builder + */ + public SplitDefinition group(String group) { + setGroup(group); + return this; + } + + /** + * Sets the error threshold as a fraction (0.0-1.0) of failed items before aborting the split operation. For + * example, 0.1 means abort if more than 10% of items fail. When the threshold is exceeded, a + * {@link org.apache.camel.CamelExchangeException} is thrown. + *

+ * This option is mutually exclusive with {@code stopOnException}. When set, individual item failures are tracked + * but processing continues until the threshold is exceeded. + *

+ * Note: When combined with {@code parallelProcessing}, the failure ratio may vary between runs because + * parallel items complete in non-deterministic order. For deterministic abort behavior with parallel processing, + * prefer {@code maxFailedRecords} (absolute count) over {@code errorThreshold} (ratio). + * + * @param errorThreshold the failure ratio threshold (0.0-1.0) + * @return the builder + */ + public SplitDefinition errorThreshold(double errorThreshold) { + return errorThreshold(Double.toString(errorThreshold)); + } + + /** + * Sets the error threshold as a fraction (0.0-1.0) of failed items before aborting the split operation. For + * example, 0.1 means abort if more than 10% of items fail. When the threshold is exceeded, a + * {@link org.apache.camel.CamelExchangeException} is thrown. + *

+ * This option is mutually exclusive with {@code stopOnException}. When set, individual item failures are tracked + * but processing continues until the threshold is exceeded. + * + * @param errorThreshold the failure ratio threshold (0.0-1.0) + * @return the builder + */ + public SplitDefinition errorThreshold(String errorThreshold) { + setErrorThreshold(errorThreshold); + return this; + } + + /** + * Sets the maximum number of failed records before aborting the split operation. When the count is exceeded, a + * {@link org.apache.camel.CamelExchangeException} is thrown. + *

+ * This option is mutually exclusive with {@code stopOnException}. Can be combined with {@code errorThreshold} — + * processing aborts when either threshold is exceeded. + * + * @param maxFailedRecords the maximum number of allowed failures + * @return the builder + */ + public SplitDefinition maxFailedRecords(int maxFailedRecords) { + return maxFailedRecords(Integer.toString(maxFailedRecords)); + } + + /** + * Sets the maximum number of failed records before aborting the split operation. When the count is exceeded, a + * {@link org.apache.camel.CamelExchangeException} is thrown. + *

+ * This option is mutually exclusive with {@code stopOnException}. Can be combined with {@code errorThreshold} — + * processing aborts when either threshold is exceeded. + * + * @param maxFailedRecords the maximum number of allowed failures + * @return the builder + */ + public SplitDefinition maxFailedRecords(String maxFailedRecords) { + setMaxFailedRecords(maxFailedRecords); + return this; + } + + /** + * Sets a {@link ResumeStrategy} and key for resume-from-last-position support. When configured, the Splitter tracks + * progress and can skip already-processed items on subsequent runs. + *

+ * With index-based watermarking (no {@code watermarkExpression}), items up to the stored index are automatically + * skipped. With value-based watermarking (with {@code watermarkExpression}), the stored value is exposed as an + * exchange property ({@link org.apache.camel.Exchange#SPLIT_WATERMARK}) for upstream filtering. + *

+ * The watermark is only updated on successful completion — aborted runs preserve the previous watermark to allow + * retry. + * + * @param strategy the ResumeStrategy to persist watermark state + * @param key the key to use in the strategy + * @return the builder + */ + public SplitDefinition resumeStrategy(ResumeStrategy strategy, String key) { + this.resumeStrategyBean = strategy; + setWatermarkKey(key); + return this; + } + + /** + * Sets a {@link ResumeStrategy} for resume-from-last-position support. The watermark key must also be configured + * via {@link #watermarkKey(String)}. + * + * @param strategy the ResumeStrategy to persist watermark state + * @return the builder + */ + public SplitDefinition resumeStrategy(ResumeStrategy strategy) { + this.resumeStrategyBean = strategy; + return this; + } + + /** + * Sets a reference to a {@link ResumeStrategy} in the registry. + * + * @param resumeStrategy reference to the strategy bean + * @return the builder + */ + public SplitDefinition resumeStrategy(String resumeStrategy) { + setResumeStrategy(resumeStrategy); + return this; + } + + /** + * Sets the key to use in the watermark store. + * + * @param watermarkKey the key + * @return the builder + */ + public SplitDefinition watermarkKey(String watermarkKey) { + setWatermarkKey(watermarkKey); + return this; + } + + /** + * Sets a Simple expression to evaluate on each completed sub-exchange to determine the new watermark value. When + * set, enables value-based watermarking instead of index-based. The expression is evaluated using the Simple + * language. + * + * @param watermarkExpression the Simple expression + * @return the builder + */ + public SplitDefinition watermarkExpression(String watermarkExpression) { + setWatermarkExpression(watermarkExpression); + return this; + } + // Properties // ------------------------------------------------------------------------- @@ -733,4 +916,77 @@ public String getExecutorService() { public void setExecutorService(String executorService) { this.executorService = executorService; } + + public String getGroup() { + return group; + } + + /** + * Groups N split messages into a single message with a {@link java.util.List} body. This allows processing items in + * chunks instead of one at a time. + */ + public void setGroup(String group) { + this.group = group; + } + + public String getErrorThreshold() { + return errorThreshold; + } + + /** + * Sets the error threshold as a fraction (0.0-1.0) of failed items before aborting the split operation. + */ + public void setErrorThreshold(String errorThreshold) { + this.errorThreshold = errorThreshold; + } + + public String getMaxFailedRecords() { + return maxFailedRecords; + } + + /** + * Sets the maximum number of failed records before aborting the split operation. + */ + public void setMaxFailedRecords(String maxFailedRecords) { + this.maxFailedRecords = maxFailedRecords; + } + + public ResumeStrategy getResumeStrategyBean() { + return resumeStrategyBean; + } + + public String getResumeStrategy() { + return resumeStrategy; + } + + /** + * Sets a reference to a {@link ResumeStrategy} in the registry for resume-from-last-position support. + */ + public void setResumeStrategy(String resumeStrategy) { + this.resumeStrategy = resumeStrategy; + } + + public String getWatermarkKey() { + return watermarkKey; + } + + /** + * Sets the key to use in the watermark store. + */ + public void setWatermarkKey(String watermarkKey) { + this.watermarkKey = watermarkKey; + } + + public String getWatermarkExpression() { + return watermarkExpression; + } + + /** + * Sets a Simple expression to evaluate on each completed sub-exchange to determine the new watermark value. When + * set, enables value-based watermarking instead of index-based. The expression is evaluated using the Simple + * language. + */ + public void setWatermarkExpression(String watermarkExpression) { + this.watermarkExpression = watermarkExpression; + } } diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java index e6a1d4cc59199..419e62ae4e4b0 100644 --- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java +++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java @@ -599,7 +599,7 @@ public void run() { msg = "Multicast processing failed for number " + index; } boolean continueProcessing = PipelineHelper.continueProcessing(exchange, msg, LOG); - if (stopOnException && !continueProcessing) { + if (!continueProcessing && !shouldContinueOnFailure(exchange, original, index)) { if (exchange.getException() != null) { // wrap in exception to explain where it failed exchange.setException(new CamelExchangeException( @@ -735,7 +735,7 @@ boolean doRun() throws Exception { msg = "Multicast processing failed for number " + index; } boolean continueProcessing = PipelineHelper.continueProcessing(exchange, msg, LOG); - if (stopOnException && !continueProcessing) { + if (!continueProcessing && !shouldContinueOnFailure(exchange, original, index)) { if (exchange.getException() != null) { // wrap in exception to explain where it failed exchange.setException(new CamelExchangeException( @@ -771,6 +771,22 @@ boolean doRun() throws Exception { } } + /** + * Determines whether processing should continue after a sub-exchange has failed. + *

+ * The default implementation returns {@code false} when {@code stopOnException} is enabled (meaning processing + * should stop). Subclasses (e.g., Splitter) can override this to implement more sophisticated failure policies such + * as error threshold checking. + * + * @param subExchange the failed sub-exchange + * @param original the original exchange + * @param index the index of the failed sub-exchange + * @return {@code true} to continue processing despite the failure, {@code false} to stop + */ + protected boolean shouldContinueOnFailure(Exchange subExchange, Exchange original, int index) { + return !stopOnException; + } + protected ScheduledFuture schedule(Executor executor, Runnable runnable, long delay, TimeUnit unit) { if (executor instanceof ScheduledExecutorService scheduledExecutorService) { return scheduledExecutorService.schedule(runnable, delay, unit); diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java index 4620ecf945f42..841e9f610459d 100644 --- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java +++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java @@ -25,7 +25,10 @@ import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutorService; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import org.apache.camel.AggregationStrategy; import org.apache.camel.AsyncCallback; @@ -37,12 +40,24 @@ import org.apache.camel.Processor; import org.apache.camel.Route; import org.apache.camel.RuntimeCamelException; +import org.apache.camel.SplitResult; import org.apache.camel.processor.aggregate.ShareUnitOfWorkAggregationStrategy; import org.apache.camel.processor.aggregate.UseOriginalAggregationStrategy; +import org.apache.camel.resume.Offset; +import org.apache.camel.resume.OffsetKey; +import org.apache.camel.resume.ResumeStrategy; +import org.apache.camel.resume.ResumeStrategyConfiguration; +import org.apache.camel.resume.cache.ResumeCache; import org.apache.camel.support.ExchangeHelper; +import org.apache.camel.support.GroupIterator; import org.apache.camel.support.ObjectHelper; +import org.apache.camel.support.resume.OffsetKeys; +import org.apache.camel.support.resume.Offsets; +import org.apache.camel.support.service.ServiceHelper; import org.apache.camel.util.IOHelper; import org.apache.camel.util.StringHelper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import static org.apache.camel.util.ObjectHelper.notNull; @@ -52,10 +67,27 @@ */ public class Splitter extends MulticastProcessor { + private static final Logger LOG = LoggerFactory.getLogger(Splitter.class); + private static final String IGNORE_DELIMITER_MARKER = "false"; private static final String SINGLE_DELIMITER_MARKER = "single"; + private static final String SPLIT_FAILURE_TRACKER = "CamelSplitFailureTracker"; + private static final String SPLIT_WATERMARK_OFFSET = "CamelSplitWatermarkOffset"; + private static final String SPLIT_WATERMARK_COUNT = "CamelSplitWatermarkCount"; + private static final String SPLIT_WATERMARK_LATEST = "CamelSplitWatermarkLatest"; + + private record IndexedWatermark(int index, String value) { + } + private final Expression expression; private final String delimiter; + private int group; + private double errorThreshold; + private int maxFailedRecords; + private ResumeStrategy resumeStrategy; + private final ConcurrentHashMap watermarkCache = new ConcurrentHashMap<>(); + private String watermarkKey; + private Expression watermarkExpression; public Splitter(CamelContext camelContext, Route route, Expression expression, Processor destination, AggregationStrategy aggregationStrategy, boolean parallelProcessing, @@ -87,15 +119,67 @@ public String getTraceLabel() { return "split[" + expression + "]"; } - @Override - protected void doBuild() throws Exception { - super.doBuild(); - } - @Override protected void doInit() throws Exception { super.doInit(); expression.init(getCamelContext()); + if (watermarkExpression != null) { + watermarkExpression.init(getCamelContext()); + } + if (resumeStrategy != null && watermarkKey != null) { + ServiceHelper.startService(resumeStrategy); + resumeStrategy.loadCache(); + } + } + + @Override + protected void doStop() throws Exception { + if (resumeStrategy != null) { + ServiceHelper.stopService(resumeStrategy); + } + super.doStop(); + } + + /** + * Reads the current watermark value. Checks the local cache first (populated from previous exchanges), then falls + * back to the strategy's resume cache (populated by {@code loadCache()} on init or backed by external storage). + */ + private String readCurrentWatermark() { + String value = watermarkCache.get(watermarkKey); + if (value != null) { + return value; + } + return readFromStrategyCache(); + } + + private String readFromStrategyCache() { + try { + ResumeStrategyConfiguration config = resumeStrategy.getResumeStrategyConfiguration(); + if (config != null) { + ResumeCache cache = config.getResumeCache(); + if (cache != null) { + String[] result = new String[1]; + cache.forEach((key, value) -> { + String keyStr = key instanceof OffsetKey ok + ? String.valueOf(ok.getValue()) : String.valueOf(key); + if (watermarkKey.equals(keyStr) && value != null) { + result[0] = toWatermarkString(value); + return false; + } + return true; + }); + return result[0]; + } + } + } catch (Exception e) { + LOG.debug("Failed to read watermark from resume strategy cache for key '{}': {}", + watermarkKey, e.getMessage(), e); + } + return null; + } + + private static String toWatermarkString(Object value) { + return value instanceof Offset o ? String.valueOf(o.getValue()) : String.valueOf(value); } @Override @@ -123,7 +207,60 @@ public boolean process(Exchange exchange, final AsyncCallback callback) { setAggregationStrategyOnExchange(exchange, original); } - return super.process(exchange, callback); + boolean hasErrorThreshold = setupFailureTracking(exchange); + boolean hasWatermark = setupWatermarkTracking(exchange); + AsyncCallback wrappedCallback = wrapCallback(callback, exchange, hasErrorThreshold, hasWatermark); + + return super.process(exchange, wrappedCallback); + } + + private boolean setupFailureTracking(Exchange exchange) { + boolean hasErrorThreshold = errorThreshold > 0 || maxFailedRecords > 0; + if (hasErrorThreshold) { + exchange.setProperty(SPLIT_FAILURE_TRACKER, new SplitFailureTracker()); + } + return hasErrorThreshold; + } + + private boolean setupWatermarkTracking(Exchange exchange) { + boolean hasWatermark = resumeStrategy != null && watermarkKey != null; + if (hasWatermark) { + String currentWatermark = readCurrentWatermark(); + if (currentWatermark != null) { + exchange.setProperty(Exchange.SPLIT_WATERMARK, currentWatermark); + } + if (watermarkExpression != null) { + exchange.setProperty(SPLIT_WATERMARK_LATEST, new AtomicReference()); + } + } + return hasWatermark; + } + + private AsyncCallback wrapCallback( + AsyncCallback callback, Exchange exchange, boolean hasErrorThreshold, boolean hasWatermark) { + if (!hasErrorThreshold && !hasWatermark) { + return callback; + } + return doneSync -> { + if (hasErrorThreshold) { + buildSplitResult(exchange); + } + if (hasWatermark) { + updateWatermark(exchange); + } + callback.done(doneSync); + }; + } + + @Override + protected ProcessorExchangePair createProcessorExchangePair( + int index, Processor processor, Exchange exchange, Route route) { + ProcessorExchangePair pair = super.createProcessorExchangePair(index, processor, exchange, route); + // wrap to evaluate watermark expression on each completed sub-exchange + if (watermarkExpression != null) { + return new WatermarkProcessorExchangePair(pair, exchange); + } + return pair; } @Override @@ -161,19 +298,60 @@ private final class SplitterIterable implements Iterable, private Exchange copy; private final Route route; private final Exchange original; + // tracks individual (raw) item count, independent of grouping + private final AtomicInteger rawItemCount = new AtomicInteger(); private SplitterIterable(Exchange exchange, Object value) { this.original = exchange; this.value = value; + Iterator rawIterator; if (IGNORE_DELIMITER_MARKER.equalsIgnoreCase(delimiter)) { - this.iterator = ObjectHelper.createIterator(value, null); + rawIterator = ObjectHelper.createIterator(value, null); } else if (SINGLE_DELIMITER_MARKER.equalsIgnoreCase(delimiter)) { // force single element - this.iterator = ObjectHelper.createIterator(List.of(value)); + rawIterator = ObjectHelper.createIterator(List.of(value)); } else { - this.iterator = ObjectHelper.createIterator(value, delimiter); + rawIterator = ObjectHelper.createIterator(value, delimiter); + } + + // index-based watermarking: skip items up to the stored watermark index + int skipCount = 0; + if (resumeStrategy != null && watermarkKey != null && watermarkExpression == null) { + // reuse watermark already read in process() and set as exchange property + String storedIndex = exchange.getProperty(Exchange.SPLIT_WATERMARK, String.class); + if (storedIndex == null) { + storedIndex = readCurrentWatermark(); + } + if (storedIndex != null) { + int skipTo = Integer.parseInt(storedIndex); + while (rawIterator.hasNext() && skipCount <= skipTo) { + rawIterator.next(); + skipCount++; + } + } } + if (skipCount > 0) { + exchange.setProperty(SPLIT_WATERMARK_OFFSET, skipCount); + } + + // wrap rawIterator in a counting decorator to track individual items + // consumed, independent of any GroupIterator chunking + Iterator countingIterator = new Iterator<>() { + @Override + public boolean hasNext() { + return rawIterator.hasNext(); + } + + @Override + public Object next() { + rawItemCount.incrementAndGet(); + return rawIterator.next(); + } + }; + + // wrap with GroupIterator if group > 0 to chunk items into List batches + this.iterator = group > 0 ? new GroupIterator(countingIterator, group) : countingIterator; this.copy = copyAndPrepareSubExchange(exchange); this.route = ExchangeHelper.getRoute(exchange); @@ -197,6 +375,14 @@ public boolean hasNext() { if (!answer) { // we are now closed closed = true; + // store raw item count for watermark tracking (guard against + // re-iteration in MulticastProcessor.doDone which re-creates + // the iterator to release exchanges). + // Use rawItemCount (individual items) not index (which counts chunks when group > 0) + if (resumeStrategy != null && watermarkKey != null && watermarkExpression == null + && original.getProperty(SPLIT_WATERMARK_COUNT) == null) { + original.setProperty(SPLIT_WATERMARK_COUNT, rawItemCount.get()); + } // nothing more so we need to close the expression value in case it needs to be try { close(); @@ -240,6 +426,12 @@ public ProcessorExchangePair next() { Message in = newExchange.getIn(); in.setBody(part); } + // track total items for SplitResult (works in both streaming and non-streaming mode) + SplitFailureTracker tracker + = original.getProperty(SPLIT_FAILURE_TRACKER, SplitFailureTracker.class); + if (tracker != null) { + tracker.incrementTotalItems(); + } return createProcessorExchangePair(index++, processor, newExchange, route); } else { return null; @@ -310,6 +502,230 @@ public Expression getExpression() { return expression; } + public int getGroup() { + return group; + } + + public void setGroup(int group) { + this.group = group; + } + + public double getErrorThreshold() { + return errorThreshold; + } + + public void setErrorThreshold(double errorThreshold) { + this.errorThreshold = errorThreshold; + } + + public int getMaxFailedRecords() { + return maxFailedRecords; + } + + public void setMaxFailedRecords(int maxFailedRecords) { + this.maxFailedRecords = maxFailedRecords; + } + + public ResumeStrategy getResumeStrategy() { + return resumeStrategy; + } + + public void setResumeStrategy(ResumeStrategy resumeStrategy) { + this.resumeStrategy = resumeStrategy; + } + + public String getWatermarkKey() { + return watermarkKey; + } + + public void setWatermarkKey(String watermarkKey) { + this.watermarkKey = watermarkKey; + } + + public Expression getWatermarkExpression() { + return watermarkExpression; + } + + public void setWatermarkExpression(Expression watermarkExpression) { + this.watermarkExpression = watermarkExpression; + } + + @Override + protected boolean shouldContinueOnFailure(Exchange subExchange, Exchange original, int index) { + SplitFailureTracker tracker = original.getProperty(SPLIT_FAILURE_TRACKER, SplitFailureTracker.class); + if (tracker == null) { + return super.shouldContinueOnFailure(subExchange, original, index); + } + + // record the failure + tracker.recordFailure(index, subExchange.getException()); + + // check if we've exceeded the max failed records + if (maxFailedRecords > 0 && tracker.getFailureCount() >= maxFailedRecords) { + return false; + } + // check if we've exceeded the error ratio threshold + if (errorThreshold > 0) { + double ratio = (double) tracker.getFailureCount() / (index + 1); + if (ratio >= errorThreshold) { + return false; + } + } + + // Continue processing — clear the exception from the sub-exchange so that aggregation + // proceeds normally. The failure is already recorded in the tracker above, so the + // SplitResult will still contain the failure details even though the exception is cleared. + subExchange.setException(null); + return true; + } + + static final class SplitFailureTracker { + private final AtomicInteger failureCount = new AtomicInteger(); + private final AtomicInteger totalItems = new AtomicInteger(); + private final CopyOnWriteArrayList failures = new CopyOnWriteArrayList<>(); + + void recordFailure(int index, Exception exception) { + failureCount.incrementAndGet(); + failures.add(new SplitResult.Failure(index, exception)); + } + + void incrementTotalItems() { + totalItems.incrementAndGet(); + } + + int getTotalItems() { + return totalItems.get(); + } + + int getFailureCount() { + return failureCount.get(); + } + + List getFailures() { + return List.copyOf(failures); + } + } + + /** + * Wraps a {@link ProcessorExchangePair} to evaluate the watermark expression on each completed sub-exchange. + */ + private final class WatermarkProcessorExchangePair implements ProcessorExchangePair { + private final ProcessorExchangePair delegate; + private final Exchange original; + + WatermarkProcessorExchangePair(ProcessorExchangePair delegate, Exchange original) { + this.delegate = delegate; + this.original = original; + } + + @Override + public int getIndex() { + return delegate.getIndex(); + } + + @Override + public Exchange getExchange() { + return delegate.getExchange(); + } + + @Override + public org.apache.camel.Producer getProducer() { + return delegate.getProducer(); + } + + @Override + public Processor getProcessor() { + return delegate.getProcessor(); + } + + @Override + public void begin() { + delegate.begin(); + } + + @Override + @SuppressWarnings("unchecked") + public void done() { + delegate.done(); + // evaluate watermark expression on completed sub-exchange (only if no exception) + Exchange subExchange = delegate.getExchange(); + if (subExchange.getException() == null) { + String value = watermarkExpression.evaluate(subExchange, String.class); + if (value != null) { + int itemIndex = delegate.getIndex(); + // Use accumulateAndGet to keep the value from the highest-indexed item. + // This ensures deterministic behavior in parallel processing mode. + AtomicReference latestRef + = (AtomicReference) original.getProperty(SPLIT_WATERMARK_LATEST); + latestRef.accumulateAndGet( + new IndexedWatermark(itemIndex, value), + (existing, candidate) -> existing == null || candidate.index() > existing.index() + ? candidate + : existing); + } + } + } + } + + @SuppressWarnings("unchecked") + private void updateWatermark(Exchange exchange) { + try { + // don't update watermark if processing was aborted (allows retry) + if (exchange.getException() != null) { + return; + } + + if (watermarkExpression != null) { + // value-based: use the value from the highest-indexed item tracked during processing + AtomicReference latestRef + = exchange.getProperty(SPLIT_WATERMARK_LATEST, AtomicReference.class); + if (latestRef != null) { + IndexedWatermark latest = latestRef.get(); + if (latest != null) { + persistWatermark(latest.value()); + } + } + } else { + // index-based: compute absolute last index and store it + int offset = exchange.getProperty(SPLIT_WATERMARK_OFFSET, 0, Integer.class); + Integer count = exchange.getProperty(SPLIT_WATERMARK_COUNT, Integer.class); + if (count != null && count > 0) { + int lastAbsoluteIndex = offset + count - 1; + persistWatermark(String.valueOf(lastAbsoluteIndex)); + } + } + } finally { + // always clean up internal properties, even on abort + exchange.removeProperty(SPLIT_WATERMARK_OFFSET); + exchange.removeProperty(SPLIT_WATERMARK_COUNT); + exchange.removeProperty(SPLIT_WATERMARK_LATEST); + } + } + + private void persistWatermark(String value) { + watermarkCache.put(watermarkKey, value); + try { + resumeStrategy.updateLastOffset(OffsetKeys.of(watermarkKey), Offsets.of(value)); + } catch (Exception e) { + throw RuntimeCamelException.wrapRuntimeCamelException(e); + } + } + + private void buildSplitResult(Exchange exchange) { + SplitFailureTracker tracker = exchange.getProperty(SPLIT_FAILURE_TRACKER, SplitFailureTracker.class); + if (tracker == null) { + return; + } + + boolean aborted = exchange.getException() != null; + SplitResult result + = new SplitResult(tracker.getTotalItems(), tracker.getFailureCount(), tracker.getFailures(), aborted); + exchange.setProperty(ExchangePropertyKey.SPLIT_RESULT, result); + + // remove internal tracker + exchange.removeProperty(SPLIT_FAILURE_TRACKER); + } + private Exchange copyAndPrepareSubExchange(Exchange exchange) { Exchange answer = processorExchangeFactory.createCopy(exchange); // must preserve exchange id diff --git a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SplitReifier.java b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SplitReifier.java index 77d0fb583dfbf..19b58ffec9f90 100644 --- a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SplitReifier.java +++ b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SplitReifier.java @@ -30,6 +30,7 @@ import org.apache.camel.processor.aggregate.AggregationStrategyBeanAdapter; import org.apache.camel.processor.aggregate.AggregationStrategyBiFunctionAdapter; import org.apache.camel.processor.aggregate.ShareUnitOfWorkAggregationStrategy; +import org.apache.camel.resume.ResumeStrategy; public class SplitReifier extends ExpressionReifier { @@ -78,9 +79,75 @@ public Processor createProcessor() throws Exception { } answer.setSynchronous(isSynchronous); answer.setDisabled(isDisabled(camelContext, definition)); + + int group = parseInt(definition.getGroup(), 0); + if (group < 0) { + throw new IllegalArgumentException( + "group must not be negative, but was: " + group); + } + if (group > 0) { + answer.setGroup(group); + } + + configureErrorThreshold(answer, isStopOnException); + configureWatermark(answer); + return answer; } + private void configureErrorThreshold(Splitter answer, boolean isStopOnException) { + String etStr = parseString(definition.getErrorThreshold()); + double errorThreshold = etStr != null ? Double.parseDouble(etStr) : 0; + int maxFailedRecords = parseInt(definition.getMaxFailedRecords(), 0); + boolean hasErrorThreshold = errorThreshold > 0 || maxFailedRecords > 0; + if (hasErrorThreshold && isStopOnException) { + throw new IllegalArgumentException( + "Cannot use both stopOnException and errorThreshold/maxFailedRecords on the Splitter EIP"); + } + if (errorThreshold != 0 && (errorThreshold < 0 || errorThreshold > 1.0)) { + throw new IllegalArgumentException( + "errorThreshold must be between 0.0 and 1.0, but was: " + errorThreshold); + } + if (maxFailedRecords < 0) { + throw new IllegalArgumentException( + "maxFailedRecords must not be negative, but was: " + maxFailedRecords); + } + if (errorThreshold > 0) { + answer.setErrorThreshold(errorThreshold); + } + if (maxFailedRecords > 0) { + answer.setMaxFailedRecords(maxFailedRecords); + } + } + + private void configureWatermark(Splitter answer) { + ResumeStrategy resumeStrategy = definition.getResumeStrategyBean(); + if (resumeStrategy == null && definition.getResumeStrategy() != null) { + resumeStrategy = mandatoryLookup(definition.getResumeStrategy(), ResumeStrategy.class); + } + if (resumeStrategy != null) { + CamelContextAware.trySetCamelContext(resumeStrategy, camelContext); + } + String watermarkKey = parseString(definition.getWatermarkKey()); + String watermarkExprStr = parseString(definition.getWatermarkExpression()); + if ((resumeStrategy != null) != (watermarkKey != null)) { + throw new IllegalArgumentException( + "Both resumeStrategy and watermarkKey must be configured together on the Splitter EIP"); + } + if (watermarkExprStr != null && resumeStrategy == null) { + throw new IllegalArgumentException( + "watermarkExpression requires resumeStrategy and watermarkKey on the Splitter EIP"); + } + if (resumeStrategy != null && watermarkKey != null) { + answer.setResumeStrategy(resumeStrategy); + answer.setWatermarkKey(watermarkKey); + if (watermarkExprStr != null) { + Expression watermarkExpr = camelContext.resolveLanguage("simple").createExpression(watermarkExprStr); + answer.setWatermarkExpression(watermarkExpr); + } + } + } + private AggregationStrategy createAggregationStrategy() { AggregationStrategy strategy = definition.getAggregationStrategyBean(); if (strategy == null && definition.getAggregationStrategy() != null) { diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterErrorThresholdTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterErrorThresholdTest.java new file mode 100644 index 0000000000000..48da1eec0dede --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterErrorThresholdTest.java @@ -0,0 +1,238 @@ +/* + * 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.processor; + +import java.util.Arrays; +import java.util.Collections; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; +import org.apache.camel.SplitResult; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class SplitterErrorThresholdTest extends ContextTestSupport { + + @Test + void testErrorThresholdStopsWhenRatioExceeded() throws Exception { + // items: FAIL, FAIL, a, b, c + // errorThreshold=0.5 (50%) + // item 0: FAIL → failCount=1, ratio=1/1=100% ≥ 50% → stop + MockEndpoint mock = getMockEndpoint("mock:split"); + mock.expectedMinimumMessageCount(0); + + Exchange result = template.send("direct:start", + e -> e.getIn().setBody(Arrays.asList("FAIL", "FAIL", "a", "b", "c"))); + + mock.assertIsSatisfied(); + + assertNotNull(result.getException(), "Should have an exception when error threshold exceeded"); + } + + @Test + void testErrorThresholdBelowRatio() throws Exception { + // items: a, b, FAIL, c, d, e, f, g, h, i + // errorThreshold=0.5 (50%) + // item 0: a → ok + // item 1: b → ok + // item 2: FAIL → failCount=1, ratio=1/3=33% < 50% → continue + // items 3-9: ok + MockEndpoint mock = getMockEndpoint("mock:split"); + mock.expectedMessageCount(9); // 10 items minus 1 failure + + template.sendBody("direct:start", Arrays.asList("a", "b", "FAIL", "c", "d", "e", "f", "g", "h", "i")); + + mock.assertIsSatisfied(); + } + + @Test + void testErrorThresholdAllSucceed() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:split"); + mock.expectedMessageCount(5); + + template.sendBody("direct:start", Arrays.asList("a", "b", "c", "d", "e")); + + mock.assertIsSatisfied(); + } + + @Test + void testStopOnExceptionAndErrorThresholdAreMutuallyExclusive() throws Exception { + try { + context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from("direct:invalid") + .split(body()).stopOnException().errorThreshold(0.5) + .to("mock:invalid"); + } + }); + throw new AssertionError("Expected IllegalArgumentException"); + } catch (Exception e) { + assertTrue(e.getCause() instanceof IllegalArgumentException + || e instanceof IllegalArgumentException, + "Should throw IllegalArgumentException"); + } + } + + @Test + void testErrorThresholdWithParallelProcessing() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:parallel-split"); + mock.expectedMinimumMessageCount(0); + + Exchange result = template.send("direct:parallel", + e -> e.getIn().setBody(Arrays.asList("FAIL", "FAIL", "a", "b", "c"))); + + mock.assertIsSatisfied(); + + assertNotNull(result.getException(), "Should have an exception when error threshold exceeded in parallel mode"); + } + + @Test + void testCombinedErrorThresholdAndMaxFailedRecords() throws Exception { + // maxFailedRecords=10 (very high), errorThreshold=0.3 (30%) + // First item FAIL: ratio=1/1=100% >= 30% -> stop due to errorThreshold + MockEndpoint mock = getMockEndpoint("mock:combined"); + mock.expectedMinimumMessageCount(0); + + Exchange result = template.send("direct:combined", + e -> e.getIn().setBody(Arrays.asList("FAIL", "a", "b", "c"))); + + mock.assertIsSatisfied(); + + assertNotNull(result.getException(), "Should stop when errorThreshold exceeded even with high maxFailedRecords"); + } + + @Test + void testErrorThresholdEmptyInput() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:split"); + mock.expectedMessageCount(0); + + Exchange result = template.send("direct:start", + e -> e.getIn().setBody(Collections.emptyList())); + + mock.assertIsSatisfied(); + + // SplitResult should be set even with empty input + SplitResult splitResult = result.getProperty(Exchange.SPLIT_RESULT, SplitResult.class); + assertNotNull(splitResult); + assertEquals(0, splitResult.getTotalItems()); + assertEquals(0, splitResult.getFailureCount()); + assertFalse(splitResult.isAborted()); + } + + @Test + void testErrorThresholdExactBoundary() throws Exception { + // items: a, FAIL, b, c — errorThreshold=0.5 + // item 0: a ok + // item 1: FAIL → failCount=1, ratio=1/2=50% ≥ 50% → stop + MockEndpoint mock = getMockEndpoint("mock:split"); + mock.expectedMinimumMessageCount(0); + + Exchange result = template.send("direct:start", + e -> e.getIn().setBody(Arrays.asList("a", "FAIL", "b", "c"))); + + mock.assertIsSatisfied(); + + assertNotNull(result.getException(), "Should stop at exact boundary (ratio == threshold)"); + } + + @Test + void testErrorThresholdJustBelowBoundary() throws Exception { + // items: a, b, FAIL, d — errorThreshold=0.5 + // item 0: a ok, item 1: b ok + // item 2: FAIL → failCount=1, ratio=1/3=33% < 50% → continue + // item 3: d ok + MockEndpoint mock = getMockEndpoint("mock:split"); + mock.expectedMessageCount(3); // a, b, d succeed + + Exchange result = template.send("direct:start", + e -> e.getIn().setBody(Arrays.asList("a", "b", "FAIL", "d"))); + + mock.assertIsSatisfied(); + assertNull(result.getException(), "Should not stop when ratio is below threshold"); + } + + @Test + void testMaxFailedRecordsExactBoundary() throws Exception { + // maxFailedRecords=2, items: a, FAIL, b, FAIL, c + // fail #1 at index 1 (count=1 < 2 → continue) + // fail #2 at index 3 (count=2 >= 2 → stop) + MockEndpoint mock = getMockEndpoint("mock:maxfail"); + mock.expectedMinimumMessageCount(2); // at least a, b + + Exchange result = template.send("direct:maxfail", + e -> e.getIn().setBody(Arrays.asList("a", "FAIL", "b", "FAIL", "c"))); + + mock.assertIsSatisfied(); + assertNotNull(result.getException(), "Should stop at exact maxFailedRecords boundary"); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:start") + .split(body()).errorThreshold(0.5) + .process(exchange -> { + String body = exchange.getIn().getBody(String.class); + if ("FAIL".equals(body)) { + throw new IllegalArgumentException("Simulated failure for: " + body); + } + }) + .to("mock:split"); + + from("direct:parallel") + .split(body()).errorThreshold(0.5).parallelProcessing() + .process(exchange -> { + String body = exchange.getIn().getBody(String.class); + if ("FAIL".equals(body)) { + throw new IllegalArgumentException("Simulated failure for: " + body); + } + }) + .to("mock:parallel-split"); + + from("direct:combined") + .split(body()).maxFailedRecords(10).errorThreshold(0.3) + .process(exchange -> { + String body = exchange.getIn().getBody(String.class); + if ("FAIL".equals(body)) { + throw new IllegalArgumentException("Simulated failure for: " + body); + } + }) + .to("mock:combined"); + + from("direct:maxfail") + .split(body()).maxFailedRecords(2) + .process(exchange -> { + String body = exchange.getIn().getBody(String.class); + if ("FAIL".equals(body)) { + throw new IllegalArgumentException("Simulated failure for: " + body); + } + }) + .to("mock:maxfail"); + } + }; + } +} diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterGroupTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterGroupTest.java new file mode 100644 index 0000000000000..afedb2b5c5bf9 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterGroupTest.java @@ -0,0 +1,144 @@ +/* + * 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.processor; + +import java.util.Arrays; +import java.util.List; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; +import org.apache.camel.SplitResult; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +class SplitterGroupTest extends ContextTestSupport { + + @Test + void testSplitterGroup() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:split"); + mock.expectedMessageCount(3); + + template.sendBody("direct:start", Arrays.asList("a", "b", "c", "d", "e", "f", "g")); + + mock.assertIsSatisfied(); + + // First group: [a, b, c] + Object body0 = mock.getReceivedExchanges().get(0).getIn().getBody(); + assertInstanceOf(List.class, body0); + assertEquals(3, ((List) body0).size()); + + // Second group: [d, e, f] + Object body1 = mock.getReceivedExchanges().get(1).getIn().getBody(); + assertInstanceOf(List.class, body1); + assertEquals(3, ((List) body1).size()); + + // Third group: [g] (remainder) + Object body2 = mock.getReceivedExchanges().get(2).getIn().getBody(); + assertInstanceOf(List.class, body2); + assertEquals(1, ((List) body2).size()); + } + + @Test + void testSplitterGroupExactMultiple() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:split"); + mock.expectedMessageCount(2); + + template.sendBody("direct:start", Arrays.asList("a", "b", "c", "d", "e", "f")); + + mock.assertIsSatisfied(); + + assertEquals(3, ((List) mock.getReceivedExchanges().get(0).getIn().getBody()).size()); + assertEquals(3, ((List) mock.getReceivedExchanges().get(1).getIn().getBody()).size()); + } + + @Test + void testSplitterGroupSingleItem() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:split"); + mock.expectedMessageCount(1); + + template.sendBody("direct:start", List.of("only")); + + mock.assertIsSatisfied(); + + List body = (List) mock.getReceivedExchanges().get(0).getIn().getBody(); + assertEquals(1, body.size()); + assertEquals("only", body.get(0)); + } + + @Test + void testSplitterGroupWithParallelProcessing() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:parallel-split"); + mock.expectedMessageCount(3); + + template.sendBody("direct:parallel", Arrays.asList(1, 2, 3, 4, 5, 6, 7)); + + mock.assertIsSatisfied(); + } + + @Test + void testGroupWithMaxFailedRecords() throws Exception { + // 9 items grouped by 3. Processor throws if the chunk contains "FAIL". + // Items: a, b, FAIL, d, e, f, g, h, i → chunks [a,b,FAIL], [d,e,f], [g,h,i] + // First chunk triggers failure, maxFailedRecords=2, so processing continues + MockEndpoint mock = getMockEndpoint("mock:group-fail"); + mock.expectedMinimumMessageCount(2); // at least the 2 non-failing chunks + + Exchange result = template.send("direct:group-fail", + e -> e.getIn().setBody(Arrays.asList("a", "b", "FAIL", "d", "e", "f", "g", "h", "i"))); + + mock.assertIsSatisfied(); + + SplitResult splitResult = result.getProperty(Exchange.SPLIT_RESULT, SplitResult.class); + assertNotNull(splitResult); + assertEquals(1, splitResult.getFailureCount()); + assertFalse(splitResult.isAborted()); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:start") + .split(body()).group(3) + .to("mock:split"); + + from("direct:parallel") + .split(body()).group(3).parallelProcessing() + .to("mock:parallel-split"); + + from("direct:group-fail") + .split(body()).group(3).maxFailedRecords(2) + .process(exchange -> { + List chunk = exchange.getIn().getBody(List.class); + for (Object item : chunk) { + if ("FAIL".equals(item)) { + throw new IllegalArgumentException("Chunk contains FAIL"); + } + } + }) + .to("mock:group-fail"); + } + }; + } +} diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterMaxFailedRecordsTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterMaxFailedRecordsTest.java new file mode 100644 index 0000000000000..6c8f63f806374 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterMaxFailedRecordsTest.java @@ -0,0 +1,136 @@ +/* + * 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.processor; + +import java.util.Arrays; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class SplitterMaxFailedRecordsTest extends ContextTestSupport { + + @Test + void testMaxFailedRecordsStopsAfterThreshold() throws Exception { + // items: a, FAIL, b, FAIL, c, FAIL, d + // maxFailedRecords=2, so processing should stop after the 2nd failure + MockEndpoint mock = getMockEndpoint("mock:split"); + // With maxFailedRecords=2: items a (ok), FAIL (fail #1, continue), b (ok), FAIL (fail #2, stop) + // Items c, FAIL, d should NOT be processed + mock.expectedMinimumMessageCount(2); // at least a, b + + Exchange result = template.send("direct:start", + e -> e.getIn().setBody(Arrays.asList("a", "FAIL", "b", "FAIL", "c", "FAIL", "d"))); + + mock.assertIsSatisfied(); + + // the exchange should have an exception because threshold was exceeded + assertNotNull(result.getException(), "Should have an exception when max failed records exceeded"); + + // verify that not all items were processed + assertTrue(mock.getReceivedCounter() < 7, "Should have stopped before processing all items"); + } + + @Test + void testMaxFailedRecordsAllSucceed() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:split"); + mock.expectedMessageCount(4); + + template.sendBody("direct:start", Arrays.asList("a", "b", "c", "d")); + + mock.assertIsSatisfied(); + } + + @Test + void testMaxFailedRecordsSingleFailure() throws Exception { + // maxFailedRecords=2, only 1 failure, should process all items + // only 3 items reach mock:split (the failed one throws before reaching the endpoint) + MockEndpoint mock = getMockEndpoint("mock:split"); + mock.expectedMessageCount(3); + + template.sendBody("direct:start", Arrays.asList("a", "FAIL", "b", "c")); + + mock.assertIsSatisfied(); + } + + @Test + void testStopOnExceptionAndMaxFailedRecordsAreMutuallyExclusive() throws Exception { + try { + context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from("direct:invalid") + .split(body()).stopOnException().maxFailedRecords(3) + .to("mock:invalid"); + } + }); + // should not reach here + throw new AssertionError("Expected IllegalArgumentException"); + } catch (Exception e) { + assertTrue(e.getCause() instanceof IllegalArgumentException + || e instanceof IllegalArgumentException, + "Should throw IllegalArgumentException"); + } + } + + @Test + void testMaxFailedRecordsWithParallelProcessing() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:parallel"); + mock.expectedMinimumMessageCount(0); + + Exchange result = template.send("direct:parallel", + e -> e.getIn().setBody(Arrays.asList("a", "FAIL", "b", "FAIL", "c", "FAIL", "d"))); + + mock.assertIsSatisfied(); + + assertNotNull(result.getException(), "Should have exception in parallel mode"); + assertTrue(mock.getReceivedCounter() < 7, "Should stop before processing all items in parallel mode"); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:start") + .split(body()).maxFailedRecords(2) + .process(exchange -> { + String body = exchange.getIn().getBody(String.class); + if ("FAIL".equals(body)) { + throw new IllegalArgumentException("Simulated failure for: " + body); + } + }) + .to("mock:split"); + + from("direct:parallel") + .split(body()).maxFailedRecords(2).parallelProcessing() + .process(exchange -> { + String body = exchange.getIn().getBody(String.class); + if ("FAIL".equals(body)) { + throw new IllegalArgumentException("Simulated failure for: " + body); + } + }) + .to("mock:parallel"); + } + }; + } +} diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterParallelErrorThresholdTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterParallelErrorThresholdTest.java new file mode 100644 index 0000000000000..3efcf15f5c1dc --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterParallelErrorThresholdTest.java @@ -0,0 +1,137 @@ +/* + * 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.processor; + +import java.util.ArrayList; +import java.util.List; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; +import org.apache.camel.SplitResult; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests for error threshold features with parallel processing. + */ +class SplitterParallelErrorThresholdTest extends ContextTestSupport { + + @Test + void testMaxFailedRecordsWithParallel() throws Exception { + // 20 items, every 4th fails (indices 3, 7, 11, 15, 19) + // maxFailedRecords=3 → should abort after 3rd failure + List items = new ArrayList<>(); + for (int i = 0; i < 20; i++) { + items.add(i % 4 == 3 ? "FAIL" : "item-" + i); + } + + MockEndpoint mock = getMockEndpoint("mock:parallel-max"); + mock.expectedMinimumMessageCount(0); + + Exchange result = template.send("direct:parallel-max", + e -> e.getIn().setBody(items)); + + mock.assertIsSatisfied(); + + assertNotNull(result.getException(), "Should have exception when maxFailedRecords exceeded"); + + SplitResult splitResult = result.getProperty(Exchange.SPLIT_RESULT, SplitResult.class); + assertNotNull(splitResult, "SplitResult should be set"); + assertTrue(splitResult.isAborted()); + assertTrue(splitResult.getFailureCount() >= 3, "Should have at least 3 failures"); + } + + @Test + void testMaxFailedRecordsAllSucceedWithParallel() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:parallel-max"); + mock.expectedMessageCount(10); + + List items = new ArrayList<>(); + for (int i = 0; i < 10; i++) { + items.add("item-" + i); + } + + Exchange result = template.send("direct:parallel-max", + e -> e.getIn().setBody(items)); + + mock.assertIsSatisfied(); + + SplitResult splitResult = result.getProperty(Exchange.SPLIT_RESULT, SplitResult.class); + assertNotNull(splitResult); + assertEquals(10, splitResult.getTotalItems()); + assertEquals(0, splitResult.getFailureCount()); + assertFalse(splitResult.isAborted()); + } + + @Test + void testErrorThresholdWithParallelHighFailureRate() throws Exception { + // 10 items, first 8 fail → very high failure rate + // errorThreshold=0.5 → should abort early + List items = new ArrayList<>(); + for (int i = 0; i < 10; i++) { + items.add(i < 8 ? "FAIL" : "item-" + i); + } + + MockEndpoint mock = getMockEndpoint("mock:parallel-threshold"); + mock.expectedMinimumMessageCount(0); + + Exchange result = template.send("direct:parallel-threshold", + e -> e.getIn().setBody(items)); + + mock.assertIsSatisfied(); + + assertNotNull(result.getException(), "Should abort with high failure rate"); + + SplitResult splitResult = result.getProperty(Exchange.SPLIT_RESULT, SplitResult.class); + assertNotNull(splitResult); + assertTrue(splitResult.isAborted()); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:parallel-max") + .split(body()).parallelProcessing().maxFailedRecords(3) + .process(exchange -> { + String body = exchange.getIn().getBody(String.class); + if ("FAIL".equals(body)) { + throw new IllegalArgumentException("Simulated failure: " + body); + } + }) + .to("mock:parallel-max"); + + from("direct:parallel-threshold") + .split(body()).parallelProcessing().errorThreshold(0.5) + .process(exchange -> { + String body = exchange.getIn().getBody(String.class); + if ("FAIL".equals(body)) { + throw new IllegalArgumentException("Simulated failure: " + body); + } + }) + .to("mock:parallel-threshold"); + } + }; + } +} diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterSplitResultTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterSplitResultTest.java new file mode 100644 index 0000000000000..d6603f6f30219 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterSplitResultTest.java @@ -0,0 +1,230 @@ +/* + * 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.processor; + +import java.util.Arrays; +import java.util.Collections; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; +import org.apache.camel.SplitResult; +import org.apache.camel.builder.RouteBuilder; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class SplitterSplitResultTest extends ContextTestSupport { + + @Test + void testSplitResultWithFailures() throws Exception { + // items: a, FAIL, b, FAIL, c — maxFailedRecords=5 so all items are processed + Exchange result = template.send("direct:tolerant", + e -> e.getIn().setBody(Arrays.asList("a", "FAIL", "b", "FAIL", "c"))); + + SplitResult splitResult = result.getProperty(Exchange.SPLIT_RESULT, SplitResult.class); + assertNotNull(splitResult, "SplitResult should be set as exchange property"); + + assertEquals(5, splitResult.getTotalItems()); + assertEquals(2, splitResult.getFailureCount()); + assertEquals(3, splitResult.getSuccessCount()); + assertFalse(splitResult.isAborted()); + + // verify failure details + assertEquals(2, splitResult.getFailures().size()); + assertEquals(1, splitResult.getFailures().get(0).index()); + assertEquals(3, splitResult.getFailures().get(1).index()); + assertNotNull(splitResult.getFailures().get(0).exception()); + } + + @Test + void testSplitResultWhenAborted() throws Exception { + // items: FAIL, FAIL, a, b — maxFailedRecords=2 so aborted after 2nd failure + Exchange result = template.send("direct:strict", + e -> e.getIn().setBody(Arrays.asList("FAIL", "FAIL", "a", "b"))); + + SplitResult splitResult = result.getProperty(Exchange.SPLIT_RESULT, SplitResult.class); + assertNotNull(splitResult, "SplitResult should be set even when aborted"); + assertTrue(splitResult.isAborted()); + assertEquals(2, splitResult.getFailureCount()); + } + + @Test + void testSplitResultAllSuccess() throws Exception { + Exchange result = template.send("direct:tolerant", + e -> e.getIn().setBody(Arrays.asList("a", "b", "c"))); + + SplitResult splitResult = result.getProperty(Exchange.SPLIT_RESULT, SplitResult.class); + assertNotNull(splitResult, "SplitResult should be set even with no failures"); + + assertEquals(3, splitResult.getTotalItems()); + assertEquals(0, splitResult.getFailureCount()); + assertEquals(3, splitResult.getSuccessCount()); + assertFalse(splitResult.isAborted()); + assertTrue(splitResult.getFailures().isEmpty()); + } + + @Test + void testSplitResultInStreamingMode() throws Exception { + Exchange result = template.send("direct:streaming", + e -> e.getIn().setBody(Arrays.asList("a", "FAIL", "b", "c", "d"))); + + SplitResult splitResult = result.getProperty(Exchange.SPLIT_RESULT, SplitResult.class); + assertNotNull(splitResult, "SplitResult should be set in streaming mode"); + assertEquals(5, splitResult.getTotalItems()); + assertEquals(1, splitResult.getFailureCount()); + assertEquals(4, splitResult.getSuccessCount()); + assertFalse(splitResult.isAborted()); + } + + @Test + void testSplitResultWithParallelProcessing() throws Exception { + Exchange result = template.send("direct:parallel", + e -> e.getIn().setBody(Arrays.asList("a", "FAIL", "b", "c"))); + + SplitResult splitResult = result.getProperty(Exchange.SPLIT_RESULT, SplitResult.class); + assertNotNull(splitResult, "SplitResult should be set in parallel mode"); + assertEquals(4, splitResult.getTotalItems()); + assertEquals(1, splitResult.getFailureCount()); + assertEquals(3, splitResult.getSuccessCount()); + assertFalse(splitResult.isAborted()); + } + + @Test + void testSplitResultToString() throws Exception { + Exchange result = template.send("direct:tolerant", + e -> e.getIn().setBody(Arrays.asList("a", "FAIL", "b", "c"))); + + SplitResult splitResult = result.getProperty(Exchange.SPLIT_RESULT, SplitResult.class); + assertNotNull(splitResult); + + String str = splitResult.toString(); + assertTrue(str.contains("total=4"), "toString should contain total: " + str); + assertTrue(str.contains("success=3"), "toString should contain success: " + str); + assertTrue(str.contains("failures=1"), "toString should contain failures: " + str); + assertTrue(str.contains("aborted=false"), "toString should contain aborted: " + str); + } + + @Test + void testSplitResultToStringWhenAborted() throws Exception { + Exchange result = template.send("direct:strict", + e -> e.getIn().setBody(Arrays.asList("FAIL", "FAIL", "a"))); + + SplitResult splitResult = result.getProperty(Exchange.SPLIT_RESULT, SplitResult.class); + assertNotNull(splitResult); + + String str = splitResult.toString(); + assertTrue(str.contains("aborted=true"), "toString should contain aborted=true: " + str); + } + + @Test + void testSplitResultConstructorWithNullFailures() { + SplitResult result = new SplitResult(5, 0, null, false); + assertEquals(5, result.getTotalItems()); + assertEquals(0, result.getFailureCount()); + assertEquals(5, result.getSuccessCount()); + assertFalse(result.isAborted()); + assertNotNull(result.getFailures()); + assertTrue(result.getFailures().isEmpty()); + } + + @Test + void testSplitResultEmptyInput() throws Exception { + Exchange result = template.send("direct:tolerant", + e -> e.getIn().setBody(Collections.emptyList())); + + SplitResult splitResult = result.getProperty(Exchange.SPLIT_RESULT, SplitResult.class); + assertNotNull(splitResult, "SplitResult should be set even with empty input"); + assertEquals(0, splitResult.getTotalItems()); + assertEquals(0, splitResult.getFailureCount()); + assertEquals(0, splitResult.getSuccessCount()); + assertFalse(splitResult.isAborted()); + } + + @Test + void testSplitResultFailureRecord() { + Exception cause = new IllegalArgumentException("test error"); + SplitResult.Failure failure = new SplitResult.Failure(3, cause); + assertEquals(3, failure.index()); + assertEquals(cause, failure.exception()); + } + + @Test + void testNoSplitResultWithoutErrorThreshold() throws Exception { + // plain split without error threshold should not set SplitResult + Exchange result = template.send("direct:plain", + e -> e.getIn().setBody(Arrays.asList("a", "b", "c"))); + + SplitResult splitResult = result.getProperty(Exchange.SPLIT_RESULT, SplitResult.class); + assertNull(splitResult, "SplitResult should not be set without error threshold"); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:tolerant") + .split(body()).maxFailedRecords(5) + .process(exchange -> { + String body = exchange.getIn().getBody(String.class); + if ("FAIL".equals(body)) { + throw new IllegalArgumentException("Simulated failure for: " + body); + } + }) + .to("mock:split"); + + from("direct:strict") + .split(body()).maxFailedRecords(2) + .process(exchange -> { + String body = exchange.getIn().getBody(String.class); + if ("FAIL".equals(body)) { + throw new IllegalArgumentException("Simulated failure for: " + body); + } + }) + .to("mock:split"); + + from("direct:plain") + .split(body()) + .to("mock:split"); + + from("direct:streaming") + .split(body()).streaming().maxFailedRecords(10) + .process(exchange -> { + String body = exchange.getIn().getBody(String.class); + if ("FAIL".equals(body)) { + throw new IllegalArgumentException("Simulated failure for: " + body); + } + }) + .to("mock:split"); + + from("direct:parallel") + .split(body()).parallelProcessing().maxFailedRecords(10) + .process(exchange -> { + String body = exchange.getIn().getBody(String.class); + if ("FAIL".equals(body)) { + throw new IllegalArgumentException("Simulated failure for: " + body); + } + }) + .to("mock:split"); + } + }; + } +} diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterStreamingTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterStreamingTest.java new file mode 100644 index 0000000000000..a2b1ad9feaabe --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterStreamingTest.java @@ -0,0 +1,182 @@ +/* + * 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.processor; + +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; +import org.apache.camel.SplitResult; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.resume.ResumeStrategy; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class SplitterStreamingTest extends ContextTestSupport { + + private final Map store = new ConcurrentHashMap<>(); + private final ResumeStrategy strategy = new SplitterTestResumeStrategy(store); + + @Test + void testStreamingSplitResultTotalItems() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:streaming"); + mock.expectedMessageCount(3); + + Exchange result = template.send("direct:streaming", + e -> e.getIn().setBody(Arrays.asList("a", "FAIL", "b", "c"))); + + mock.assertIsSatisfied(); + + SplitResult splitResult = result.getProperty(Exchange.SPLIT_RESULT, SplitResult.class); + assertNotNull(splitResult, "SplitResult should be set in streaming mode"); + assertEquals(4, splitResult.getTotalItems(), "totalItems should reflect all iterated items"); + assertEquals(1, splitResult.getFailureCount()); + assertEquals(3, splitResult.getSuccessCount()); + assertFalse(splitResult.isAborted()); + } + + @Test + void testStreamingSplitResultWhenAborted() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:streaming-abort"); + mock.expectedMinimumMessageCount(0); + + Exchange result = template.send("direct:streaming-abort", + e -> e.getIn().setBody(Arrays.asList("FAIL", "FAIL", "a", "b"))); + + mock.assertIsSatisfied(); + + SplitResult splitResult = result.getProperty(Exchange.SPLIT_RESULT, SplitResult.class); + assertNotNull(splitResult, "SplitResult should be set even when streaming and aborted"); + assertTrue(splitResult.isAborted()); + assertEquals(2, splitResult.getFailureCount()); + // totalItems reflects how many were iterated before abort (at least 2) + assertTrue(splitResult.getTotalItems() >= 2, "totalItems should reflect items iterated before abort"); + } + + @Test + void testStreamingGrouping() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:streaming-group"); + mock.expectedMessageCount(3); + + template.sendBody("direct:streaming-group", Arrays.asList("a", "b", "c", "d", "e", "f", "g")); + + mock.assertIsSatisfied(); + + assertInstanceOf(List.class, mock.getReceivedExchanges().get(0).getIn().getBody()); + assertEquals(3, ((List) mock.getReceivedExchanges().get(0).getIn().getBody()).size()); + assertEquals(1, ((List) mock.getReceivedExchanges().get(2).getIn().getBody()).size()); + } + + @Test + void testStreamingWatermark() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:streaming-wm"); + mock.expectedMessageCount(5); + + template.sendBody("direct:streaming-wm", Arrays.asList("a", "b", "c", "d", "e")); + + mock.assertIsSatisfied(); + + assertEquals("4", store.get("streamJob")); + } + + @Test + void testStreamingWatermarkSecondRun() throws Exception { + store.put("streamJob", "2"); + + MockEndpoint mock = getMockEndpoint("mock:streaming-wm"); + mock.expectedMessageCount(2); + + template.sendBody("direct:streaming-wm", Arrays.asList("a", "b", "c", "d", "e")); + + mock.assertIsSatisfied(); + + assertEquals("d", mock.getReceivedExchanges().get(0).getIn().getBody(String.class)); + assertEquals("e", mock.getReceivedExchanges().get(1).getIn().getBody(String.class)); + assertEquals("4", store.get("streamJob")); + } + + @Test + void testStreamingParallelMaxFailedRecords() throws Exception { + // streaming + parallel + maxFailedRecords combination + MockEndpoint mock = getMockEndpoint("mock:streaming-parallel"); + mock.expectedMinimumMessageCount(0); + + Exchange result = template.send("direct:streaming-parallel", + e -> e.getIn().setBody(Arrays.asList("a", "FAIL", "b", "FAIL", "FAIL", "c"))); + + mock.assertIsSatisfied(); + + assertNotNull(result.getException(), "Should abort after 3 failures"); + + SplitResult splitResult = result.getProperty(Exchange.SPLIT_RESULT, SplitResult.class); + assertNotNull(splitResult); + assertTrue(splitResult.isAborted()); + assertTrue(splitResult.getFailureCount() >= 3); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:streaming") + .split(body()).streaming().maxFailedRecords(10) + .process(exchange -> { + if ("FAIL".equals(exchange.getIn().getBody(String.class))) { + throw new IllegalArgumentException("Simulated"); + } + }) + .to("mock:streaming"); + + from("direct:streaming-abort") + .split(body()).streaming().maxFailedRecords(2) + .process(exchange -> { + if ("FAIL".equals(exchange.getIn().getBody(String.class))) { + throw new IllegalArgumentException("Simulated"); + } + }) + .to("mock:streaming-abort"); + + from("direct:streaming-group") + .split(body()).streaming().group(3) + .to("mock:streaming-group"); + + from("direct:streaming-wm") + .split(body()).streaming().resumeStrategy(strategy, "streamJob") + .to("mock:streaming-wm"); + + from("direct:streaming-parallel") + .split(body()).streaming().parallelProcessing().maxFailedRecords(3) + .process(exchange -> { + if ("FAIL".equals(exchange.getIn().getBody(String.class))) { + throw new IllegalArgumentException("Simulated"); + } + }) + .to("mock:streaming-parallel"); + } + }; + } +} diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterTestResumeStrategy.java b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterTestResumeStrategy.java new file mode 100644 index 0000000000000..5771d8ca91859 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterTestResumeStrategy.java @@ -0,0 +1,173 @@ +/* + * 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.processor; + +import java.util.Map; +import java.util.function.BiFunction; +import java.util.function.Function; + +import org.apache.camel.resume.Offset; +import org.apache.camel.resume.OffsetKey; +import org.apache.camel.resume.Resumable; +import org.apache.camel.resume.ResumeAdapter; +import org.apache.camel.resume.ResumeStrategy; +import org.apache.camel.resume.ResumeStrategyConfiguration; +import org.apache.camel.resume.cache.ResumeCache; + +/** + * A simple in-memory {@link ResumeStrategy} for Splitter watermark tests. Wraps a {@link Map} that tests can + * pre-populate and assert against. The cache is a live view backed by the store, so changes to the store are + * immediately visible. + */ +class SplitterTestResumeStrategy implements ResumeStrategy { + + private final Map store; + private final ResumeStrategyConfiguration config; + + SplitterTestResumeStrategy(Map store) { + this.store = store; + ResumeCache cache = new LiveStoreCache(store); + ResumeStrategyConfiguration cfg = new ResumeStrategyConfiguration() { + @Override + public String resumeStrategyService() { + return "test"; + } + }; + cfg.setResumeCache(cache); + this.config = cfg; + } + + @Override + public void loadCache() { + // no-op: cache is a live view of the store + } + + @Override + public void updateLastOffset(OffsetKey offsetKey, Offset offsetValue) { + store.put(offsetKey.getValue().toString(), offsetValue.getValue().toString()); + } + + @Override + public void updateLastOffset(T offset) { + // no-op + } + + @Override + public void updateLastOffset(T offset, UpdateCallBack updateCallBack) { + // no-op + } + + @Override + public void updateLastOffset(OffsetKey offsetKey, Offset offset, UpdateCallBack updateCallBack) + throws Exception { + updateLastOffset(offsetKey, offset); + } + + @Override + public void setAdapter(ResumeAdapter adapter) { + // no-op + } + + @Override + public ResumeAdapter getAdapter() { + return null; + } + + @Override + public void setResumeStrategyConfiguration(ResumeStrategyConfiguration resumeStrategyConfiguration) { + // no-op + } + + @Override + public ResumeStrategyConfiguration getResumeStrategyConfiguration() { + return config; + } + + @Override + public void start() { + // no-op + } + + @Override + public void stop() { + // no-op + } + + /** + * A {@link ResumeCache} that directly delegates to the backing store map, providing a live view. + */ + private static class LiveStoreCache implements ResumeCache { + private final Map store; + + LiveStoreCache(Map store) { + this.store = store; + } + + @Override + public Object get(Object key) { + return store.get(String.valueOf(key)); + } + + @Override + @SuppressWarnings("unchecked") + public T get(Object key, Class clazz) { + Object value = get(key); + return value != null ? clazz.cast(value) : null; + } + + @Override + public void add(Object key, Object offsetValue) { + store.put(String.valueOf(key), String.valueOf(offsetValue)); + } + + @Override + public boolean contains(Object key, Object entry) { + Object value = get(key); + return value != null && value.equals(entry); + } + + @Override + public Object computeIfAbsent(Object key, Function mapping) { + return store.computeIfAbsent(String.valueOf(key), k -> String.valueOf(mapping.apply(k))); + } + + @Override + public Object computeIfPresent(Object key, BiFunction remapping) { + return store.computeIfPresent(String.valueOf(key), + (k, v) -> String.valueOf(remapping.apply(k, v))); + } + + @Override + public boolean isFull() { + return false; + } + + @Override + public long capacity() { + return Integer.MAX_VALUE; + } + + @Override + public void forEach(BiFunction action) { + for (Map.Entry e : store.entrySet()) { + if (!action.apply(e.getKey(), e.getValue())) { + break; + } + } + } + } +} diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterTransactedTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterTransactedTest.java new file mode 100644 index 0000000000000..8f5dab810e236 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterTransactedTest.java @@ -0,0 +1,213 @@ +/* + * 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.processor; + +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; +import org.apache.camel.SplitResult; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.resume.ResumeStrategy; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests for Splitter EIP enhancements (group, error threshold, watermark) using the transacted code path + * ({@link MulticastProcessor.MulticastTransactedTask}). + */ +class SplitterTransactedTest extends ContextTestSupport { + + private final Map store = new ConcurrentHashMap<>(); + private final ResumeStrategy strategy = new SplitterTestResumeStrategy(store); + + @Test + void testGroupTransacted() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:group"); + mock.expectedMessageCount(3); + + Exchange exchange = createExchangeWithBody(Arrays.asList("a", "b", "c", "d", "e", "f", "g")); + exchange.getExchangeExtension().setTransacted(true); + + template.send("direct:group", exchange); + + mock.assertIsSatisfied(); + + // verify chunks + assertEquals(List.of("a", "b", "c"), mock.getReceivedExchanges().get(0).getIn().getBody()); + assertEquals(List.of("d", "e", "f"), mock.getReceivedExchanges().get(1).getIn().getBody()); + assertEquals(List.of("g"), mock.getReceivedExchanges().get(2).getIn().getBody()); + } + + @Test + void testMaxFailedRecordsTransacted() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:maxfail"); + mock.expectedMinimumMessageCount(0); + + Exchange exchange = createExchangeWithBody(Arrays.asList("ok", "FAIL", "ok2", "FAIL2", "ok3")); + exchange.getExchangeExtension().setTransacted(true); + + Exchange result = template.send("direct:maxfail", exchange); + + mock.assertIsSatisfied(); + + SplitResult splitResult = result.getProperty(Exchange.SPLIT_RESULT, SplitResult.class); + assertNotNull(splitResult); + assertEquals(2, splitResult.getFailureCount()); + assertTrue(splitResult.isAborted()); + } + + @Test + void testErrorThresholdTransacted() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:threshold"); + mock.expectedMinimumMessageCount(0); + + // 2 out of 4 fail = 50% error rate, threshold is 0.4 (40%) + Exchange exchange = createExchangeWithBody(Arrays.asList("ok", "FAIL", "FAIL2", "ok2")); + exchange.getExchangeExtension().setTransacted(true); + + Exchange result = template.send("direct:threshold", exchange); + + mock.assertIsSatisfied(); + + SplitResult splitResult = result.getProperty(Exchange.SPLIT_RESULT, SplitResult.class); + assertNotNull(splitResult); + assertTrue(splitResult.isAborted()); + } + + @Test + void testSplitResultTransacted() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:result"); + mock.expectedMessageCount(3); + + Exchange exchange = createExchangeWithBody(Arrays.asList("ok", "FAIL", "ok2", "ok3")); + exchange.getExchangeExtension().setTransacted(true); + + Exchange result = template.send("direct:result", exchange); + + mock.assertIsSatisfied(); + + SplitResult splitResult = result.getProperty(Exchange.SPLIT_RESULT, SplitResult.class); + assertNotNull(splitResult); + assertEquals(4, splitResult.getTotalItems()); + assertEquals(1, splitResult.getFailureCount()); + assertEquals(3, splitResult.getSuccessCount()); + assertFalse(splitResult.isAborted()); + } + + @Test + void testIndexWatermarkTransacted() throws Exception { + store.put("txJob", "1"); + + MockEndpoint mock = getMockEndpoint("mock:watermark"); + mock.expectedMessageCount(3); + + Exchange exchange = createExchangeWithBody(Arrays.asList("a", "b", "c", "d", "e")); + exchange.getExchangeExtension().setTransacted(true); + + Exchange result = template.send("direct:watermark", exchange); + + mock.assertIsSatisfied(); + + // should skip items 0 and 1, process c, d, e + assertEquals("c", mock.getReceivedExchanges().get(0).getIn().getBody(String.class)); + assertEquals("d", mock.getReceivedExchanges().get(1).getIn().getBody(String.class)); + assertEquals("e", mock.getReceivedExchanges().get(2).getIn().getBody(String.class)); + + // watermark updated to 4 + assertEquals("4", store.get("txJob")); + + // previous watermark exposed + assertEquals("1", result.getProperty(Exchange.SPLIT_WATERMARK)); + } + + @Test + void testValueWatermarkTransacted() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:valuewm"); + mock.expectedMessageCount(3); + + Exchange exchange = createExchangeWithBody(Arrays.asList("2024-01-01", "2024-01-02", "2024-01-03")); + exchange.getExchangeExtension().setTransacted(true); + + template.send("direct:valuewm", exchange); + + mock.assertIsSatisfied(); + + // watermark should be the last processed item + assertEquals("2024-01-03", store.get("txDateJob")); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:group") + .split(body()).group(3) + .to("mock:group"); + + from("direct:maxfail") + .split(body()).maxFailedRecords(2) + .process(exchange -> { + String body = exchange.getIn().getBody(String.class); + if (body.startsWith("FAIL")) { + throw new IllegalArgumentException("Simulated failure: " + body); + } + }) + .to("mock:maxfail"); + + from("direct:threshold") + .split(body()).errorThreshold(0.4) + .process(exchange -> { + String body = exchange.getIn().getBody(String.class); + if (body.startsWith("FAIL")) { + throw new IllegalArgumentException("Simulated failure: " + body); + } + }) + .to("mock:threshold"); + + from("direct:result") + .split(body()).maxFailedRecords(10) + .process(exchange -> { + String body = exchange.getIn().getBody(String.class); + if (body.startsWith("FAIL")) { + throw new IllegalArgumentException("Simulated failure: " + body); + } + }) + .to("mock:result"); + + from("direct:watermark") + .split(body()).resumeStrategy(strategy, "txJob") + .to("mock:watermark"); + + from("direct:valuewm") + .split(body()) + .resumeStrategy(strategy, "txDateJob") + .watermarkExpression("${body}") + .to("mock:valuewm"); + } + }; + } +} diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterValidationTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterValidationTest.java new file mode 100644 index 0000000000000..d13ed1c7e2c04 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterValidationTest.java @@ -0,0 +1,119 @@ +/* + * 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.processor; + +import java.util.concurrent.ConcurrentHashMap; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.builder.RouteBuilder; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class SplitterValidationTest extends ContextTestSupport { + + @Test + void testErrorThresholdNegativeIsRejected() { + Exception e = assertThrows(Exception.class, () -> context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from("direct:start") + .split(body()).errorThreshold(-0.1) + .to("mock:split"); + } + })); + assertTrue(findIllegalArgumentException(e, "errorThreshold")); + } + + @Test + void testErrorThresholdAboveOneIsRejected() { + Exception e = assertThrows(Exception.class, () -> context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from("direct:start") + .split(body()).errorThreshold(1.5) + .to("mock:split"); + } + })); + assertTrue(findIllegalArgumentException(e, "errorThreshold")); + } + + @Test + void testMaxFailedRecordsNegativeIsRejected() { + Exception e = assertThrows(Exception.class, () -> context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from("direct:start") + .split(body()).maxFailedRecords(-1) + .to("mock:split"); + } + })); + assertTrue(findIllegalArgumentException(e, "maxFailedRecords")); + } + + @Test + void testWatermarkKeyWithoutResumeStrategyIsRejected() { + Exception e = assertThrows(Exception.class, () -> context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from("direct:start") + .split(body()).watermarkKey("someKey") + .to("mock:split"); + } + })); + assertTrue(findIllegalArgumentException(e, "resumeStrategy")); + } + + @Test + void testResumeStrategyWithoutKeyIsRejected() { + context.getRegistry().bind("myStrategy", + new SplitterTestResumeStrategy(new ConcurrentHashMap<>())); + Exception e = assertThrows(Exception.class, () -> context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from("direct:start") + .split(body()).resumeStrategy("myStrategy") + .to("mock:split"); + } + })); + assertTrue(findIllegalArgumentException(e, "resumeStrategy")); + } + + @Test + void testWatermarkExpressionWithoutResumeStrategyIsRejected() { + Exception e = assertThrows(Exception.class, () -> context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from("direct:start") + .split(body()).watermarkExpression("${body}") + .to("mock:split"); + } + })); + assertTrue(findIllegalArgumentException(e, "watermarkExpression")); + } + + private boolean findIllegalArgumentException(Throwable e, String keyword) { + while (e != null) { + if (e instanceof IllegalArgumentException && e.getMessage() != null && e.getMessage().contains(keyword)) { + return true; + } + e = e.getCause(); + } + return false; + } +} diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterWatermarkTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterWatermarkTest.java new file mode 100644 index 0000000000000..490571ff3a22c --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterWatermarkTest.java @@ -0,0 +1,379 @@ +/* + * 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.processor; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; +import org.apache.camel.SplitResult; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.resume.ResumeStrategy; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; + +class SplitterWatermarkTest extends ContextTestSupport { + + private final Map store = new ConcurrentHashMap<>(); + private final ResumeStrategy strategy = new SplitterTestResumeStrategy(store); + + @Test + void testIndexBasedWatermarkFirstRun() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:split"); + mock.expectedMessageCount(5); + + template.sendBody("direct:index", Arrays.asList("a", "b", "c", "d", "e")); + + mock.assertIsSatisfied(); + + // watermark should be stored as last index (4) + assertEquals("4", store.get("testJob")); + } + + @Test + void testIndexBasedWatermarkSecondRun() throws Exception { + // simulate a previous run that processed items 0-2 + store.put("testJob", "2"); + + MockEndpoint mock = getMockEndpoint("mock:split"); + mock.expectedMessageCount(2); // only items 3 and 4 + + template.sendBody("direct:index", Arrays.asList("a", "b", "c", "d", "e")); + + mock.assertIsSatisfied(); + + // bodies should be d, e (items at index 3 and 4) + assertEquals("d", mock.getReceivedExchanges().get(0).getIn().getBody(String.class)); + assertEquals("e", mock.getReceivedExchanges().get(1).getIn().getBody(String.class)); + + // watermark updated to 4 + assertEquals("4", store.get("testJob")); + } + + @Test + void testIndexBasedWatermarkNoUpdateOnAbort() throws Exception { + // watermark at 0, maxFailedRecords=1 + store.put("testJob2", "0"); + + MockEndpoint mock = getMockEndpoint("mock:split2"); + mock.expectedMinimumMessageCount(0); + + // items after skipping index 0: FAIL, c, d — FAIL triggers abort + template.send("direct:index-abort", + e -> e.getIn().setBody(Arrays.asList("a", "FAIL", "c", "d"))); + + mock.assertIsSatisfied(); + + // watermark should NOT be updated (still 0) + assertEquals("0", store.get("testJob2")); + } + + @Test + void testValueBasedWatermark() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:split3"); + mock.expectedMessageCount(3); + + Exchange result = template.send("direct:value", + e -> e.getIn().setBody(Arrays.asList("2024-01-01", "2024-01-02", "2024-01-03"))); + + mock.assertIsSatisfied(); + + // no previous watermark, so SPLIT_WATERMARK should not be set + assertNull(result.getProperty(Exchange.SPLIT_WATERMARK)); + + // watermark expression evaluated per-item: last successful item's body is stored + assertEquals("2024-01-03", store.get("dateJob")); + } + + @Test + void testValueBasedWatermarkWithPreviousValue() throws Exception { + store.put("dateJob", "2024-01-01"); + + MockEndpoint mock = getMockEndpoint("mock:split3"); + mock.expectedMessageCount(3); + + Exchange result = template.send("direct:value", + e -> e.getIn().setBody(Arrays.asList("2024-01-02", "2024-01-03", "2024-01-04"))); + + mock.assertIsSatisfied(); + + // previous watermark should be exposed as exchange property + assertEquals("2024-01-01", result.getProperty(Exchange.SPLIT_WATERMARK)); + + // watermark updated to the last processed date + assertEquals("2024-01-04", store.get("dateJob")); + } + + @Test + void testValueBasedWatermarkWithParallelProcessing() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:parallel-wm"); + mock.expectedMessageCount(5); + + template.sendBody("direct:parallel-value", + Arrays.asList("val-0", "val-1", "val-2", "val-3", "val-4")); + + mock.assertIsSatisfied(); + + // In parallel mode, the watermark should be from the highest-indexed item (val-4) + assertEquals("val-4", store.get("parallelJob")); + } + + @Test + void testIndexBasedWatermarkWithGroup() throws Exception { + // 10 items, group=3 → 4 chunks: [a,b,c], [d,e,f], [g,h,i], [j] + // Watermark should store "9" (last raw item index), not "3" (last chunk index) + MockEndpoint mock = getMockEndpoint("mock:group-wm"); + mock.expectedMessageCount(4); + + template.sendBody("direct:group-index", + Arrays.asList("a", "b", "c", "d", "e", "f", "g", "h", "i", "j")); + + mock.assertIsSatisfied(); + + assertEquals("9", store.get("groupJob"), "Watermark should be last raw item index, not chunk index"); + } + + @Test + void testIndexBasedWatermarkWithGroupSecondRun() throws Exception { + // Pre-populate watermark at 4 → skip items 0-4 (a,b,c,d,e) + // Remaining: f,g,h,i,j → chunks [f,g,h], [i,j] + store.put("groupJob", "4"); + + MockEndpoint mock = getMockEndpoint("mock:group-wm"); + mock.expectedMessageCount(2); + + template.sendBody("direct:group-index", + Arrays.asList("a", "b", "c", "d", "e", "f", "g", "h", "i", "j")); + + mock.assertIsSatisfied(); + + // first chunk should be [f,g,h] + List firstChunk = mock.getReceivedExchanges().get(0).getIn().getBody(List.class); + assertEquals(Arrays.asList("f", "g", "h"), firstChunk); + + // watermark updated to 9 (last raw item index) + assertEquals("9", store.get("groupJob")); + } + + @Test + void testIndexBasedWatermarkWithParallelProcessing() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:parallel-idx"); + mock.expectedMessageCount(5); + + template.sendBody("direct:parallel-index", + Arrays.asList("a", "b", "c", "d", "e")); + + mock.assertIsSatisfied(); + + assertEquals("4", store.get("parallelIdxJob")); + } + + @Test + void testValueBasedWatermarkExpressionReturnsNull() throws Exception { + // watermarkExpression=${header.wm} — some items won't have the header, so expression returns null + MockEndpoint mock = getMockEndpoint("mock:null-expr"); + mock.expectedMessageCount(3); + + template.sendBody("direct:null-expr", Arrays.asList("a", "b", "c")); + + mock.assertIsSatisfied(); + + // only items where the header is set contribute to the watermark + // the processor sets the header only for "b", so watermark should be "wm-b" + assertEquals("wm-b", store.get("nullExprJob")); + } + + @Test + void testValueBasedWatermarkAllItemsFail() throws Exception { + // all items fail → no watermark expression evaluated → latestRef stays null → no watermark persisted + MockEndpoint mock = getMockEndpoint("mock:all-fail-wm"); + mock.expectedMessageCount(0); + + template.send("direct:all-fail-wm", + e -> e.getIn().setBody(Arrays.asList("FAIL", "FAIL", "FAIL"))); + + mock.assertIsSatisfied(); + + // watermark should NOT be persisted (all items failed) + assertNull(store.get("allFailJob"), "Watermark should not be persisted when all items fail"); + } + + @Test + void testCombinedErrorThresholdAndWatermark() throws Exception { + // both errorThreshold and watermark together + MockEndpoint mock = getMockEndpoint("mock:combined-wm"); + mock.expectedMessageCount(4); + + Exchange result = template.send("direct:combined-wm", + e -> e.getIn().setBody(Arrays.asList("a", "FAIL", "b", "c", "d"))); + + mock.assertIsSatisfied(); + + // watermark should be updated (processing succeeded overall) + assertEquals("4", store.get("combinedJob")); + + // SplitResult should be set (errorThreshold is configured) + SplitResult splitResult = result.getProperty(Exchange.SPLIT_RESULT, SplitResult.class); + assertNotNull(splitResult); + assertEquals(5, splitResult.getTotalItems()); + assertEquals(1, splitResult.getFailureCount()); + assertFalse(splitResult.isAborted()); + } + + @Test + void testIndexBasedWatermarkEmptyInput() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:split"); + mock.expectedMessageCount(0); + + template.sendBody("direct:index", Collections.emptyList()); + + mock.assertIsSatisfied(); + + // no items processed, watermark should not be set + assertNull(store.get("testJob"), "Watermark should not be set for empty input"); + } + + @Test + void testIndexBasedWatermarkFirstRunSingleItem() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:split"); + mock.expectedMessageCount(1); + + template.sendBody("direct:index", List.of("only")); + + mock.assertIsSatisfied(); + + assertEquals("0", store.get("testJob"), "Watermark should be 0 for single item"); + } + + @Test + void testValueBasedWatermarkNoUpdateOnAbort() throws Exception { + // value-based watermark should NOT update when processing is aborted + store.put("abortValJob", "previous"); + + MockEndpoint mock = getMockEndpoint("mock:abort-val-wm"); + mock.expectedMinimumMessageCount(0); + + template.send("direct:abort-val-wm", + e -> e.getIn().setBody(Arrays.asList("FAIL", "FAIL", "a"))); + + mock.assertIsSatisfied(); + + // watermark should retain its previous value + assertEquals("previous", store.get("abortValJob"), + "Watermark should not be updated when processing is aborted"); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:index") + .split(body()).resumeStrategy(strategy, "testJob") + .to("mock:split"); + + from("direct:index-abort") + .split(body()).resumeStrategy(strategy, "testJob2").maxFailedRecords(1) + .process(exchange -> { + String body = exchange.getIn().getBody(String.class); + if ("FAIL".equals(body)) { + throw new IllegalArgumentException("Simulated failure"); + } + }) + .to("mock:split2"); + + from("direct:value") + .split(body()) + .resumeStrategy(strategy, "dateJob") + .watermarkExpression("${body}") + .to("mock:split3"); + + from("direct:group-index") + .split(body()).group(3).resumeStrategy(strategy, "groupJob") + .to("mock:group-wm"); + + from("direct:parallel-value") + .split(body()).parallelProcessing() + .resumeStrategy(strategy, "parallelJob") + .watermarkExpression("${body}") + .to("mock:parallel-wm"); + + from("direct:parallel-index") + .split(body()).parallelProcessing() + .resumeStrategy(strategy, "parallelIdxJob") + .to("mock:parallel-idx"); + + from("direct:null-expr") + .split(body()) + .resumeStrategy(strategy, "nullExprJob") + .watermarkExpression("${header.wm}") + .process(exchange -> { + // only set the watermark header for specific items + String body = exchange.getIn().getBody(String.class); + if ("b".equals(body)) { + exchange.getIn().setHeader("wm", "wm-b"); + } + }) + .to("mock:null-expr"); + + from("direct:all-fail-wm") + .split(body()) + .resumeStrategy(strategy, "allFailJob") + .watermarkExpression("${body}") + .maxFailedRecords(10) + .process(exchange -> { + throw new IllegalArgumentException("All fail"); + }) + .to("mock:all-fail-wm"); + + from("direct:combined-wm") + .split(body()) + .resumeStrategy(strategy, "combinedJob") + .maxFailedRecords(5) + .process(exchange -> { + String body = exchange.getIn().getBody(String.class); + if ("FAIL".equals(body)) { + throw new IllegalArgumentException("Simulated"); + } + }) + .to("mock:combined-wm"); + + from("direct:abort-val-wm") + .split(body()) + .resumeStrategy(strategy, "abortValJob") + .watermarkExpression("${body}") + .maxFailedRecords(2) + .process(exchange -> { + String body = exchange.getIn().getBody(String.class); + if ("FAIL".equals(body)) { + throw new IllegalArgumentException("Simulated failure"); + } + }) + .to("mock:abort-val-wm"); + } + }; + } +} 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 9640d635bbe7d..3011b59f69f1e 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 @@ -2516,6 +2516,12 @@ protected void doWriteSplitDefinition(StringBuilder sb, SplitDefinition def) { doWriteAttribute(sb, "executorService", def.getExecutorService(), null); doWriteAttribute(sb, "onPrepare", def.getOnPrepare(), null); doWriteAttribute(sb, "shareUnitOfWork", def.getShareUnitOfWork(), null); + doWriteAttribute(sb, "group", def.getGroup(), null); + doWriteAttribute(sb, "errorThreshold", def.getErrorThreshold(), null); + doWriteAttribute(sb, "maxFailedRecords", def.getMaxFailedRecords(), null); + doWriteAttribute(sb, "resumeStrategy", def.getResumeStrategy(), null); + doWriteAttribute(sb, "watermarkKey", def.getWatermarkKey(), null); + doWriteAttribute(sb, "watermarkExpression", def.getWatermarkExpression(), null); doWriteOutputExpressionNodeElements(sb, def); } protected void doWriteStepDefinition(StringBuilder sb, StepDefinition def) { 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 abcef047096cb..c54e2fd73f631 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 @@ -1140,15 +1140,21 @@ protected SplitDefinition doParseSplitDefinition() throws IOException, XmlPullPa case "aggregationStrategyMethodAllowNull": def.setAggregationStrategyMethodAllowNull(val); yield true; case "aggregationStrategyMethodName": def.setAggregationStrategyMethodName(val); yield true; case "delimiter": def.setDelimiter(val); yield true; + case "errorThreshold": def.setErrorThreshold(val); yield true; case "executorService": def.setExecutorService(val); yield true; + case "group": def.setGroup(val); yield true; + case "maxFailedRecords": def.setMaxFailedRecords(val); yield true; case "onPrepare": def.setOnPrepare(val); yield true; case "parallelAggregate": def.setParallelAggregate(val); yield true; case "parallelProcessing": def.setParallelProcessing(val); yield true; + case "resumeStrategy": def.setResumeStrategy(val); yield true; case "shareUnitOfWork": def.setShareUnitOfWork(val); yield true; case "stopOnException": def.setStopOnException(val); yield true; case "streaming": def.setStreaming(val); yield true; case "synchronous": def.setSynchronous(val); yield true; case "timeout": def.setTimeout(val); yield true; + case "watermarkExpression": def.setWatermarkExpression(val); yield true; + case "watermarkKey": def.setWatermarkKey(val); yield true; default: yield processorDefinitionAttributeHandler().accept(def, key, val); }, outputExpressionNodeElementHandler(), noValueHandler()); } 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 0e15504b96da5..01cb827aa25ab 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 @@ -1769,6 +1769,12 @@ protected void doWriteSplitDefinition(String name, SplitDefinition def) throws I doWriteAttribute("executorService", def.getExecutorService(), null); doWriteAttribute("onPrepare", def.getOnPrepare(), null); doWriteAttribute("shareUnitOfWork", def.getShareUnitOfWork(), null); + doWriteAttribute("group", def.getGroup(), null); + doWriteAttribute("errorThreshold", def.getErrorThreshold(), null); + doWriteAttribute("maxFailedRecords", def.getMaxFailedRecords(), null); + doWriteAttribute("resumeStrategy", def.getResumeStrategy(), null); + doWriteAttribute("watermarkKey", def.getWatermarkKey(), null); + doWriteAttribute("watermarkExpression", def.getWatermarkExpression(), null); doWriteOutputExpressionNodeElements(def); endElement(name); } 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 8e9db27a681a8..de61b92c6eabf 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 @@ -1765,6 +1765,12 @@ protected JsonObject doWriteSplitDefinition(SplitDefinition def) { doWriteAttribute(jo, "executorService", def.getExecutorService(), null); doWriteAttribute(jo, "onPrepare", def.getOnPrepare(), null); doWriteAttribute(jo, "shareUnitOfWork", def.getShareUnitOfWork(), null); + doWriteAttribute(jo, "group", def.getGroup(), null); + doWriteAttribute(jo, "errorThreshold", def.getErrorThreshold(), null); + doWriteAttribute(jo, "maxFailedRecords", def.getMaxFailedRecords(), null); + doWriteAttribute(jo, "resumeStrategy", def.getResumeStrategy(), null); + doWriteAttribute(jo, "watermarkKey", def.getWatermarkKey(), null); + doWriteAttribute(jo, "watermarkExpression", def.getWatermarkExpression(), null); doWriteOutputExpressionNodeElements(jo, def); return jo; } 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 03a1de268bb11..9100e8858fd7e 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 @@ -16960,19 +16960,25 @@ protected boolean setProperty(SpELExpression target, String propertyKey, @YamlProperty(name = "delimiter", type = "string", defaultValue = ",", description = "Delimiter used in splitting messages. Can be turned off using the value false. To force not splitting then the delimiter can be set to single to use the value as a single list. The default value is comma.", displayName = "Delimiter"), @YamlProperty(name = "description", type = "string", description = "The description for this node", displayName = "Description"), @YamlProperty(name = "disabled", type = "boolean", defaultValue = "false", description = "Whether to disable this EIP from the route during build time. Once an EIP has been disabled then it cannot be enabled later at runtime.", displayName = "Disabled"), + @YamlProperty(name = "errorThreshold", type = "number", description = "Sets the error threshold as a fraction (0.0-1.0) of failed items before aborting the split operation. For example, 0.1 means abort if more than 10% of items fail. When the threshold is exceeded, a org.apache.camel.CamelExchangeException is thrown. This option is mutually exclusive with stopOnException . When set, individual item failures are tracked but processing continues until the threshold is exceeded. Note: When combined with parallelProcessing , the failure ratio may vary between runs because parallel items complete in non-deterministic order. For deterministic abort behavior with parallel processing, prefer maxFailedRecords (absolute count) over errorThreshold (ratio).", displayName = "Error Threshold"), @YamlProperty(name = "executorService", type = "string", description = "Reference to a custom thread pool to use for parallel processing. Setting this option implies parallel processing.", displayName = "Executor Service"), @YamlProperty(name = "expression", type = "object:org.apache.camel.model.language.ExpressionDefinition", description = "The expression that returns the value to use for splitting. The result can be an Iterator, Iterable, Array, Collection, Map, NodeList, or a delimited String.", displayName = "Expression", oneOf = "expression"), + @YamlProperty(name = "group", type = "number", description = "Groups N split messages into a single message with a java.util.List body. This allows processing items in chunks instead of one at a time.", displayName = "Group"), @YamlProperty(name = "id", type = "string", description = "The id of this node", displayName = "Id"), + @YamlProperty(name = "maxFailedRecords", type = "number", description = "Sets the maximum number of failed records before aborting the split operation. When the count is exceeded, a org.apache.camel.CamelExchangeException is thrown. This option is mutually exclusive with stopOnException . Can be combined with errorThreshold processing aborts when either threshold is exceeded.", displayName = "Max Failed Records"), @YamlProperty(name = "note", type = "string", description = "The note for this node", displayName = "Note"), @YamlProperty(name = "onPrepare", type = "string", description = "Reference to a processor for preparing the exchange to be sent. Can be used to deep-clone messages that should be sent.", displayName = "On Prepare"), @YamlProperty(name = "parallelAggregate", type = "boolean", deprecated = true, defaultValue = "false", description = "If enabled then the aggregate method on AggregationStrategy can be called concurrently. Notice that this would require the implementation of AggregationStrategy to be implemented as thread-safe. By default this is false meaning that Camel synchronizes the call to the aggregate method. Though in some use-cases this can be used to archive higher performance when the AggregationStrategy is implemented as thread-safe.", displayName = "Parallel Aggregate"), @YamlProperty(name = "parallelProcessing", type = "boolean", defaultValue = "false", description = "If enabled then processing each split message occurs concurrently. The caller thread still waits until all messages are fully processed before it continues.", displayName = "Parallel Processing"), + @YamlProperty(name = "resumeStrategy", type = "string", description = "Sets a ResumeStrategy for resume-from-last-position support. The watermark key must also be configured via watermarkKey(String) .", displayName = "Resume Strategy"), @YamlProperty(name = "shareUnitOfWork", type = "boolean", defaultValue = "false", description = "Shares the unit of work with the parent and each of the split messages. By default each split exchange has its own individual unit of work.", displayName = "Share Unit Of Work"), @YamlProperty(name = "steps", type = "array:org.apache.camel.model.ProcessorDefinition"), @YamlProperty(name = "stopOnException", type = "boolean", defaultValue = "false", description = "If enabled then stops further split processing if an exception or failure occurred during processing of a split message, and the caused exception will be thrown. The default behavior is to not stop but continue processing till the end.", displayName = "Stop On Exception"), @YamlProperty(name = "streaming", type = "boolean", defaultValue = "false", description = "When enabled then the splitter splits the original message on-demand, and each split message is processed one by one. This reduces memory usage as the splitter does not split all messages first.", displayName = "Streaming"), @YamlProperty(name = "synchronous", type = "boolean", defaultValue = "false", description = "When enabled then the same thread is used to continue routing after the split is complete, even if parallel processing is enabled.", displayName = "Synchronous"), - @YamlProperty(name = "timeout", type = "string", defaultValue = "0", description = "Total timeout in millis when using parallel processing. If the splitter has not been able to process all replies within the given timeframe, then the timeout triggers and the splitter breaks out and continues.", displayName = "Timeout") + @YamlProperty(name = "timeout", type = "string", defaultValue = "0", description = "Total timeout in millis when using parallel processing. If the splitter has not been able to process all replies within the given timeframe, then the timeout triggers and the splitter breaks out and continues.", displayName = "Timeout"), + @YamlProperty(name = "watermarkExpression", type = "string", description = "Sets a Simple expression to evaluate on each completed sub-exchange to determine the new watermark value. When set, enables value-based watermarking instead of index-based. The expression is evaluated using the Simple language.", displayName = "Watermark Expression"), + @YamlProperty(name = "watermarkKey", type = "string", description = "Sets the key to use in the watermark store.", displayName = "Watermark Key") } ) public static class SplitDefinitionDeserializer extends YamlDeserializerBase { @@ -17015,6 +17021,11 @@ protected boolean setProperty(SplitDefinition target, String propertyKey, target.setDisabled(val); break; } + case "errorThreshold": { + String val = asText(node); + target.setErrorThreshold(val); + break; + } case "executorService": { String val = asText(node); target.setExecutorService(val); @@ -17025,6 +17036,16 @@ protected boolean setProperty(SplitDefinition target, String propertyKey, target.setExpression(val); break; } + case "group": { + String val = asText(node); + target.setGroup(val); + break; + } + case "maxFailedRecords": { + String val = asText(node); + target.setMaxFailedRecords(val); + break; + } case "onPrepare": { String val = asText(node); target.setOnPrepare(val); @@ -17040,6 +17061,11 @@ protected boolean setProperty(SplitDefinition target, String propertyKey, target.setParallelProcessing(val); break; } + case "resumeStrategy": { + String val = asText(node); + target.setResumeStrategy(val); + break; + } case "shareUnitOfWork": { String val = asText(node); target.setShareUnitOfWork(val); @@ -17065,6 +17091,16 @@ protected boolean setProperty(SplitDefinition target, String propertyKey, target.setTimeout(val); break; } + case "watermarkExpression": { + String val = asText(node); + target.setWatermarkExpression(val); + break; + } + case "watermarkKey": { + String val = asText(node); + target.setWatermarkKey(val); + break; + } case "id": { String val = asText(node); target.setId(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 307f90a9553a9..b255cbe5f0a27 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 @@ -4655,6 +4655,11 @@ "description" : "Whether to disable this EIP from the route during build time. Once an EIP has been disabled then it cannot be enabled later at runtime.", "default" : false }, + "errorThreshold" : { + "type" : "number", + "title" : "Error Threshold", + "description" : "Sets the error threshold as a fraction (0.0-1.0) of failed items before aborting the split operation. For example, 0.1 means abort if more than 10% of items fail. When the threshold is exceeded, a org.apache.camel.CamelExchangeException is thrown. This option is mutually exclusive with stopOnException . When set, individual item failures are tracked but processing continues until the threshold is exceeded. Note: When combined with parallelProcessing , the failure ratio may vary between runs because parallel items complete in non-deterministic order. For deterministic abort behavior with parallel processing, prefer maxFailedRecords (absolute count) over errorThreshold (ratio)." + }, "executorService" : { "type" : "string", "title" : "Executor Service", @@ -4665,11 +4670,21 @@ "description" : "The expression that returns the value to use for splitting. The result can be an Iterator, Iterable, Array, Collection, Map, NodeList, or a delimited String.", "$ref" : "#/items/definitions/org.apache.camel.model.language.ExpressionDefinition" }, + "group" : { + "type" : "number", + "title" : "Group", + "description" : "Groups N split messages into a single message with a java.util.List body. This allows processing items in chunks instead of one at a time." + }, "id" : { "type" : "string", "title" : "Id", "description" : "The id of this node" }, + "maxFailedRecords" : { + "type" : "number", + "title" : "Max Failed Records", + "description" : "Sets the maximum number of failed records before aborting the split operation. When the count is exceeded, a org.apache.camel.CamelExchangeException is thrown. This option is mutually exclusive with stopOnException . Can be combined with errorThreshold processing aborts when either threshold is exceeded." + }, "note" : { "type" : "string", "title" : "Note", @@ -4693,6 +4708,11 @@ "description" : "If enabled then processing each split message occurs concurrently. The caller thread still waits until all messages are fully processed before it continues.", "default" : false }, + "resumeStrategy" : { + "type" : "string", + "title" : "Resume Strategy", + "description" : "Sets a ResumeStrategy for resume-from-last-position support. The watermark key must also be configured via watermarkKey(String) ." + }, "shareUnitOfWork" : { "type" : "boolean", "title" : "Share Unit Of Work", @@ -4728,6 +4748,16 @@ "title" : "Timeout", "description" : "Total timeout in millis when using parallel processing. If the splitter has not been able to process all replies within the given timeframe, then the timeout triggers and the splitter breaks out and continues.", "default" : "0" + }, + "watermarkExpression" : { + "type" : "string", + "title" : "Watermark Expression", + "description" : "Sets a Simple expression to evaluate on each completed sub-exchange to determine the new watermark value. When set, enables value-based watermarking instead of index-based. The expression is evaluated using the Simple language." + }, + "watermarkKey" : { + "type" : "string", + "title" : "Watermark Key", + "description" : "Sets the key to use in the watermark store." } } }, 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 08e2826ac762f..f97e7cb55df1f 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 @@ -7319,16 +7319,31 @@ "description" : "Whether to disable this EIP from the route during build time. Once an EIP has been disabled then it cannot be enabled later at runtime.", "default" : false }, + "errorThreshold" : { + "type" : "number", + "title" : "Error Threshold", + "description" : "Sets the error threshold as a fraction (0.0-1.0) of failed items before aborting the split operation. For example, 0.1 means abort if more than 10% of items fail. When the threshold is exceeded, a org.apache.camel.CamelExchangeException is thrown. This option is mutually exclusive with stopOnException . When set, individual item failures are tracked but processing continues until the threshold is exceeded. Note: When combined with parallelProcessing , the failure ratio may vary between runs because parallel items complete in non-deterministic order. For deterministic abort behavior with parallel processing, prefer maxFailedRecords (absolute count) over errorThreshold (ratio)." + }, "executorService" : { "type" : "string", "title" : "Executor Service", "description" : "Reference to a custom thread pool to use for parallel processing. Setting this option implies parallel processing." }, + "group" : { + "type" : "number", + "title" : "Group", + "description" : "Groups N split messages into a single message with a java.util.List body. This allows processing items in chunks instead of one at a time." + }, "id" : { "type" : "string", "title" : "Id", "description" : "The id of this node" }, + "maxFailedRecords" : { + "type" : "number", + "title" : "Max Failed Records", + "description" : "Sets the maximum number of failed records before aborting the split operation. When the count is exceeded, a org.apache.camel.CamelExchangeException is thrown. This option is mutually exclusive with stopOnException . Can be combined with errorThreshold processing aborts when either threshold is exceeded." + }, "note" : { "type" : "string", "title" : "Note", @@ -7352,6 +7367,11 @@ "description" : "If enabled then processing each split message occurs concurrently. The caller thread still waits until all messages are fully processed before it continues.", "default" : false }, + "resumeStrategy" : { + "type" : "string", + "title" : "Resume Strategy", + "description" : "Sets a ResumeStrategy for resume-from-last-position support. The watermark key must also be configured via watermarkKey(String) ." + }, "shareUnitOfWork" : { "type" : "boolean", "title" : "Share Unit Of Work", @@ -7388,6 +7408,16 @@ "description" : "Total timeout in millis when using parallel processing. If the splitter has not been able to process all replies within the given timeframe, then the timeout triggers and the splitter breaks out and continues.", "default" : "0" }, + "watermarkExpression" : { + "type" : "string", + "title" : "Watermark Expression", + "description" : "Sets a Simple expression to evaluate on each completed sub-exchange to determine the new watermark value. When set, enables value-based watermarking instead of index-based. The expression is evaluated using the Simple language." + }, + "watermarkKey" : { + "type" : "string", + "title" : "Watermark Key", + "description" : "Sets the key to use in the watermark store." + }, "constant" : { }, "csimple" : { }, "datasonnet" : { },