[postgres] close jdbc connection when close replication connection(ap…

…ache#2425)
loserwang1024 · Aug 23, 2023 · 7adb037 · 7adb037
1 parent 2f79b24
commit 7adb037
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 5 deletions.
diff --git a/...tgres-cdc/src/main/java/com/ververica/cdc/connectors/postgres/source/PostgresDialect.java b/...tgres-cdc/src/main/java/com/ververica/cdc/connectors/postgres/source/PostgresDialect.java
@@ -100,10 +100,9 @@ public PostgresConnection openJdbcConnection() {
         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 =

diff --git a/...ava/com/ververica/cdc/connectors/postgres/source/enumerator/PostgresSourceEnumerator.java b/...ava/com/ververica/cdc/connectors/postgres/source/enumerator/PostgresSourceEnumerator.java
@@ -78,9 +78,9 @@ private void createSlotForGlobalStreamSplit() {
             return;
         }
 
-        try {
+        try (PostgresConnection connection = postgresDialect.openJdbcConnection()) {
             PostgresReplicationConnection replicationConnection =
-                    postgresDialect.openPostgresReplicationConnection();
+                    postgresDialect.openPostgresReplicationConnection(connection);
             replicationConnection.createReplicationSlot();
             replicationConnection.close(false);
         } catch (Throwable t) {