[Fix][Connector-V2][Postgres-CDC] Skip createReplicationSlot when slotInfo is present

seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/reader/PostgresSourceFetchTaskContext.java

@@ -183,7 +183,6 @@ public void configure(SourceSplitBase sourceSplitBase) {
                 log.warn(
                         "unable to load info of replication slot, Debezium will try to create the slot");
             }
-
             if (offsetContext == null) {
                 log.info("No previous offset found");
                 // if we have no initial offset, indicate that to Snapshotter by passing null
@@ -204,18 +203,20 @@ public void configure(SourceSplitBase sourceSplitBase) {
                                     dataConnection,
                                     snapshotter.shouldSnapshot(),
                                     connectorConfig);
-                    try {
-                        // create the slot if it doesn't exist, otherwise update slot to add new
-                        // table(job restore and add table)
-                        replicationConnection.createReplicationSlot().orElse(null);
-                    } catch (SQLException ex) {
-                        String message = "Creation of replication slot failed";
-                        if (ex.getMessage().contains("already exists")) {
-                            message +=
-                                    "; when setting up multiple connectors for the same database host, please make sure to use a distinct replication slot name for each.";
-                            log.warn(message);
-                        } else {
-                            throw new DebeziumException(message, ex);
+                    if (slotInfo == null) {
+                        try {
+                            // create the slot if it doesn't exist, otherwise update slot to add new
+                            // table(job restore and add table)
+                            replicationConnection.createReplicationSlot().orElse(null);
+                        } catch (SQLException ex) {
+                            String message = "Creation of replication slot failed";
+                            if (ex.getMessage().contains("already exists")) {
+                                message +=
+                                        "; when setting up multiple connectors for the same database host, please make sure to use a distinct replication slot name for each.";
+                                log.warn(message);
+                            } else {
+                                throw new DebeziumException(message, ex);
+                            }
                         }
                     }
                 }