This is an automated email from the ASF dual-hosted git repository. acosentino pushed a commit to branch flink-improv in repository https://gitbox.apache.org/repos/asf/camel.git
commit 4542dacf701d78708f0f355ad9b82863b80988f0 Author: Andrea Cosentino <[email protected]> AuthorDate: Fri Oct 17 10:25:25 2025 +0200 CAMEL-22567 - Camel-Flink: Deprecate DataSet API in favor of DataStream API Signed-off-by: Andrea Cosentino <[email protected]> --- .../org/apache/camel/catalog/components/flink.json | 12 +- .../dsl/FlinkComponentBuilderFactory.java | 1 + .../endpoint/dsl/FlinkEndpointBuilderFactory.java | 199 +++++++++++++++++++++ 3 files changed, 210 insertions(+), 2 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/flink.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/flink.json index 85f0a4f96399..a19a18dabbb5 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/flink.json +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/flink.json @@ -24,7 +24,7 @@ "remote": true }, "componentProperties": { - "dataSetCallback": { "index": 0, "kind": "property", "displayName": "Data Set Callback", "group": "producer", "label": "", "required": false, "type": "object", "javaType": "org.apache.camel.component.flink.DataSetCallback", "deprecated": false, "autowired": false, "secret": false, "description": "Function performing action against a DataSet." }, + "dataSetCallback": { "index": 0, "kind": "property", "displayName": "Data Set Callback", "group": "producer", "label": "", "required": false, "type": "object", "javaType": "org.apache.camel.component.flink.DataSetCallback", "deprecated": true, "autowired": false, "secret": false, "description": "Function performing action against a DataSet." }, "dataStream": { "index": 1, "kind": "property", "displayName": "Data Stream", "group": "producer", "label": "", "required": false, "type": "object", "javaType": "org.apache.flink.streaming.api.datastream.DataStream", "deprecated": false, "autowired": false, "secret": false, "description": "DataStream to compute against." }, "dataStreamCallback": { "index": 2, "kind": "property", "displayName": "Data Stream Callback", "group": "producer", "label": "", "required": false, "type": "object", "javaType": "org.apache.camel.component.flink.DataStreamCallback", "deprecated": false, "autowired": false, "secret": false, "description": "Function performing action against a DataStream." }, "lazyStartProducer": { "index": 3, "kind": "property", "displayName": "Lazy Start Producer", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail [...] @@ -43,6 +43,14 @@ "dataSetCallback": { "index": 3, "kind": "parameter", "displayName": "Data Set Callback", "group": "producer", "label": "", "required": false, "type": "object", "javaType": "org.apache.camel.component.flink.DataSetCallback", "deprecated": false, "autowired": false, "secret": false, "description": "Function performing action against a DataSet." }, "dataStream": { "index": 4, "kind": "parameter", "displayName": "Data Stream", "group": "producer", "label": "", "required": false, "type": "object", "javaType": "org.apache.flink.streaming.api.datastream.DataStream", "deprecated": false, "autowired": false, "secret": false, "description": "DataStream to compute against." }, "dataStreamCallback": { "index": 5, "kind": "parameter", "displayName": "Data Stream Callback", "group": "producer", "label": "", "required": false, "type": "object", "javaType": "org.apache.camel.component.flink.DataStreamCallback", "deprecated": false, "autowired": false, "secret": false, "description": "Function performing action against a DataStream." }, - "lazyStartProducer": { "index": 6, "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a produc [...] + "checkpointingMode": { "index": 6, "kind": "parameter", "displayName": "Checkpointing Mode", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "enum", "javaType": "java.lang.String", "enum": [ "EXACTLY_ONCE", "AT_LEAST_ONCE" ], "deprecated": false, "autowired": false, "secret": false, "description": "Checkpointing mode: EXACTLY_ONCE (default) or AT_LEAST_ONCE. EXACTLY_ONCE provides stronger guarantees but may have higher overhead." }, + "checkpointInterval": { "index": 7, "kind": "parameter", "displayName": "Checkpoint Interval", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "integer", "javaType": "java.lang.Long", "deprecated": false, "autowired": false, "secret": false, "description": "Interval in milliseconds between checkpoints. Enables checkpointing when set. Recommended for streaming jobs to ensure fault tolerance." }, + "checkpointTimeout": { "index": 8, "kind": "parameter", "displayName": "Checkpoint Timeout", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "integer", "javaType": "java.lang.Long", "deprecated": false, "autowired": false, "secret": false, "description": "Timeout in milliseconds for checkpoints. Checkpoints that take longer will be aborted." }, + "executionMode": { "index": 9, "kind": "parameter", "displayName": "Execution Mode", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "enum", "javaType": "java.lang.String", "enum": [ "STREAMING", "BATCH", "AUTOMATIC" ], "deprecated": false, "autowired": false, "secret": false, "description": "Execution mode for the Flink job. Options: STREAMING (default), BATCH, AUTOMATIC. BATCH mode is recommended for bounded streams (batch processing)." }, + "jobName": { "index": 10, "kind": "parameter", "displayName": "Job Name", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Name for the Flink job. Useful for identification in the Flink UI and logs." }, + "lazyStartProducer": { "index": 11, "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a produ [...] + "maxParallelism": { "index": 12, "kind": "parameter", "displayName": "Max Parallelism", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "Maximum parallelism for the Flink job. Defines the upper bound for dynamic scaling and the number of key groups for stateful operators." }, + "minPauseBetweenCheckpoints": { "index": 13, "kind": "parameter", "displayName": "Min Pause Between Checkpoints", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "integer", "javaType": "java.lang.Long", "deprecated": false, "autowired": false, "secret": false, "description": "Minimum pause in milliseconds between consecutive checkpoints. Helps prevent checkpoint storms under heavy load." }, + "parallelism": { "index": 14, "kind": "parameter", "displayName": "Parallelism", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "Parallelism for the Flink job. If not set, uses the default parallelism of the execution environment." } } } diff --git a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/FlinkComponentBuilderFactory.java b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/FlinkComponentBuilderFactory.java index 897e3dc0bb2d..7ec9ea2c6c5f 100644 --- a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/FlinkComponentBuilderFactory.java +++ b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/FlinkComponentBuilderFactory.java @@ -61,6 +61,7 @@ public interface FlinkComponentBuilderFactory { * @param dataSetCallback the value to set * @return the dsl builder */ + @Deprecated default FlinkComponentBuilder dataSetCallback(org.apache.camel.component.flink.DataSetCallback dataSetCallback) { doSetProperty("dataSetCallback", dataSetCallback); return this; diff --git a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/FlinkEndpointBuilderFactory.java b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/FlinkEndpointBuilderFactory.java index 11a4d5f42bee..270771d8568c 100644 --- a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/FlinkEndpointBuilderFactory.java +++ b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/FlinkEndpointBuilderFactory.java @@ -209,6 +209,113 @@ public interface FlinkEndpointBuilderFactory { return (FlinkEndpointBuilder) this; } + /** + * Checkpointing mode: EXACTLY_ONCE (default) or AT_LEAST_ONCE. + * EXACTLY_ONCE provides stronger guarantees but may have higher + * overhead. + * + * The option is a: <code>java.lang.String</code> type. + * + * Group: producer (advanced) + * + * @param checkpointingMode the value to set + * @return the dsl builder + */ + default AdvancedFlinkEndpointBuilder checkpointingMode(String checkpointingMode) { + doSetProperty("checkpointingMode", checkpointingMode); + return this; + } + /** + * Interval in milliseconds between checkpoints. Enables checkpointing + * when set. Recommended for streaming jobs to ensure fault tolerance. + * + * The option is a: <code>java.lang.Long</code> type. + * + * Group: producer (advanced) + * + * @param checkpointInterval the value to set + * @return the dsl builder + */ + default AdvancedFlinkEndpointBuilder checkpointInterval(Long checkpointInterval) { + doSetProperty("checkpointInterval", checkpointInterval); + return this; + } + /** + * Interval in milliseconds between checkpoints. Enables checkpointing + * when set. Recommended for streaming jobs to ensure fault tolerance. + * + * The option will be converted to a <code>java.lang.Long</code> type. + * + * Group: producer (advanced) + * + * @param checkpointInterval the value to set + * @return the dsl builder + */ + default AdvancedFlinkEndpointBuilder checkpointInterval(String checkpointInterval) { + doSetProperty("checkpointInterval", checkpointInterval); + return this; + } + /** + * Timeout in milliseconds for checkpoints. Checkpoints that take longer + * will be aborted. + * + * The option is a: <code>java.lang.Long</code> type. + * + * Group: producer (advanced) + * + * @param checkpointTimeout the value to set + * @return the dsl builder + */ + default AdvancedFlinkEndpointBuilder checkpointTimeout(Long checkpointTimeout) { + doSetProperty("checkpointTimeout", checkpointTimeout); + return this; + } + /** + * Timeout in milliseconds for checkpoints. Checkpoints that take longer + * will be aborted. + * + * The option will be converted to a <code>java.lang.Long</code> type. + * + * Group: producer (advanced) + * + * @param checkpointTimeout the value to set + * @return the dsl builder + */ + default AdvancedFlinkEndpointBuilder checkpointTimeout(String checkpointTimeout) { + doSetProperty("checkpointTimeout", checkpointTimeout); + return this; + } + /** + * Execution mode for the Flink job. Options: STREAMING (default), + * BATCH, AUTOMATIC. BATCH mode is recommended for bounded streams + * (batch processing). + * + * The option is a: <code>java.lang.String</code> type. + * + * Group: producer (advanced) + * + * @param executionMode the value to set + * @return the dsl builder + */ + default AdvancedFlinkEndpointBuilder executionMode(String executionMode) { + doSetProperty("executionMode", executionMode); + return this; + } + /** + * Name for the Flink job. Useful for identification in the Flink UI and + * logs. + * + * The option is a: <code>java.lang.String</code> type. + * + * Group: producer (advanced) + * + * @param jobName the value to set + * @return the dsl builder + */ + default AdvancedFlinkEndpointBuilder jobName(String jobName) { + doSetProperty("jobName", jobName); + return this; + } /** * Whether the producer should be started lazy (on the first message). * By starting lazy you can use this to allow CamelContext and routes to @@ -255,6 +362,98 @@ public interface FlinkEndpointBuilderFactory { doSetProperty("lazyStartProducer", lazyStartProducer); return this; } + /** + * Maximum parallelism for the Flink job. Defines the upper bound for + * dynamic scaling and the number of key groups for stateful operators. + * + * The option is a: <code>java.lang.Integer</code> type. + * + * Group: producer (advanced) + * + * @param maxParallelism the value to set + * @return the dsl builder + */ + default AdvancedFlinkEndpointBuilder maxParallelism(Integer maxParallelism) { + doSetProperty("maxParallelism", maxParallelism); + return this; + } + /** + * Maximum parallelism for the Flink job. Defines the upper bound for + * dynamic scaling and the number of key groups for stateful operators. + * + * The option will be converted to a <code>java.lang.Integer</code> + * type. + * + * Group: producer (advanced) + * + * @param maxParallelism the value to set + * @return the dsl builder + */ + default AdvancedFlinkEndpointBuilder maxParallelism(String maxParallelism) { + doSetProperty("maxParallelism", maxParallelism); + return this; + } + /** + * Minimum pause in milliseconds between consecutive checkpoints. Helps + * prevent checkpoint storms under heavy load. + * + * The option is a: <code>java.lang.Long</code> type. + * + * Group: producer (advanced) + * + * @param minPauseBetweenCheckpoints the value to set + * @return the dsl builder + */ + default AdvancedFlinkEndpointBuilder minPauseBetweenCheckpoints(Long minPauseBetweenCheckpoints) { + doSetProperty("minPauseBetweenCheckpoints", minPauseBetweenCheckpoints); + return this; + } + /** + * Minimum pause in milliseconds between consecutive checkpoints. Helps + * prevent checkpoint storms under heavy load. + * + * The option will be converted to a <code>java.lang.Long</code> type. + * + * Group: producer (advanced) + * + * @param minPauseBetweenCheckpoints the value to set + * @return the dsl builder + */ + default AdvancedFlinkEndpointBuilder minPauseBetweenCheckpoints(String minPauseBetweenCheckpoints) { + doSetProperty("minPauseBetweenCheckpoints", minPauseBetweenCheckpoints); + return this; + } + /** + * Parallelism for the Flink job. If not set, uses the default + * parallelism of the execution environment. + * + * The option is a: <code>java.lang.Integer</code> type. + * + * Group: producer (advanced) + * + * @param parallelism the value to set + * @return the dsl builder + */ + default AdvancedFlinkEndpointBuilder parallelism(Integer parallelism) { + doSetProperty("parallelism", parallelism); + return this; + } + /** + * Parallelism for the Flink job. If not set, uses the default + * parallelism of the execution environment. + * + * The option will be converted to a <code>java.lang.Integer</code> + * type. + * + * Group: producer (advanced) + * + * @param parallelism the value to set + * @return the dsl builder + */ + default AdvancedFlinkEndpointBuilder parallelism(String parallelism) { + doSetProperty("parallelism", parallelism); + return this; + } } public interface FlinkBuilders {
