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 24d08ea96 [FLINK-41058][pipeline-connector/fluss] Merge schema options
with global sink table properties when creating tables. (#4587)
24d08ea96 is described below
commit 24d08ea96ad5461c6da8e5cd7892e5956958765b
Author: Hongshun Wang <[email protected]>
AuthorDate: Wed Oct 21 17:28:06 2026 +0800
[FLINK-41058][pipeline-connector/fluss] Merge schema options with global
sink table properties when creating tables. (#4587)
---
.../docs/connectors/pipeline-connectors/fluss.md | 22 ++++++
.../docs/connectors/pipeline-connectors/fluss.md | 24 +++++++
.../fluss/sink/FlussMetaDataApplier.java | 5 +-
.../fluss/sink/FlussMetadataApplierTest.java | 79 +++++++++++++++++++++-
4 files changed, 128 insertions(+), 2 deletions(-)
diff --git a/docs/content.zh/docs/connectors/pipeline-connectors/fluss.md
b/docs/content.zh/docs/connectors/pipeline-connectors/fluss.md
index 9196e9024..d0399225b 100644
--- a/docs/content.zh/docs/connectors/pipeline-connectors/fluss.md
+++ b/docs/content.zh/docs/connectors/pipeline-connectors/fluss.md
@@ -160,6 +160,28 @@ Pipeline Connector Options
</table>
</div>
+### 在 transform 规则中配置表参数
+
+可以通过 transform 规则中的 `table-options` 为匹配的表配置 Fluss 表参数。
+参数名直接使用 `table.replication.factor` 等 Fluss 表属性名称，无需添加 sink 配置中的 `properties.`
前缀。
+这些参数会覆盖同名的 `sink.properties.table.*` 配置，未覆盖的 sink 表参数仍然生效。
+
+```yaml
+transform:
+ - source-table: inventory.products
+ projection: "*"
+ table-options: table.replication.factor=2,table.log.ttl=7d
+
+sink:
+ type: fluss
+ bootstrap.servers: localhost:9123
+ properties.table.replication.factor: 3
+```
+
+在此示例中，为 `inventory.products` 创建的 Fluss 表副本数为 2，日志保留时间为 7 天，其他表的副本数为 3。
+这些参数仅在建表时生效，不会修改已有表的属性。
+分桶键和桶数量仍通过 sink 的 `bucket.key` 和 `bucket.num` 配置。
+
## 使用说明
* 支持 Fluss 主键表和日志表。
diff --git a/docs/content/docs/connectors/pipeline-connectors/fluss.md
b/docs/content/docs/connectors/pipeline-connectors/fluss.md
index 041813738..e43bee43b 100644
--- a/docs/content/docs/connectors/pipeline-connectors/fluss.md
+++ b/docs/content/docs/connectors/pipeline-connectors/fluss.md
@@ -163,6 +163,30 @@ Pipeline Connector Options
</table>
</div>
+### Table options in transform rules
+
+Use `table-options` in a transform rule to configure Fluss table properties
for the matched table.
+Use Fluss property names such as `table.replication.factor`, without the
`properties.` prefix used
+in the sink configuration. These options override matching
`sink.properties.table.*` options;
+other sink table properties still apply.
+
+```yaml
+transform:
+ - source-table: inventory.products
+ projection: "*"
+ table-options: table.replication.factor=2,table.log.ttl=7d
+
+sink:
+ type: fluss
+ bootstrap.servers: localhost:9123
+ properties.table.replication.factor: 3
+```
+
+In this example, the Fluss table created for `inventory.products` has a
replication factor of 2
+and a log retention time of 7 days. Other tables use a replication factor of 3.
+These options apply when creating tables and do not update the properties of
existing tables.
+Continue to configure bucket keys and bucket counts with the sink options
`bucket.key` and `bucket.num`.
+
## Usage Notes
* Support Fluss primary key table and log table.
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetaDataApplier.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetaDataApplier.java
index 3ec152430..3f6c9fc45 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetaDataApplier.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetaDataApplier.java
@@ -50,6 +50,7 @@ import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -171,8 +172,10 @@ public class FlussMetaDataApplier implements
MetadataApplier, ExistingTableSchem
String tableIdentifier = tablePath.getDatabaseName() + "." +
tablePath.getTableName();
List<String> bucketKeys = bucketKeysMap.get(tableIdentifier);
Integer bucketNum = bucketNumMap.get(tableIdentifier);
+ Map<String, String> fullTableProperties = new
HashMap<>(tableProperties);
+ fullTableProperties.putAll(event.getSchema().options());
TableDescriptor inferredFlussTable =
- toFlussTable(event.getSchema(), bucketKeys, bucketNum,
tableProperties);
+ toFlussTable(event.getSchema(), bucketKeys, bucketNum,
fullTableProperties);
admin.createDatabase(tablePath.getDatabaseName(),
DatabaseDescriptor.EMPTY, true);
if (!admin.tableExists(tablePath).get()) {
admin.createTable(tablePath, inferredFlussTable, false).get();
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 1faf881f8..ed2ab81fc 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
@@ -57,6 +57,7 @@ import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.params.provider.ValueSource;
import java.lang.reflect.Proxy;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -71,13 +72,15 @@ import java.util.stream.Stream;
import static
org.apache.flink.cdc.connectors.fluss.utils.FlussConversions.toFlussSchema;
import static
org.apache.flink.cdc.connectors.fluss.utils.FlussConversions.toFlussType;
+import static org.apache.fluss.config.ConfigOptions.DEFAULT_REPLICATION_FACTOR;
+import static org.apache.fluss.config.ConfigOptions.TABLE_LOG_TTL;
import static org.apache.fluss.config.ConfigOptions.TABLE_REPLICATION_FACTOR;
import static org.apache.fluss.types.DataTypeChecks.equalsWithFieldId;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test for {@link FlussMetaDataApplier}. */
-public class FlussMetadataApplierTest {
+class FlussMetadataApplierTest {
@RegisterExtension
private static final FlussClusterExtension FLUSS_CLUSTER_EXTENSION =
FlussClusterExtension.builder().setNumOfTabletServers(3).build();
@@ -1204,4 +1207,78 @@ public class FlussMetadataApplierTest {
assertThat(tableInfo.getProperties().get(TABLE_REPLICATION_FACTOR)).isEqualTo(5);
}
}
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testCreateTableWithSchemaOptions(boolean hasSinkOptions) throws
Exception {
+ Map<String, String> tableProperties =
+ hasSinkOptions
+ ? Map.of(TABLE_REPLICATION_FACTOR.key(), "3",
TABLE_LOG_TTL.key(), "7 d")
+ : Collections.emptyMap();
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .primaryKey("id")
+ .option(TABLE_REPLICATION_FACTOR.key(), "2")
+ .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));
+ TableInfo tableInfo = admin.getTableInfo(new
TablePath(DATABASE_NAME, "table1")).get();
+
assertThat(tableInfo.getProperties().get(TABLE_REPLICATION_FACTOR)).isEqualTo(2);
+ if (hasSinkOptions) {
+ assertThat(tableInfo.getProperties().get(TABLE_LOG_TTL))
+ .isEqualTo(Duration.ofDays(7));
+ }
+ assertThat(schema.options())
+ .containsExactlyEntriesOf(
+
Collections.singletonMap(TABLE_REPLICATION_FACTOR.key(), "2"));
+
+ Schema schemaWithoutOptions =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .primaryKey("id")
+ .build();
+ applier.applySchemaChange(
+ new CreateTableEvent(
+ TableId.tableId(DATABASE_NAME, "table2"),
schemaWithoutOptions));
+ TableInfo otherTableInfo =
+ admin.getTableInfo(new TablePath(DATABASE_NAME,
"table2")).get();
+
assertThat(otherTableInfo.getProperties().get(TABLE_REPLICATION_FACTOR))
+ .isEqualTo(hasSinkOptions ? 3 :
DEFAULT_REPLICATION_FACTOR.defaultValue());
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"key", "properties.table.replication.factor"})
+ void testCreateTableWithInvalidSchemaOptions(String optionKey) throws
Exception {
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .primaryKey("id")
+ .option(optionKey, "2")
+ .build();
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap())) {
+ assertThatThrownBy(
+ () ->
+ applier.applySchemaChange(
+ new CreateTableEvent(
+
TableId.tableId(DATABASE_NAME, "table1"),
+ schema)))
+ .rootCause()
+ .isExactlyInstanceOf(InvalidConfigException.class)
+ .hasMessageContaining("'" + optionKey + "' is not a Fluss
table property");
+ assertThat(admin.tableExists(new TablePath(DATABASE_NAME,
"table1")).get()).isFalse();
+ }
+ }
}
