[postgres] Close jdbc connection after creating replication slot for stream split (#2425)

flink-connector-postgres-cdc/src/main/java/com/ververica/cdc/connectors/postgres/source/PostgresDialect.java

@@ -100,10 +100,9 @@ public class PostgresDialect implements JdbcDataSourceDialect {
         return (PostgresConnection) openJdbcConnection(sourceConfig);
     }
 
-    public PostgresReplicationConnection openPostgresReplicationConnection() {
+    public PostgresReplicationConnection openPostgresReplicationConnection(
+            PostgresConnection jdbcConnection) {
         try {
-            PostgresConnection jdbcConnection =
-                    (PostgresConnection) openJdbcConnection(sourceConfig);
             PostgresConnectorConfig pgConnectorConfig = sourceConfig.getDbzConnectorConfig();
             TopicSelector<TableId> topicSelector = PostgresTopicSelector.create(pgConnectorConfig);
             PostgresConnection.PostgresValueConverterBuilder valueConverterBuilder =

flink-connector-postgres-cdc/src/main/java/com/ververica/cdc/connectors/postgres/source/enumerator/PostgresSourceEnumerator.java

@@ -28,8 +28,6 @@ import io.debezium.connector.postgresql.connection.PostgresConnection;
 import io.debezium.connector.postgresql.connection.PostgresReplicationConnection;
 import io.debezium.connector.postgresql.spi.SlotState;
 
-import java.sql.SQLException;
-
 /**
  * The Postgres source enumerator that enumerates receive the split request and assign the split to
  * source readers.
@@ -60,32 +58,25 @@ public class PostgresSourceEnumerator extends IncrementalSourceEnumerator {
      * reading the globalStreamSplit to catch all data changes.
      */
     private void createSlotForGlobalStreamSplit() {
-        SlotState slotInfo = null;
         try (PostgresConnection connection = postgresDialect.openJdbcConnection()) {
-            slotInfo =
+            SlotState slotInfo =
                     connection.getReplicationSlotState(
                             postgresDialect.getSlotName(), postgresDialect.getPluginName());
-        } catch (SQLException e) {
-            throw new RuntimeException(
-                    String.format(
-                            "Fail to get the replication slot info, the slot name is %s.",
-                            postgresDialect.getSlotName()),
-                    e);
-        }
-
-        // skip creating the replication slot when the slot exists.
-        if (slotInfo != null) {
-            return;
-        }
-
-        try {
+            // skip creating the replication slot when the slot exists.
+            if (slotInfo != null) {
+                return;
+            }
             PostgresReplicationConnection replicationConnection =
-                    postgresDialect.openPostgresReplicationConnection();
+                    postgresDialect.openPostgresReplicationConnection(connection);
             replicationConnection.createReplicationSlot();
             replicationConnection.close(false);
+
         } catch (Throwable t) {
             throw new FlinkRuntimeException(
-                    "Create Slot For Global Stream Split failed due to ", t);
+                    String.format(
+                            "Fail to get or create slot for global stream split, the slot name is %s. Due to: ",
+                            postgresDialect.getSlotName()),
+                    t);
         }
     }
 }

flink-connector-postgres-cdc/src/test/java/com/ververica/cdc/connectors/postgres/testutils/UniqueDatabase.java

@@ -23,7 +23,6 @@ import java.net.URL;
 import java.nio.file.Files;
 import java.nio.file.Paths;
 import java.sql.Connection;
-import java.sql.DriverManager;
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.Arrays;
@@ -155,10 +154,6 @@ public class UniqueDatabase {
         }
     }
 
-    public Connection getJdbcConnection() throws SQLException {
-        return DriverManager.getConnection(container.getJdbcUrl(), username, password);
-    }
-
     private String convertSQL(final String sql) {
         return sql.replace("$DBNAME$", schemaName);
     }