[DBZ-PGYB][yugabyte/yugabyte-db#24555] Add task ID to PostgresPartition

debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresPartition.java

@@ -21,15 +21,21 @@ public class PostgresPartition extends AbstractPartition implements Partition {
     private static final String SERVER_PARTITION_KEY = "server";
 
     private final String serverName;
+    private int taskId = 0;
 
     public PostgresPartition(String serverName, String databaseName) {
         super(databaseName);
         this.serverName = serverName;
     }
 
+    public PostgresPartition(String serverName, String databaseName, int taskId) {
+        this(serverName, databaseName);
+        this.taskId = taskId;
+    }
+
     @Override
     public Map<String, String> getSourcePartition() {
-        return Collect.hashMapOf(SERVER_PARTITION_KEY, serverName);
+        return Collect.hashMapOf(SERVER_PARTITION_KEY, getPartitionIdentificationKey());
     }
 
     @Override
@@ -54,6 +60,10 @@ public String toString() {
         return "PostgresPartition [sourcePartition=" + getSourcePartition() + "]";
     }
 
+    public String getPartitionIdentificationKey() {
+        return String.format("%s_%d", serverName, taskId);
+    }
+
     static class Provider implements Partition.Provider<PostgresPartition> {
         private final PostgresConnectorConfig connectorConfig;
         private final Configuration taskConfig;
@@ -66,7 +76,8 @@ static class Provider implements Partition.Provider<PostgresPartition> {
         @Override
         public Set<PostgresPartition> getPartitions() {
             return Collections.singleton(new PostgresPartition(
-                    connectorConfig.getLogicalName(), taskConfig.getString(DATABASE_NAME.name())));
+                    connectorConfig.getLogicalName(), taskConfig.getString(DATABASE_NAME.name()),
+                    connectorConfig.taskId()));
         }
     }
 }