This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 3464d84544f2 CAMEL-26121: camel-flink - use the component dataStream
options, and return a DataStream with the default collect (#28898)
3464d84544f2 is described below
commit 3464d84544f2112c9369a971aaddcdd9c9a08e74
Author: Claus Ibsen <[email protected]>
AuthorDate: Mon Nov 2 20:37:58 2026 +0100
CAMEL-26121: camel-flink - use the component dataStream options, and return
a DataStream with the default collect (#28898)
…return a DataStream with the default collect
FlinkEndpoint.doInit fell back to the component's dataSet and
dataSetCallback, but not to
its dataStream and dataStreamCallback, so flink:datastream failed "No
DataStream defined"
when those were set only on the component. It now falls back to them too.
collect defaults to true, and DataStreamFlinkProducer threw "collect mode
not supported for
Flink DataStreams." whenever a callback returned a DataStream, so it always
failed unless
collect=false was set. The DataStream is now returned as the body and
header unless collect
is explicitly set to true.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../org/apache/camel/catalog/components/flink.json | 4 +-
.../apache/camel/catalog/docs/flink-component.adoc | 3 +-
.../org/apache/camel/component/flink/flink.json | 4 +-
.../camel-flink/src/main/docs/flink-component.adoc | 3 +-
.../component/flink/DataStreamFlinkProducer.java | 6 +-
.../camel/component/flink/FlinkEndpoint.java | 27 ++++++-
.../DataStreamComponentOptionsAndCollectTest.java | 87 ++++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 8 ++
.../endpoint/dsl/FlinkEndpointBuilderFactory.java | 8 +-
9 files changed, 136 insertions(+), 14 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 7ed3d102de8c..f0157a0df7ad 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
@@ -38,11 +38,11 @@
},
"bodyTypes": {
"producerIn": { "carriers": [ "any" ], "description": "The body is handed
to the DataSet or DataStream callback as its payload, a List as one payload per
element; the data to compute on is the DataSet or DataStream of the endpoint or
of the CamelFlinkDataSet or CamelFlinkDataStream header" },
- "producerOut": { "carriers": [ "unknown" ], "description": "What the
callback returns, null for a void callback. A DataSet it returns is collected
into a List; a DataStream it returns fails unless collect is false, then the
DataStream (or DataSet) is the body and the CamelFlinkDataStream (or
CamelFlinkDataSet) header" }
+ "producerOut": { "carriers": [ "unknown" ], "description": "What the
callback returns, null for a void callback. A DataSet it returns is collected
into a List unless collect is false; a DataStream it returns is not collected
(it fails if collect is explicitly set to true). A DataSet or DataStream that
is not collected is the body and the CamelFlinkDataSet or CamelFlinkDataStream
header" }
},
"properties": {
"endpointType": { "index": 0, "kind": "path", "displayName": "Endpoint
Type", "group": "producer", "label": "", "required": true, "type": "enum",
"javaType": "org.apache.camel.component.flink.EndpointType", "enum": [
"dataset", "datastream" ], "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "Type of the endpoint
(dataset, datastream)." },
- "collect": { "index": 1, "kind": "parameter", "displayName": "Collect",
"group": "producer", "label": "", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": true, "description": "Indicates if results should be
collected or counted." },
+ "collect": { "index": 1, "kind": "parameter", "displayName": "Collect",
"group": "producer", "label": "", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": true, "description": "Indicates if results should be
collected or counted. A DataStream returned by a datastream callback cannot be
collected: it is returned as is, unless collect is explicitly set to true,
which fails." },
"dataSet": { "index": 2, "kind": "parameter", "displayName": "Data Set",
"group": "producer", "label": "", "required": false, "type": "object",
"javaType": "org.apache.flink.api.java.DataSet", "deprecated": false,
"autowired": false, "secret": false, "description": "DataSet to compute
against." },
"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." },
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/flink-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/flink-component.adoc
index fef2e53d0a9b..0ccc19741104 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/flink-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/flink-component.adoc
@@ -373,7 +373,8 @@ YAML::
|collect
|Boolean
|true
-|Whether to collect results (not applicable for unbounded streams).
+|Whether to collect the DataSet returned by a dataset callback into a List. A
DataStream returned by a datastream callback is
+not collected: it is the message body and the `CamelFlinkDataStream` header,
and an explicit `collect=true` fails.
|===
diff --git
a/components/camel-flink/src/generated/resources/META-INF/org/apache/camel/component/flink/flink.json
b/components/camel-flink/src/generated/resources/META-INF/org/apache/camel/component/flink/flink.json
index 7ed3d102de8c..f0157a0df7ad 100644
---
a/components/camel-flink/src/generated/resources/META-INF/org/apache/camel/component/flink/flink.json
+++
b/components/camel-flink/src/generated/resources/META-INF/org/apache/camel/component/flink/flink.json
@@ -38,11 +38,11 @@
},
"bodyTypes": {
"producerIn": { "carriers": [ "any" ], "description": "The body is handed
to the DataSet or DataStream callback as its payload, a List as one payload per
element; the data to compute on is the DataSet or DataStream of the endpoint or
of the CamelFlinkDataSet or CamelFlinkDataStream header" },
- "producerOut": { "carriers": [ "unknown" ], "description": "What the
callback returns, null for a void callback. A DataSet it returns is collected
into a List; a DataStream it returns fails unless collect is false, then the
DataStream (or DataSet) is the body and the CamelFlinkDataStream (or
CamelFlinkDataSet) header" }
+ "producerOut": { "carriers": [ "unknown" ], "description": "What the
callback returns, null for a void callback. A DataSet it returns is collected
into a List unless collect is false; a DataStream it returns is not collected
(it fails if collect is explicitly set to true). A DataSet or DataStream that
is not collected is the body and the CamelFlinkDataSet or CamelFlinkDataStream
header" }
},
"properties": {
"endpointType": { "index": 0, "kind": "path", "displayName": "Endpoint
Type", "group": "producer", "label": "", "required": true, "type": "enum",
"javaType": "org.apache.camel.component.flink.EndpointType", "enum": [
"dataset", "datastream" ], "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "Type of the endpoint
(dataset, datastream)." },
- "collect": { "index": 1, "kind": "parameter", "displayName": "Collect",
"group": "producer", "label": "", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": true, "description": "Indicates if results should be
collected or counted." },
+ "collect": { "index": 1, "kind": "parameter", "displayName": "Collect",
"group": "producer", "label": "", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": true, "description": "Indicates if results should be
collected or counted. A DataStream returned by a datastream callback cannot be
collected: it is returned as is, unless collect is explicitly set to true,
which fails." },
"dataSet": { "index": 2, "kind": "parameter", "displayName": "Data Set",
"group": "producer", "label": "", "required": false, "type": "object",
"javaType": "org.apache.flink.api.java.DataSet", "deprecated": false,
"autowired": false, "secret": false, "description": "DataSet to compute
against." },
"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." },
diff --git a/components/camel-flink/src/main/docs/flink-component.adoc
b/components/camel-flink/src/main/docs/flink-component.adoc
index fef2e53d0a9b..0ccc19741104 100644
--- a/components/camel-flink/src/main/docs/flink-component.adoc
+++ b/components/camel-flink/src/main/docs/flink-component.adoc
@@ -373,7 +373,8 @@ YAML::
|collect
|Boolean
|true
-|Whether to collect results (not applicable for unbounded streams).
+|Whether to collect the DataSet returned by a dataset callback into a List. A
DataStream returned by a datastream callback is
+not collected: it is the message body and the `CamelFlinkDataStream` header,
and an explicit `collect=true` fails.
|===
diff --git
a/components/camel-flink/src/main/java/org/apache/camel/component/flink/DataStreamFlinkProducer.java
b/components/camel-flink/src/main/java/org/apache/camel/component/flink/DataStreamFlinkProducer.java
index 5ab877b20e8f..241db5a23509 100644
---
a/components/camel-flink/src/main/java/org/apache/camel/component/flink/DataStreamFlinkProducer.java
+++
b/components/camel-flink/src/main/java/org/apache/camel/component/flink/DataStreamFlinkProducer.java
@@ -75,8 +75,10 @@ public class DataStreamFlinkProducer extends DefaultProducer
{
protected void collectResults(Exchange exchange, Object result) {
if (result instanceof DataStream) {
- if (getEndpoint().isCollect()) {
- throw new IllegalArgumentException("collect mode not supported
for Flink DataStreams.");
+ // collect defaults to true for the DataSet API; a DataStream is
only rejected when collect was asked for
+ if (getEndpoint().isCollect() &&
getEndpoint().isCollectExplicit()) {
+ throw new IllegalArgumentException(
+ "collect mode not supported for Flink DataStreams. Set
collect=false to return the DataStream.");
} else {
exchange.getIn().setBody(result);
exchange.getIn().setHeader(FlinkConstants.FLINK_DATASTREAM_HEADER, result);
diff --git
a/components/camel-flink/src/main/java/org/apache/camel/component/flink/FlinkEndpoint.java
b/components/camel-flink/src/main/java/org/apache/camel/component/flink/FlinkEndpoint.java
index 8dfb4bd02d67..6bd330ad4f34 100644
---
a/components/camel-flink/src/main/java/org/apache/camel/component/flink/FlinkEndpoint.java
+++
b/components/camel-flink/src/main/java/org/apache/camel/component/flink/FlinkEndpoint.java
@@ -43,9 +43,10 @@ import org.apache.flink.streaming.api.datastream.DataStream;
+ " header"),
producerOut = @BodyType(value = Carrier.UNKNOWN,
description = "What the callback returns,
null for a void callback. A DataSet it returns is"
- + " collected into a List; a
DataStream it returns fails unless collect is false,"
- + " then the DataStream (or
DataSet) is the body and the CamelFlinkDataStream (or"
- + " CamelFlinkDataSet)
header"))
+ + " collected into a List
unless collect is false; a DataStream it returns is not"
+ + " collected (it fails if
collect is explicitly set to true). A DataSet or"
+ + " DataStream that is not
collected is the body and the CamelFlinkDataSet or"
+ + " CamelFlinkDataStream
header"))
public class FlinkEndpoint extends DefaultEndpoint {
@UriPath
@@ -61,6 +62,7 @@ public class FlinkEndpoint extends DefaultEndpoint {
private DataStreamCallback dataStreamCallback;
@UriParam(defaultValue = "true")
private boolean collect = true;
+ private boolean collectExplicit;
@UriParam(label = "producer,advanced", enums = "STREAMING,BATCH,AUTOMATIC")
private String executionMode;
@UriParam(label = "producer,advanced")
@@ -94,6 +96,14 @@ public class FlinkEndpoint extends DefaultEndpoint {
if (dataSetCallback == null) {
dataSetCallback = getComponent().getDataSetCallback();
}
+
+ if (dataStream == null) {
+ dataStream = getComponent().getDataStream();
+ }
+
+ if (dataStreamCallback == null) {
+ dataStreamCallback = getComponent().getDataStreamCallback();
+ }
}
@Override
@@ -179,10 +189,19 @@ public class FlinkEndpoint extends DefaultEndpoint {
}
/**
- * Indicates if results should be collected or counted.
+ * Indicates if results should be collected or counted. A DataStream
returned by a datastream callback cannot be
+ * collected: it is returned as is, unless collect is explicitly set to
true, which fails.
*/
public void setCollect(boolean collect) {
this.collect = collect;
+ this.collectExplicit = true;
+ }
+
+ /**
+ * Whether collect was explicitly set on the endpoint.
+ */
+ boolean isCollectExplicit() {
+ return collectExplicit;
}
public String getExecutionMode() {
diff --git
a/components/camel-flink/src/test/java/org/apache/camel/component/flink/DataStreamComponentOptionsAndCollectTest.java
b/components/camel-flink/src/test/java/org/apache/camel/component/flink/DataStreamComponentOptionsAndCollectTest.java
new file mode 100644
index 000000000000..48cd47530145
--- /dev/null
+++
b/components/camel-flink/src/test/java/org/apache/camel/component/flink/DataStreamComponentOptionsAndCollectTest.java
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.flink;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.Exchange;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class DataStreamComponentOptionsAndCollectTest extends CamelTestSupport {
+
+ static StreamExecutionEnvironment streamExecutionEnvironment =
Flinks.createStreamExecutionEnvironment();
+
+ @BindToRegistry("myDataStream")
+ private DataStreamSource<String> dss =
streamExecutionEnvironment.readTextFile("src/test/resources/testds.txt");
+
+ @Test
+ void shouldUseDataStreamAndCallbackOfComponent() {
+ AtomicReference<DataStream> used = new AtomicReference<>();
+ FlinkComponent flink = context.getComponent("flink",
FlinkComponent.class);
+ flink.setDataStream(dss);
+ flink.setDataStreamCallback(new VoidDataStreamCallback() {
+ @Override
+ public void doOnDataStream(DataStream ds, Object... payloads) {
+ used.set(ds);
+ }
+ });
+
+ Exchange result = template.send("flink:datastream", exchange ->
exchange.getIn().setBody("hello"));
+
+ assertThat(result.getException()).isNull();
+ assertThat(used.get()).isSameAs(dss);
+ }
+
+ @Test
+ void shouldReturnDataStreamWithDefaultCollect() {
+ Exchange result =
template.send("flink:datastream?dataStream=#myDataStream",
+ exchange ->
exchange.getIn().setHeader(FlinkConstants.FLINK_DATASTREAM_CALLBACK_HEADER,
+ (DataStreamCallback<DataStream>) (ds, payloads) ->
ds.rebalance()));
+
+ assertThat(result.getException()).isNull();
+
assertThat(result.getMessage().getBody()).isInstanceOf(DataStream.class).isNotSameAs(dss);
+
assertThat(result.getMessage().getHeader(FlinkConstants.FLINK_DATASTREAM_HEADER))
+ .isSameAs(result.getMessage().getBody());
+ }
+
+ @Test
+ void shouldReturnDataStreamWithCollectFalse() {
+ Exchange result =
template.send("flink:datastream?dataStream=#myDataStream&collect=false",
+ exchange ->
exchange.getIn().setHeader(FlinkConstants.FLINK_DATASTREAM_CALLBACK_HEADER,
+ (DataStreamCallback<DataStream>) (ds, payloads) ->
ds.rebalance()));
+
+ assertThat(result.getException()).isNull();
+
assertThat(result.getMessage().getBody()).isInstanceOf(DataStream.class);
+ }
+
+ @Test
+ void shouldFailToReturnDataStreamWithExplicitCollect() {
+ Exchange result =
template.send("flink:datastream?dataStream=#myDataStream&collect=true",
+ exchange ->
exchange.getIn().setHeader(FlinkConstants.FLINK_DATASTREAM_CALLBACK_HEADER,
+ (DataStreamCallback<DataStream>) (ds, payloads) ->
ds.rebalance()));
+
+
assertThat(result.getException()).isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("collect mode not supported for Flink
DataStreams");
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 603bc21117e4..b0657a286b5f 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -830,6 +830,14 @@ as the body, so `+${body}+` was the file found. A
`fileName` that uses the file
before. A `fileName` that read the file found through its headers, such as
`${header.CamelFileName}`, must use the file
functions instead (`${file:name}`).
+=== camel-flink - the datastream endpoint returns a DataStream with the
default collect option
+
+A `flink:datastream` endpoint whose callback returns a `DataStream` no longer
fails with "collect mode not supported for
+Flink DataStreams." when `collect` is not set: the DataStream is the message
body and the `CamelFlinkDataStream`
+header, as with `collect=false`. An explicit `collect=true` still fails. The
`dataStream` and `dataStreamCallback`
+options of the flink component are now used by endpoints that do not set them,
as `dataSet` and `dataSetCallback`
+already were.
+
=== camel-mina-sftp - the local work file is deleted when the exchange is done
With `localWorkDirectory`, camel-mina-sftp now deletes the local work file of
a consumed file when the exchange is done,
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 2b51e62ec50c..8bea44fcdf65 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
@@ -45,7 +45,9 @@ public interface FlinkEndpointBuilderFactory {
}
/**
- * Indicates if results should be collected or counted.
+ * Indicates if results should be collected or counted. A DataStream
+ * returned by a datastream callback cannot be collected: it is
returned
+ * as is, unless collect is explicitly set to true, which fails.
*
* The option is a: <code>boolean</code> type.
*
@@ -60,7 +62,9 @@ public interface FlinkEndpointBuilderFactory {
return this;
}
/**
- * Indicates if results should be collected or counted.
+ * Indicates if results should be collected or counted. A DataStream
+ * returned by a datastream callback cannot be collected: it is
returned
+ * as is, unless collect is explicitly set to true, which fails.
*
* The option will be converted to a <code>boolean</code> type.
*
