Skip to content

Commit b093f52

Browse files
loserwang1024loserwang1024
committed
expose table option and client option.
1 parent 3d73038 commit b093f52

7 files changed

Lines changed: 106 additions & 271 deletions

File tree

‎flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/factory/FlussDataSinkFactory.java‎

Lines changed: 35 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import org.apache.flink.cdc.common.sink.DataSink;
2424
import org.apache.flink.cdc.connectors.fluss.sink.FlussDataSink;
2525

26+
import com.alibaba.fluss.config.ConfigOptions;
2627
import com.alibaba.fluss.config.Configuration;
2728

2829
import java.util.HashMap;
@@ -36,37 +37,27 @@
3637
import static org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.BOOTSTRAP_SERVERS;
3738
import static org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.BUCKET_KEY;
3839
import static org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.BUCKET_NUMBER;
39-
import static org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.CLIENT_ID;
40-
import static org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.PREFIX_FLUSS_PROPERTIES;
40+
import static org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.CLIENT_PROPERTIES_PREFIX;
41+
import static org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.TABLE_PROPERTIES_PREFIX;
4142

4243
/** Factory for creating configured instances of {@link FlussDataSink}. */
4344
public class FlussDataSinkFactory implements DataSinkFactory {
4445
public static final String IDENTIFIER = "fluss";
4546

4647
@Override
4748
public DataSink createDataSink(Context context) {
48-
FactoryHelper.createFactoryHelper(this, context).validateExcept(PREFIX_FLUSS_PROPERTIES);
49+
FactoryHelper.createFactoryHelper(this, context)
50+
.validateExcept(CLIENT_PROPERTIES_PREFIX, TABLE_PROPERTIES_PREFIX);
4951
org.apache.flink.cdc.common.configuration.Configuration factoryConfiguration =
5052
context.getFactoryConfiguration();
5153

52-
Map<String, String> flussOptions = new HashMap<>();
53-
factoryConfiguration
54-
.toMap()
55-
.forEach(
56-
(key, value) -> {
57-
if (key.startsWith(PREFIX_FLUSS_PROPERTIES)) {
58-
flussOptions.put(
59-
key.substring(PREFIX_FLUSS_PROPERTIES.length()), value);
60-
} else {
61-
flussOptions.put(key, value);
62-
}
63-
});
64-
Configuration config = Configuration.fromMap(flussOptions);
54+
Configuration flussClientConfig = toFlussClientConfig(factoryConfiguration.toMap());
55+
Map<String, String> tableProperties = toFlussTableOptions(factoryConfiguration.toMap());
6556
Map<String, List<String>> bucketKeysMap =
6657
parseBucketKeys(factoryConfiguration.get(BUCKET_KEY));
6758
Map<String, Integer> bucketNumMap =
6859
parseBucketNumber(factoryConfiguration.get(BUCKET_NUMBER));
69-
return new FlussDataSink(config, bucketKeysMap, bucketNumMap);
60+
return new FlussDataSink(flussClientConfig, tableProperties, bucketKeysMap, bucketNumMap);
7061
}
7162

7263
@Override
@@ -86,7 +77,33 @@ public Set<ConfigOption<?>> optionalOptions() {
8677
Set<ConfigOption<?>> options = new HashSet<>();
8778
options.add(BUCKET_KEY);
8879
options.add(BUCKET_NUMBER);
89-
options.add(CLIENT_ID);
9080
return options;
9181
}
82+
83+
private static Configuration toFlussClientConfig(Map<String, String> tableOptions) {
84+
Configuration flussConfig = new Configuration();
85+
flussConfig.setString(
86+
ConfigOptions.BOOTSTRAP_SERVERS.key(), tableOptions.get(BOOTSTRAP_SERVERS.key()));
87+
88+
// forward all client configs
89+
tableOptions.forEach(
90+
(key, value) -> {
91+
if (key.startsWith(CLIENT_PROPERTIES_PREFIX)) {
92+
flussConfig.setString(key.substring("property.".length()), value);
93+
}
94+
});
95+
return flussConfig;
96+
}
97+
98+
private static Map<String, String> toFlussTableOptions(Map<String, String> tableOptions) {
99+
Map<String, String> flussTableConfig = new HashMap<>();
100+
// forward all client configs
101+
tableOptions.forEach(
102+
(key, value) -> {
103+
if (key.startsWith(TABLE_PROPERTIES_PREFIX)) {
104+
flussTableConfig.put(key.substring("property.".length()), value);
105+
}
106+
});
107+
return flussTableConfig;
108+
}
92109
}

‎flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussDataSink.java‎

