This is an automated email from the ASF dual-hosted git repository.
loserwang1024 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 9ea8bf028 [FLINK-41028] Support data lake table properties in the
Fluss pipeline sink (#4584)
9ea8bf028 is described below
commit 9ea8bf0283ad774a56062ccf545168304c5c90d0
Author: Hongshun Wang <[email protected]>
AuthorDate: Fri Oct 23 17:37:28 2026 +0800
[FLINK-41028] Support data lake table properties in the Fluss pipeline sink
(#4584)
---
.../docs/connectors/pipeline-connectors/fluss.md | 32 ++++++++
.../docs/connectors/pipeline-connectors/fluss.md | 33 ++++++++
.../fluss/factory/FlussDataSinkFactory.java | 25 ++++--
.../connectors/fluss/utils/FlussConversions.java | 26 ++++--
.../fluss/factory/FlussDataSinkFactoryTest.java | 95 ++++++++++++++++++++--
.../fluss/sink/FlussMetadataApplierTest.java | 31 +++++++
.../fluss/utils/FlussConversionsTest.java | 10 ++-
7 files changed, 234 insertions(+), 18 deletions(-)
diff --git a/docs/content.zh/docs/connectors/pipeline-connectors/fluss.md
b/docs/content.zh/docs/connectors/pipeline-connectors/fluss.md
index 2d21ae1c7..10a6d553e 100644
--- a/docs/content.zh/docs/connectors/pipeline-connectors/fluss.md
+++ b/docs/content.zh/docs/connectors/pipeline-connectors/fluss.md
@@ -149,6 +149,13 @@ Pipeline Connector Options
<td>String</td>
<td>将 Fluss table 支持的参数传递给 pipeline，参考，See <a
href="https://fluss.apache.org/docs/engine-flink/options/#storage-options";>Fluss
table options</a>。</td>
</tr>
+ <tr>
+ <td>properties.<data-lake-format>.*</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>建表时向 Fluss 传递数据湖表参数。支持的前缀跟随 Fluss 客户端的数据湖格式，当前包括
<code>properties.paimon.*</code>、<code>properties.iceberg.*</code>、<code>properties.hudi.*</code>
和 <code>properties.lance.*</code>。仅移除最外层的 <code>properties.</code> 前缀，Flink
CDC 不校验其中的具体参数。</td>
+ </tr>
<tr>
<td>properties.client.*</td>
<td>optional</td>
@@ -186,6 +193,31 @@ sink:
* 支持 Fluss 主键表和日志表。
+### 数据湖表参数
+
+要自动创建启用湖存储的 Fluss 表，可以使用以下 Sink 配置：
+
+```yaml
+sink:
+ type: fluss
+ bootstrap.servers: localhost:9123
+ properties.table.datalake.enabled: 'true'
+ properties.table.datalake.freshness: 30s
+ properties.paimon.file.format: orc
+ properties.paimon.deletion-vectors.enabled: 'true'
+```
+
+连接器仅移除 `properties.`，因此 `properties.paimon.file.format` 会作为
+`paimon.file.format` 传入 Fluss 表参数。使用 Iceberg 时，配置 `properties.iceberg.*`，
+例如 `properties.iceberg.write.format.default: parquet`。
+数据湖格式和分层服务需要在 Fluss 中配置，详见
+[Fluss
湖存储文档](https://fluss.apache.org/docs/streaming-lakehouse/datalake-formats/paimon/)。
+
+这些参数对该 Sink 创建的所有表生效，不会修改已有表的属性。
+Flink CDC 允许支持的数据湖前缀下的任意参数名和值，由 Fluss 和对应的湖集成负责解释和校验；
+如果下游实现不拒绝未知参数，拼写错误也可能被忽略。
+不支持的前缀（例如 `properties.icerberg.*`）仍会被 Flink CDC 拒绝。
+
### 动态 Source 订阅
当 Fluss 作为带表发现器的 Source 使用时，每次成功发现的结果都是当前订阅表的完整权威集合。配置正数
diff --git a/docs/content/docs/connectors/pipeline-connectors/fluss.md
b/docs/content/docs/connectors/pipeline-connectors/fluss.md
index 6526fd00e..64f4a77f7 100644
--- a/docs/content/docs/connectors/pipeline-connectors/fluss.md
+++ b/docs/content/docs/connectors/pipeline-connectors/fluss.md
@@ -152,6 +152,13 @@ Pipeline Connector Options
<td>String</td>
<td>Pass options of Fluss table to pipeline. See <a
href="https://fluss.apache.org/docs/engine-flink/options/#storage-options";>Fluss
table options</a>. </td>
</tr>
+ <tr>
+ <td>properties.<data-lake-format>.*</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Pass data lake table properties to Fluss when creating tables.
Supported prefixes follow the Fluss client's data lake formats, currently
<code>properties.paimon.*</code>, <code>properties.iceberg.*</code>,
<code>properties.hudi.*</code>, and <code>properties.lance.*</code>. Only the
outer <code>properties.</code> prefix is removed. Individual options are not
validated by Flink CDC.</td>
+ </tr>
<tr>
<td>properties.client.*</td>
<td>optional</td>
@@ -191,6 +198,32 @@ Continue to configure bucket keys and bucket counts with
the sink options `bucke
* Support Fluss primary key table and log table.
+### Data lake table properties
+
+To create Fluss tables with lakehouse storage enabled, configure the Fluss
sink as follows:
+
+```yaml
+sink:
+ type: fluss
+ bootstrap.servers: localhost:9123
+ properties.table.datalake.enabled: 'true'
+ properties.table.datalake.freshness: 30s
+ properties.paimon.file.format: orc
+ properties.paimon.deletion-vectors.enabled: 'true'
+```
+
+The connector removes only `properties.`, so `properties.paimon.file.format`
becomes
+`paimon.file.format` in the Fluss table properties. For Iceberg, use
`properties.iceberg.*`,
+for example `properties.iceberg.write.format.default: parquet`.
+The lake format and tiering service must be configured in Fluss; see the
+[Fluss lakehouse
documentation](https://fluss.apache.org/docs/streaming-lakehouse/datalake-formats/paimon/).
+
+These properties apply to all tables created by this sink. They do not alter
properties of
+existing tables. Flink CDC accepts arbitrary option names and values under
supported data lake
+prefixes and delegates their interpretation and validation to Fluss and the
lake integration;
+an unrecognized option may be ignored if the downstream implementation does
not reject it.
+Unsupported prefixes, such as `properties.icerberg.*`, are still rejected by
Flink CDC.
+
### Dynamic source subscriptions
When Fluss is used as a source with a table discoverer, each successful
discovery result is the
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/factory/FlussDataSinkFactory.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/factory/FlussDataSinkFactory.java
index 455bbdc17..d3797feee 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/factory/FlussDataSinkFactory.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/factory/FlussDataSinkFactory.java
@@ -25,12 +25,15 @@ import
org.apache.flink.cdc.connectors.fluss.sink.FlussDataSink;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.metadata.DataLakeFormat;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.stream.Stream;
import static
org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.BOOTSTRAP_SERVERS;
import static
org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.BUCKET_KEY;
@@ -45,10 +48,22 @@ import static
org.apache.flink.cdc.connectors.fluss.utils.FlussConfigUtils.parse
public class FlussDataSinkFactory implements DataSinkFactory {
public static final String IDENTIFIER = "fluss";
+ private static final String PROPERTIES_PREFIX = "properties.";
+ private static final String[] TABLE_PROPERTIES_PREFIXES =
+ Stream.concat(
+ Stream.of(TABLE_PROPERTIES_PREFIX),
+ Arrays.stream(DataLakeFormat.values())
+ .map(format -> PROPERTIES_PREFIX + format
+ "."))
+ .toArray(String[]::new);
+
@Override
public DataSink createDataSink(Context context) {
FactoryHelper.createFactoryHelper(this, context)
- .validateExcept(CLIENT_PROPERTIES_PREFIX,
TABLE_PROPERTIES_PREFIX);
+ .validateExcept(
+ Stream.concat(
+ Stream.of(CLIENT_PROPERTIES_PREFIX),
+
Arrays.stream(TABLE_PROPERTIES_PREFIXES))
+ .toArray(String[]::new));
org.apache.flink.cdc.common.configuration.Configuration
factoryConfiguration =
context.getFactoryConfiguration();
@@ -96,7 +111,7 @@ public class FlussDataSinkFactory implements DataSinkFactory
{
tableOptions.forEach(
(key, value) -> {
if (key.startsWith(CLIENT_PROPERTIES_PREFIX)) {
-
flussConfig.setString(key.substring("properties.".length()), value);
+
flussConfig.setString(key.substring(PROPERTIES_PREFIX.length()), value);
}
});
return flussConfig;
@@ -104,11 +119,11 @@ public class FlussDataSinkFactory implements
DataSinkFactory {
private static Map<String, String> toFlussTableOptions(Map<String, String>
tableOptions) {
Map<String, String> flussTableConfig = new HashMap<>();
- // forward all client configs
+ // Forward Fluss and data lake table properties without validating
individual options.
tableOptions.forEach(
(key, value) -> {
- if (key.startsWith(TABLE_PROPERTIES_PREFIX)) {
-
flussTableConfig.put(key.substring("properties.".length()), value);
+ if
(Arrays.stream(TABLE_PROPERTIES_PREFIXES).anyMatch(key::startsWith)) {
+
flussTableConfig.put(key.substring(PROPERTIES_PREFIX.length()), value);
}
});
return flussTableConfig;
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/utils/FlussConversions.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/utils/FlussConversions.java
index 35bef53a4..23dcc0f9f 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/utils/FlussConversions.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/utils/FlussConversions.java
@@ -42,12 +42,14 @@ import org.apache.flink.cdc.common.types.ZonedTimestampType;
import org.apache.flink.cdc.common.utils.Preconditions;
import org.apache.flink.util.CollectionUtil;
+import org.apache.fluss.metadata.DataLakeFormat;
import org.apache.fluss.metadata.Schema;
import org.apache.fluss.metadata.TableDescriptor;
import javax.annotation.Nullable;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -177,13 +179,23 @@ public class FlussConversions {
.orElse(Collections.emptyList());
}
- return TableDescriptor.builder()
- .schema(schema)
- .partitionedBy(cdcSchema.partitionKeys())
- .distributedBy(bucketNum, bucketKeys)
- .comment(cdcSchema.comment())
- .properties(tableProperties)
- .build();
+ TableDescriptor.Builder builder =
+ TableDescriptor.builder()
+ .schema(schema)
+ .partitionedBy(cdcSchema.partitionKeys())
+ .distributedBy(bucketNum, bucketKeys)
+ .comment(cdcSchema.comment());
+ tableProperties.forEach(
+ (key, value) -> {
+ if (Arrays.stream(DataLakeFormat.values())
+ .anyMatch(format -> key.startsWith(format + ".")))
{
+ // Lake options must use custom properties to pass
Fluss table validation.
+ builder.customProperty(key, value);
+ } else {
+ builder.property(key, value);
+ }
+ });
+ return builder.build();
}
public static org.apache.fluss.metadata.Schema toFlussSchema(
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/factory/FlussDataSinkFactoryTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/factory/FlussDataSinkFactoryTest.java
index 7b287d122..3a75fbcb5 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/factory/FlussDataSinkFactoryTest.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/factory/FlussDataSinkFactoryTest.java
@@ -33,11 +33,17 @@ import
org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test for {@link FlussDataSinkFactory}. */
-public class FlussDataSinkFactoryTest {
+class FlussDataSinkFactoryTest {
@Test
void testCreateDataSink() {
DataSinkFactory sinkFactory =
@@ -70,8 +76,20 @@ public class FlussDataSinkFactoryTest {
.isInstanceOf(ForwardHashFunctionProvider.class);
}
- @Test
- void testUnsupportedOption() {
+ @ParameterizedTest
+ @ValueSource(
+ strings = {
+ "unsupported_key",
+ "properties.unsupported.option",
+ "properties.paimon",
+ "properties.paimonx.option",
+ "properties.iceberg",
+ "properties.icebergx.option",
+ "properties.icerberg.option",
+ "paimon.file.format",
+ "iceberg.write.format.default"
+ })
+ void testUnsupportedOption(String option) {
DataSinkFactory sinkFactory =
FactoryDiscoveryUtils.getFactoryByIdentifier("fluss",
DataSinkFactory.class);
Assertions.assertThat(sinkFactory).isInstanceOf(FlussDataSinkFactory.class);
@@ -80,7 +98,7 @@ public class FlussDataSinkFactoryTest {
Configuration.fromMap(
ImmutableMap.<String, String>builder()
.put(FlussDataSinkOptions.BOOTSTRAP_SERVERS.key(), "localhost:9123")
- .put("unsupported_key", "unsupported_value")
+ .put(option, "unsupported_value")
.build());
Assertions.assertThatThrownBy(
@@ -94,7 +112,74 @@ public class FlussDataSinkFactoryTest {
.hasMessageContaining(
"Unsupported options found for 'fluss'.\n\n"
+ "Unsupported options:\n\n"
- + "unsupported_key");
+ + option);
+ }
+
+ @ParameterizedTest
+ @CsvSource({
+ "paimon.file.format, orc",
+ "paimon.deletion-vectors.enabled, true",
+ "paimon.unknown-option, unknown-value",
+ "iceberg.write.format.default, parquet",
+ "iceberg.unknown-option, unknown-value",
+ "hudi.unknown-option, unknown-value",
+ "lance.unknown-option, unknown-value"
+ })
+ void testDataLakeProperties(String option, String value) {
+ Configuration conf =
+ Configuration.fromMap(
+ ImmutableMap.<String, String>builder()
+
.put(FlussDataSinkOptions.BOOTSTRAP_SERVERS.key(), "localhost:9123")
+ .put("properties.table.datalake.enabled",
"true")
+ .put("properties.table.datalake.freshness",
"30s")
+ .put("properties." + option, value)
+ .put("properties.client.writer.acks", "all")
+ .build());
+
+ DataSink dataSink =
+ new FlussDataSinkFactory()
+ .createDataSink(
+ new FactoryHelper.DefaultContext(
+ conf,
+ conf,
+
Thread.currentThread().getContextClassLoader()));
+
+ Map<String, String> expectedTableProperties =
+ ImmutableMap.of(
+ "table.datalake.enabled",
+ "true",
+ "table.datalake.freshness",
+ "30s",
+ option,
+ value);
+ assertThat(dataSink.getMetadataApplier())
+ .extracting("tableProperties")
+ .isEqualTo(expectedTableProperties);
+ assertThat(dataSink)
+ .extracting("flussClientConfig")
+ .isEqualTo(
+ org.apache.fluss.config.Configuration.fromMap(
+ ImmutableMap.of(
+ "bootstrap.servers", "localhost:9123",
+ "client.writer.acks", "all")));
+ }
+
+ @Test
+ void testDataLakePropertiesRequireBootstrapServers() {
+ Configuration conf =
+
Configuration.fromMap(ImmutableMap.of("properties.paimon.file.format", "orc"));
+
+ assertThatThrownBy(
+ () ->
+ new FlussDataSinkFactory()
+ .createDataSink(
+ new
FactoryHelper.DefaultContext(
+ conf,
+ conf,
+ Thread.currentThread()
+
.getContextClassLoader())))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("Missing required options
are:\n\nbootstrap.servers");
}
@Test
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetadataApplierTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetadataApplierTest.java
index ed2ab81fc..2fb83b0b5 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetadataApplierTest.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetadataApplierTest.java
@@ -1174,6 +1174,37 @@ class FlussMetadataApplierTest {
}
}
+ @Test
+ void testCreateTableWithDataLakeProperties() throws Exception {
+ Map<String, String> tableProperties = new HashMap<>();
+ tableProperties.put("paimon.file.format", "orc");
+ tableProperties.put("iceberg.write.format.default", "parquet");
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.BIGINT().notNull())
+ .primaryKey("id")
+ .build();
+ TableId tableId = TableId.tableId(DATABASE_NAME, "table1");
+
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ tableProperties,
+ Collections.emptyMap(),
+ Collections.emptyMap())) {
+ applier.applySchemaChange(new CreateTableEvent(tableId, schema));
+
+ TableDescriptor tableDescriptor =
+ admin.getTableInfo(new TablePath(DATABASE_NAME, "table1"))
+ .get()
+ .toTableDescriptor();
+ assertThat(tableDescriptor.getCustomProperties())
+ .containsExactlyInAnyOrderEntriesOf(tableProperties);
+ assertThat(tableDescriptor.getProperties())
+ .doesNotContainKeys("paimon.file.format",
"iceberg.write.format.default");
+ }
+ }
+
@Test
void testCreateTableWithTableOptions() throws Exception {
Schema schema =
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/utils/FlussConversionsTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/utils/FlussConversionsTest.java
index ba5535477..8a9b3798f 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/utils/FlussConversionsTest.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/utils/FlussConversionsTest.java
@@ -219,11 +219,19 @@ class FlussConversionsTest {
Map<String, String> properties = new HashMap<>();
properties.put("table.replication.factor", "3");
+ properties.put("paimon.file.format", "orc");
+ properties.put("iceberg.write.format.default", "parquet");
TableDescriptor descriptor =
FlussConversions.toFlussTable(cdcSchema, null, null,
properties);
-
assertThat(descriptor.getProperties()).containsEntry("table.replication.factor",
"3");
+ assertThat(descriptor.getProperties())
+ .containsExactlyEntriesOf(
+ Collections.singletonMap("table.replication.factor",
"3"));
+ assertThat(descriptor.getCustomProperties())
+ .hasSize(2)
+ .containsEntry("paimon.file.format", "orc")
+ .containsEntry("iceberg.write.format.default", "parquet");
}
@Test
