CAMEL-23264: Enhance Splitter EIP with chunking, error threshold, failure tracking, and watermark resume

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<org.apache.camel.model.ProcessorDefinition<java.lang.Object>>", "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<org.apache.camel.model.ProcessorDefinition<java.lang.Object>>", "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." }
   }
 }

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
 <![CDATA[
 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. Default value: false
+]]>
+            </xs:documentation>
+          </xs:annotation>
+        </xs:attribute>
+        <xs:attribute name="group" type="xs:string">
+          <xs:annotation>
+            <xs:documentation xml:lang="en">
+<![CDATA[
+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.
+]]>
+            </xs:documentation>
+          </xs:annotation>
+        </xs:attribute>
+        <xs:attribute name="errorThreshold" type="xs:string">
+          <xs:annotation>
+            <xs:documentation xml:lang="en">
+<![CDATA[
+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).
+]]>
+            </xs:documentation>
+          </xs:annotation>
+        </xs:attribute>
+        <xs:attribute name="maxFailedRecords" type="xs:string">
+          <xs:annotation>
+            <xs:documentation xml:lang="en">
+<![CDATA[
+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.
+]]>
+            </xs:documentation>
+          </xs:annotation>
+        </xs:attribute>
+        <xs:attribute name="resumeStrategy" type="xs:string">
+          <xs:annotation>
+            <xs:documentation xml:lang="en">
+<![CDATA[
+Sets a ResumeStrategy for resume-from-last-position support. The watermark key must also be configured via
+watermarkKey(String) .
+]]>
+            </xs:documentation>
+          </xs:annotation>
+        </xs:attribute>
+        <xs:attribute name="watermarkKey" type="xs:string">
+          <xs:annotation>
+            <xs:documentation xml:lang="en">
+<![CDATA[
+Sets the key to use in the watermark store.
+]]>
+            </xs:documentation>
+          </xs:annotation>
+        </xs:attribute>
+        <xs:attribute name="watermarkExpression" type="xs:string">
+          <xs:annotation>
+            <xs:documentation xml:lang="en">
+<![CDATA[
+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.
 ]]>
             </xs:documentation>
           </xs:annotation>

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
 <![CDATA[
 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. Default value: false
+]]>
+            </xs:documentation>
+          </xs:annotation>
+        </xs:attribute>
+        <xs:attribute name="group" type="xs:string">
+          <xs:annotation>
+            <xs:documentation xml:lang="en">
+<![CDATA[
+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.
+]]>
+            </xs:documentation>
+          </xs:annotation>
+        </xs:attribute>
+        <xs:attribute name="errorThreshold" type="xs:string">
+          <xs:annotation>
+            <xs:documentation xml:lang="en">
+<![CDATA[
+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).
+]]>
+            </xs:documentation>
+          </xs:annotation>
+        </xs:attribute>
+        <xs:attribute name="maxFailedRecords" type="xs:string">
+          <xs:annotation>
+            <xs:documentation xml:lang="en">
+<![CDATA[
+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.
+]]>
+            </xs:documentation>
+          </xs:annotation>
+        </xs:attribute>
+        <xs:attribute name="resumeStrategy" type="xs:string">
+          <xs:annotation>
+            <xs:documentation xml:lang="en">
+<![CDATA[
+Sets a ResumeStrategy for resume-from-last-position support. The watermark key must also be configured via
+watermarkKey(String) .
+]]>
+            </xs:documentation>
+          </xs:annotation>
+        </xs:attribute>
+        <xs:attribute name="watermarkKey" type="xs:string">
+          <xs:annotation>
+            <xs:documentation xml:lang="en">
+<![CDATA[
+Sets the key to use in the watermark store.
+]]>
+            </xs:documentation>
+          </xs:annotation>
+        </xs:attribute>
+        <xs:attribute name="watermarkExpression" type="xs:string">
+          <xs:annotation>
+            <xs:documentation xml:lang="en">
+<![CDATA[
+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.
 ]]>
             </xs:documentation>
           </xs:annotation>

core/camel-api/src/generated/java/org/apache/camel/ExchangeConstantProvider.java

@@ -17,7 +17,7 @@ public class ExchangeConstantProvider {
 
     private static final Map<String, String> MAP;
     static {
-        Map<String, String> map = new HashMap<>(151);
+        Map<String, String> 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");

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";
 

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: