From f8a6738922eab55b5de499a574bcb41753f1e7fd Mon Sep 17 00:00:00 2001 From: Hongshun Wang Date: Mon, 19 Oct 2026 19:31:12 +0800 Subject: [PATCH] [FLINK-41028] Support data lake table properties in the Fluss pipeline sink Co-Authored-By: Codex AI-Model: gpt-6-astra AI-Contributed/Feature: 116/116 AI-Contributed/UT: 136/136 --- .../connectors/pipeline-connectors/fluss.md | 32 +++++++ .../connectors/pipeline-connectors/fluss.md | 33 +++++++ .../fluss/factory/FlussDataSinkFactory.java | 25 ++++- .../fluss/utils/FlussConversions.java | 26 +++-- .../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 9196e902484..cbe54ae3769 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 String 将 Fluss table 支持的参数传递给 pipeline，参考，See Fluss table options. + + properties.<data-lake-format>.* + optional + (none) + String + 建表时向 Fluss 传递数据湖表参数。支持的前缀跟随 Fluss 客户端的数据湖格式，当前包括 properties.paimon.*、properties.iceberg.*、properties.hudi.* 和 properties.lance.*。仅移除最外层的 properties. 前缀，Flink CDC 不校验其中的具体参数。 + properties.client.* optional @@ -164,6 +171,31 @@ Pipeline Connector Options * 支持 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 04181373844..5cb08605376 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 String Pass options of Fluss table to pipeline，See Fluss table options. + + properties.<data-lake-format>.* + optional + (none) + String + Pass data lake table properties to Fluss when creating tables. Supported prefixes follow the Fluss client's data lake formats, currently properties.paimon.*, properties.iceberg.*, properties.hudi.*, and properties.lance.*. Only the outer properties. prefix is removed. Individual options are not validated by Flink CDC. + properties.client.* optional @@ -167,6 +174,32 @@ Pipeline Connector Options * 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 455bbdc1717..d3797feee6c 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.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 @@ 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 @@ private static Configuration toFlussClientConfig(Map tableOption 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 @@ private static Configuration toFlussClientConfig(Map tableOption private static Map toFlussTableOptions(Map tableOptions) { Map 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 35bef53a4b1..23dcc0f9f53 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.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 static TableDescriptor toFlussTable( .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 7b287d122c9..3a75fbcb558 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.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 @@ void testCreateDataSinkWithForwardPartitioningStrategy() { .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 @@ void testUnsupportedOption() { Configuration.fromMap( ImmutableMap.builder() .put(FlussDataSinkOptions.BOOTSTRAP_SERVERS.key(), "localhost:9123") - .put("unsupported_key", "unsupported_value") + .put(option, "unsupported_value") .build()); Assertions.assertThatThrownBy( @@ -94,7 +112,74 @@ void testUnsupportedOption() { .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.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 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 dface17986a..0224031f626 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 @@ -729,6 +729,37 @@ void testExistingPrimaryKeyTableWithCustomBucketKeys(String bucketKeys) throws E } } + @Test + void testCreateTableWithDataLakeProperties() throws Exception { + Map 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 ba553547764..8a9b3798fdc 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 @@ void testToFlussTableWithProperties() { Map 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