Skip to content

Commit 1d128b7

Browse files
authored
[cdc] Set eager_init with option during kafka_sync_database (#5175)
1 parent b874f21 commit 1d128b7

File tree

2 files changed

+4
-1
lines changed

2 files changed

+4
-1
lines changed

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/CdcActionCommonUtils.java

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,7 @@ public class CdcActionCommonUtils {
6969
public static final String COMPUTED_COLUMN = "computed_column";
7070
public static final String METADATA_COLUMN = "metadata_column";
7171
public static final String MULTIPLE_TABLE_PARTITION_KEYS = "multiple_table_partition_keys";
72+
public static final String EAGER_INIT = "eager_init";
7273

7374
public static void assertSchemaCompatible(
7475
TableSchema paimonSchema, List<DataField> sourceTableFields) {

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionFactoryBase.java

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424

2525
import java.util.Optional;
2626

27+
import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.EAGER_INIT;
2728
import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.EXCLUDING_DBS;
2829
import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.EXCLUDING_TABLES;
2930
import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.INCLUDING_DBS;
@@ -63,7 +64,8 @@ protected void withParams(MultipleParameterToolAdapter params, T action) {
6364
.excludingDbs(params.get(EXCLUDING_DBS))
6465
.withPartitionKeyMultiple(
6566
optionalConfigMapList(params, MULTIPLE_TABLE_PARTITION_KEYS))
66-
.withPartitionKeys();
67+
.withPartitionKeys()
68+
.eagerInit(Boolean.valueOf(params.get(EAGER_INIT)));
6769

6870
if (params.has(PARTITION_KEYS)) {
6971
action.withPartitionKeys(params.get(PARTITION_KEYS).split(","));

0 commit comments

Comments
 (0)