[FLINK-41028] Support data lake table properties in the Fluss pipeline sink

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.&lt;data-lake-format&gt;.*</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>
@@ -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 使用时，每次成功发现的结果都是当前订阅表的完整权威集合。配置正数

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.&lt;data-lake-format&gt;.*</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>
@@ -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

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,19 +111,19 @@ private static Configuration toFlussClientConfig(Map<String, String> 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;
     }
 
     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;

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(

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.<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 @@ 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.<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

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<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 =

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