Lines changed: 9 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -31,28 +31,32 @@
3131
/** A DataSink implementation for Fluss. */
3232
public class FlussDataSink implements DataSink {
3333

34-
private final Configuration flussConfig;
34+
private final Configuration flussClientConfig;
35+
private final Map<String, String> tableProperties;
3536
private final Map<String, List<String>> bucketKeysMap;
3637
private final Map<String, Integer> bucketNumMap;
3738

3839
public FlussDataSink(
39-
Configuration flussConfig,
40+
Configuration flussClientConfig,
41+
Map<String, String> tableProperties,
4042
Map<String, List<String>> bucketKeysMap,
4143
Map<String, Integer> bucketNumMap) {
42-
this.flussConfig = flussConfig;
44+
this.flussClientConfig = flussClientConfig;
45+
this.tableProperties = tableProperties;
4346
this.bucketKeysMap = bucketKeysMap;
4447
this.bucketNumMap = bucketNumMap;
4548
}
4649

4750
@Override
4851
public EventSinkProvider getEventSinkProvider() {
4952
return FlinkSinkProvider.of(
50-
new FlinkSink<>(flussConfig, new FlussEventSerializationSchema()));
53+
new FlinkSink<>(flussClientConfig, new FlussEventSerializationSchema()));
5154
}
5255

5356
@Override
5457
public MetadataApplier getMetadataApplier() {
5558

56-
return new FlussMetaDataApplier(flussConfig, bucketKeysMap, bucketNumMap);
59+
return new FlussMetaDataApplier(
60+
flussClientConfig, tableProperties, bucketKeysMap, bucketNumMap);
5761
}
5862
}

‎flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussDataSinkOptions.java‎

Lines changed: 2 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,8 @@
2424
public class FlussDataSinkOptions {
2525

2626
// prefix for passing properties for fluss options.
27-
public static final String PREFIX_FLUSS_PROPERTIES = "fluss.properties.";
27+
public static final String TABLE_PROPERTIES_PREFIX = "property.table.";
28+
public static final String CLIENT_PROPERTIES_PREFIX = "property.client.";
2829

2930
public static final ConfigOption<String> BOOTSTRAP_SERVERS =
3031
ConfigOptions.key("bootstrap.servers")
@@ -53,10 +54,4 @@ public class FlussDataSinkOptions {
5354
"The number of buckets of each Fluss table."
5455
+ "Tables are separated by ';'. "
5556
+ "Format: database1.table1:4;database1.table2:8.");
56-
57-
public static final ConfigOption<String> CLIENT_ID =
58-
ConfigOptions.key("client.id")
59-
.stringType()
60-
.defaultValue("")
61-
.withDescription("An id string to pass to the server when making requests.");
6257
}

‎flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetaDataApplier.java‎

