avoid slot name conflict in the backfill task

apache · Jun 15, 2023 · 3a0891e · 3a0891e
1 parent c4c3f91
commit 3a0891e
Show file tree

Hide file tree

Showing 8 changed files with 28 additions and 10 deletions.
diff --git a/...c/main/java/com/ververica/cdc/connectors/postgres/source/config/PostgresSourceConfig.java b/...c/main/java/com/ververica/cdc/connectors/postgres/source/config/PostgresSourceConfig.java
@@ -33,7 +33,10 @@ public class PostgresSourceConfig extends JdbcSourceConfig {
 
     private static final long serialVersionUID = 1L;
 
+    private final int subtaskId;
+
     public PostgresSourceConfig(
+            int subtaskId,
             StartupOptions startupOptions,
             List<String> databaseList,
             List<String> schemaList,
@@ -81,6 +84,11 @@ public PostgresSourceConfig(
                 connectMaxRetries,
                 connectionPoolSize,
                 chunkKeyColumn);
+        this.subtaskId = subtaskId;
+    }
+
+    public int getSubtaskId() {
+        return subtaskId;
     }
 
     @Override

diff --git a/...java/com/ververica/cdc/connectors/postgres/source/config/PostgresSourceConfigFactory.java b/...java/com/ververica/cdc/connectors/postgres/source/config/PostgresSourceConfigFactory.java
@@ -90,6 +90,7 @@ public PostgresSourceConfig create(int subtaskId) {
 
         Configuration dbzConfiguration = Configuration.from(props);
         return new PostgresSourceConfig(
+                subtaskId,
                 startupOptions,
                 Collections.singletonList(database),
                 schemaList,

diff --git a/...c/main/java/com/ververica/cdc/connectors/postgres/source/fetch/PostgresScanFetchTask.java b/...c/main/java/com/ververica/cdc/connectors/postgres/source/fetch/PostgresScanFetchTask.java
@@ -24,6 +24,7 @@
 import com.ververica.cdc.connectors.base.source.meta.split.StreamSplit;
 import com.ververica.cdc.connectors.base.source.meta.wartermark.WatermarkKind;
 import com.ververica.cdc.connectors.base.source.reader.external.FetchTask;
+import com.ververica.cdc.connectors.postgres.source.config.PostgresSourceConfig;
 import com.ververica.cdc.connectors.postgres.source.offset.PostgresOffset;
 import com.ververica.cdc.connectors.postgres.source.offset.PostgresOffsetUtils;
 import com.ververica.cdc.connectors.postgres.source.utils.PostgresQueryUtils;
@@ -152,11 +153,17 @@ private void executeBackfillTask(
 
         // we should only capture events for the current table,
         // otherwise, we may not find corresponding schema
+        PostgresSourceConfig pgSourceConfig = (PostgresSourceConfig) ctx.getSourceConfig();
         Configuration dbzConf =
                 ctx.getDbzConnectorConfig()
                         .getConfig()
                         .edit()
                         .with("table.include.list", split.getTableId().toString())
+                        .with(
+                                "slot.name",
+                                pgSourceConfig.getDbzProperties().getProperty("slot.name")
+                                        + "_"
+                                        + pgSourceConfig.getSubtaskId())
                         // Disable heartbeat event in snapshot split fetcher
                         .with(Heartbeat.HEARTBEAT_INTERVAL, 0)
                         .build();
@@ -174,6 +181,7 @@ private void executeBackfillTask(
                         ctx.getReplicationConnection(),
                         backfillSplit);
         LOG.info("Execute backfillReadTask for split {}", split);
+        LOG.info("Slot name {}", dbzConf.getString("slot.name"));
         backfillReadTask.execute(new PostgresChangeEventSourceContext(), postgresOffsetContext);
     }
 

diff --git a/...cdc/src/main/java/com/ververica/cdc/connectors/postgres/source/offset/PostgresOffset.java b/...cdc/src/main/java/com/ververica/cdc/connectors/postgres/source/offset/PostgresOffset.java
@@ -24,6 +24,7 @@
 import org.slf4j.Logger;
 
 import javax.annotation.Nullable;
+
 import java.time.Instant;
 import java.util.HashMap;
 import java.util.Map;

diff --git a/...rc/main/java/com/ververica/cdc/connectors/postgres/source/offset/PostgresOffsetUtils.java b/...rc/main/java/com/ververica/cdc/connectors/postgres/source/offset/PostgresOffsetUtils.java
@@ -19,9 +19,9 @@
 import com.ververica.cdc.connectors.base.source.meta.offset.Offset;
 import io.debezium.connector.postgresql.PostgresOffsetContext;
 
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
-import java.util.stream.Collectors;
 
 /** Utils for handling {@link PostgresOffset}. */
 public class PostgresOffsetUtils {
@@ -32,11 +32,13 @@ public static PostgresOffsetContext getPostgresOffsetContext(
                 Objects.requireNonNull(offset, "offset is null for the sourceSplitBase")
                         .getOffset();
         // all the keys happen to be long type for PostgresOffsetContext.Loader.load
-        Map<String, Object> offsetMap =
-                offsetStrMap.entrySet().stream()
-                        .collect(
-                                Collectors.toMap(
-                                        Map.Entry::getKey, e -> Long.parseLong(e.getValue())));
+        Map<String, Object> offsetMap = new HashMap<>();
+        for (String key : offsetStrMap.keySet()) {
+            String value = offsetStrMap.get(key);
+            if (value != null) {
+                offsetMap.put(key, Long.parseLong(value));
+            }
+        }
         return loader.load(offsetMap);
     }
 }
diff --git a/...or-postgres-cdc/src/test/java/com/ververica/cdc/connectors/postgres/PostgresTestBase.java b/...or-postgres-cdc/src/test/java/com/ververica/cdc/connectors/postgres/PostgresTestBase.java
@@ -54,8 +54,7 @@ public abstract class PostgresTestBase extends AbstractTestBase {
     // use newer version of postgresql image to support pgoutput plugin
     // when testing postgres 13, only 13-alpine supports both amd64 and arm64
     protected static final DockerImageName PG_IMAGE =
-            DockerImageName.parse("debezium/postgres:13")
-                    .asCompatibleSubstituteFor("postgres");
+            DockerImageName.parse("debezium/postgres:13").asCompatibleSubstituteFor("postgres");
 
     public static final PostgreSQLContainer<?> POSTGERS_CONTAINER =
             new PostgreSQLContainer<>(PG_IMAGE)

diff --git a/...-cdc/src/test/java/com/ververica/cdc/connectors/postgres/source/PostgresSourceITCase.java b/...-cdc/src/test/java/com/ververica/cdc/connectors/postgres/source/PostgresSourceITCase.java
@@ -372,7 +372,7 @@ private void checkStreamData(
 
     private String getSlotName() {
         final Random random = new Random();
-        int id = random.nextInt(100);
+        int id = random.nextInt(10000);
         return "flink_" + id;
     }
 

diff --git a/.../src/test/java/com/ververica/cdc/connectors/postgres/table/PostgreSQLConnectorITCase.java b/.../src/test/java/com/ververica/cdc/connectors/postgres/table/PostgreSQLConnectorITCase.java
@@ -299,7 +299,6 @@ public void testAllTypes() throws Throwable {
                                 + " 'database-name' = '%s',"
                                 + " 'schema-name' = '%s',"
                                 + " 'table-name' = '%s',"
-
                                 + " 'scan.incremental.snapshot.enabled' = '%s',"
                                 + " 'slot.name' = '%s'"
                                 + ")",