Lines changed: 9 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -52,17 +52,20 @@
5252
/** {@link MetadataApplier} for fluss. */
5353
public class FlussMetaDataApplier implements MetadataApplier {
5454
private static final Logger LOG = LoggerFactory.getLogger(FlussMetaDataApplier.class);
55-
private final Configuration flussConfig;
55+
private final Configuration flussClientConfig;
56+
private final Map<String, String> tableProperties;
5657
private final Map<String, List<String>> bucketKeysMap;
5758
private final Map<String, Integer> bucketNumMap;
5859
private Set<SchemaChangeEventType> enabledEventTypes =
5960
new HashSet<>(Arrays.asList(CREATE_TABLE, DROP_TABLE));
6061

6162
public FlussMetaDataApplier(
62-
Configuration flussConfig,
63+
Configuration flussClientConfig,
64+
Map<String, String> tableProperties,
6365
Map<String, List<String>> bucketKeysMap,
6466
Map<String, Integer> bucketNumMap) {
65-
this.flussConfig = flussConfig;
67+
this.flussClientConfig = flussClientConfig;
68+
this.tableProperties = tableProperties;
6669
this.bucketKeysMap = bucketKeysMap;
6770
this.bucketNumMap = bucketNumMap;
6871
}
@@ -106,10 +109,10 @@ private void applyCreateTable(CreateTableEvent event) {
106109
String tableIdentifier = tablePath.getDatabaseName() + "." + tablePath.getTableName();
107110
List<String> bucketKeys = bucketKeysMap.get(tableIdentifier);
108111
Integer bucketNum = bucketNumMap.get(tableIdentifier);
109-
try (Connection connection = ConnectionFactory.createConnection(flussConfig);
112+
try (Connection connection = ConnectionFactory.createConnection(flussClientConfig);
110113
Admin admin = connection.getAdmin()) {
111114
TableDescriptor inferredFlussTable =
112-
toFlussTable(event.getSchema(), bucketKeys, bucketNum);
115+
toFlussTable(event.getSchema(), bucketKeys, bucketNum, tableProperties);
113116
admin.createDatabase(tablePath.getDatabaseName(), DatabaseDescriptor.EMPTY, true);
114117
if (!admin.tableExists(tablePath).get()) {
115118
admin.createTable(tablePath, inferredFlussTable, false).get();
@@ -126,7 +129,7 @@ private void applyCreateTable(CreateTableEvent event) {
126129
private void applyDropTable(DropTableEvent event) {
127130
TableId tableId = event.tableId();
128131
TablePath tablePath = new TablePath(tableId.getSchemaName(), tableId.getTableName());
129-
try (Connection connection = ConnectionFactory.createConnection(flussConfig);
132+
try (Connection connection = ConnectionFactory.createConnection(flussClientConfig);
130133
Admin admin = connection.getAdmin()) {
131134
admin.dropTable(tablePath, true).get();
132135
} catch (Exception e) {

‎flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/utils/FlinkConversions.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@
5151
import java.util.ArrayList;
5252
import java.util.Collections;
5353
import java.util.List;
54+
import java.util.Map;
5455
import java.util.stream.Collectors;
5556

5657
/** Converter from Flink's type to Fluss's type. */
@@ -60,7 +61,8 @@ public class FlinkConversions {
6061
public static TableDescriptor toFlussTable(
6162
org.apache.flink.cdc.common.schema.Schema cdcSchema,
6263
List<String> bucketKeys,
63-
@Nullable Integer bucketNum) {
64+
@Nullable Integer bucketNum,
65+
Map<String, String> tableProperties) {
6466
// now, build Fluss's table
6567
Schema.Builder schemBuilder = Schema.newBuilder();
6668
if (!CollectionUtil.isNullOrEmpty(cdcSchema.primaryKeys())) {
@@ -99,7 +101,7 @@ public static TableDescriptor toFlussTable(
99101
.partitionedBy(cdcSchema.partitionKeys())
100102
.distributedBy(bucketNum, bucketKeys)
101103
.comment(cdcSchema.comment())
102-
.properties(Collections.emptyMap())
104+
.properties(tableProperties)
103105
.build();
104106
}
105107

‎flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetadataApplierTest.java‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import com.alibaba.fluss.client.Connection;
2828
import com.alibaba.fluss.client.ConnectionFactory;
2929
import com.alibaba.fluss.client.admin.Admin;
30+
import com.alibaba.fluss.exception.InvalidConfigException;
3031
import com.alibaba.fluss.metadata.TableDescriptor;
3132
import com.alibaba.fluss.metadata.TableInfo;
3233
import com.alibaba.fluss.metadata.TablePath;
@@ -46,6 +47,7 @@
4647
import java.util.Map;
4748
import java.util.concurrent.ExecutionException;
4849

50+
import static com.alibaba.fluss.config.ConfigOptions.TABLE_REPLICATION_FACTOR;
4951
import static org.assertj.core.api.Assertions.assertThat;
5052
import static org.assertj.core.api.Assertions.assertThatThrownBy;
5153

@@ -155,6 +157,7 @@ void testCreateTableAllTypes(boolean primaryKeyTable) throws Exception {
155157
new FlussMetaDataApplier(
156158
FLUSS_CLUSTER_EXTENSION.getClientConfig(),
157159
Collections.emptyMap(),
160+
Collections.emptyMap(),
158161
Collections.emptyMap())) {
159162
// apply create-table event
160163
Schema.Builder builder = Schema.newBuilder();
@@ -201,6 +204,7 @@ void testUnsupportedType() throws Exception {
201204
new FlussMetaDataApplier(
202205
FLUSS_CLUSTER_EXTENSION.getClientConfig(),
203206
Collections.emptyMap(),
207+
Collections.emptyMap(),
204208
Collections.emptyMap())) {
205209
for (int i = 0; i < fieldNames.length; i++) {
206210
// apply create-table event
@@ -235,6 +239,7 @@ void testDropTableEvent() throws Exception {
235239
new FlussMetaDataApplier(
236240
FLUSS_CLUSTER_EXTENSION.getClientConfig(),
237241
Collections.emptyMap(),
242+
Collections.emptyMap(),
238243
Collections.emptyMap())) {
239244
assertThat(admin.tableExists(tablePath).get()).isTrue();
240245
applier.applySchemaChange(
@@ -251,6 +256,7 @@ void testCreatePartitionedTable() throws Exception {
251256
new FlussMetaDataApplier(
252257
FLUSS_CLUSTER_EXTENSION.getClientConfig(),
253258
Collections.emptyMap(),
259+
Collections.emptyMap(),
254260
Collections.emptyMap())) {
255261
// apply create-table event
256262
Schema schema =
@@ -275,6 +281,7 @@ void testPartitionKeyIsNotSubsetOfPrimaryKeys() throws Exception {
275281
new FlussMetaDataApplier(
276282
FLUSS_CLUSTER_EXTENSION.getClientConfig(),
277283
Collections.emptyMap(),
284+
Collections.emptyMap(),
278285
Collections.emptyMap())) {
279286
// apply create-table event
280287
Schema schema =
@@ -300,6 +307,7 @@ void testBuckNumber() throws Exception {
300307
new FlussMetaDataApplier(
301308
FLUSS_CLUSTER_EXTENSION.getClientConfig(),
302309
Collections.emptyMap(),
310+
Collections.emptyMap(),
303311
Collections.singletonMap(DATABASE_NAME + ".table1", 3))) {
304312
Schema schema =
305313
Schema.newBuilder()
@@ -332,6 +340,7 @@ void testBucketKey(boolean primaryKeyTable) throws Exception {
332340
try (FlussMetaDataApplier applier =
333341
new FlussMetaDataApplier(
334342
FLUSS_CLUSTER_EXTENSION.getClientConfig(),
343+
Collections.emptyMap(),
335344
Collections.singletonMap(
336345
DATABASE_NAME + ".table1", Collections.singletonList("id")),
337346
Collections.emptyMap())) {
@@ -398,6 +407,7 @@ void testBucketKeyNotSetOfPrimaryKeys(boolean partitionTables) throws Exception
398407
try (FlussMetaDataApplier applier =
399408
new FlussMetaDataApplier(
400409
FLUSS_CLUSTER_EXTENSION.getClientConfig(),
410+
Collections.emptyMap(),
401411
bucketKeys,
402412
Collections.emptyMap())) {
403413
if (partitionTables) {
@@ -484,6 +494,7 @@ void testRecreateTableWithDifferentSchema() throws Exception {
484494
new FlussMetaDataApplier(
485495
FLUSS_CLUSTER_EXTENSION.getClientConfig(),
486496
Collections.emptyMap(),
497+
Collections.emptyMap(),
487498
Collections.emptyMap())) {
488499
applier.applySchemaChange(new CreateTableEvent(tableId, initialSchema));
489500
}
@@ -493,6 +504,7 @@ void testRecreateTableWithDifferentSchema() throws Exception {
493504
new FlussMetaDataApplier(
494505
FLUSS_CLUSTER_EXTENSION.getClientConfig(),
495506
Collections.emptyMap(),
507+
Collections.emptyMap(),
496508
Collections.emptyMap())) {
497509
assertThatThrownBy(
498510
() ->
@@ -507,8 +519,43 @@ void testRecreateTableWithDifferentSchema() throws Exception {
507519
new FlussMetaDataApplier(
508520
FLUSS_CLUSTER_EXTENSION.getClientConfig(),
509521
Collections.emptyMap(),
522+
Collections.emptyMap(),
510523
Collections.emptyMap())) {
511524
applier.applySchemaChange(new CreateTableEvent(tableId, sameSchema));
512525
}
513526
}
527+
528+
@Test
529+
void testCreateTableWithTableOptions() throws Exception {
530+
Schema schema =
531+
Schema.newBuilder()
532+
.physicalColumn("id", DataTypes.INT())
533+
.physicalColumn("name", DataTypes.STRING())
534+
.primaryKey("id", "name")
535+
.partitionKey("name")
536+
.build();
537+
TableId tableId = TableId.tableId("default_namespace", DATABASE_NAME, "table1");
538+
CreateTableEvent createTableEvent = new CreateTableEvent(tableId, schema);
539+
try (FlussMetaDataApplier applier =
540+
new FlussMetaDataApplier(
541+
FLUSS_CLUSTER_EXTENSION.getClientConfig(),
542+
Collections.singletonMap("key", "value"),
543+
Collections.emptyMap(),
544+
Collections.emptyMap())) {
545+
assertThatThrownBy(() -> applier.applySchemaChange(createTableEvent))
546+
.rootCause()
547+
.isExactlyInstanceOf(InvalidConfigException.class)
548+
.hasMessageContaining("'key' is not a Fluss table property");
549+
}
550+
try (FlussMetaDataApplier applier =
551+
new FlussMetaDataApplier(
552+
FLUSS_CLUSTER_EXTENSION.getClientConfig(),
553+
Collections.singletonMap(TABLE_REPLICATION_FACTOR.key(), "5"),
554+
Collections.emptyMap(),
555+
Collections.emptyMap())) {
556+
applier.applySchemaChange(createTableEvent);
557+
TableInfo tableInfo = admin.getTableInfo(new TablePath(DATABASE_NAME, "table1")).get();
558+
assertThat(tableInfo.getProperties().get(TABLE_REPLICATION_FACTOR)).isEqualTo(5);
559+
}
560+
}
514561
}

0 commit comments

Comments
 (0)