[postgres] close jdbc connection when close replication connection

apache · Aug 23, 2023 · e85e388 · e85e388
1 parent 2f79b24
commit e85e388
Show file tree

Hide file tree

Showing 4 changed files with 9 additions and 11 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) {

diff --git a/...va/com/ververica/cdc/connectors/postgres/source/fetch/PostgresSourceFetchTaskContext.java b/...va/com/ververica/cdc/connectors/postgres/source/fetch/PostgresSourceFetchTaskContext.java
@@ -286,7 +286,11 @@ public Offset getStreamOffset(SourceRecord sourceRecord) {
     }
 
     @Override
-    public void close() {
+    public void close() throws Exception {
+        ReplicationConnection replicationCon = replicationConnection.get();
+        if (replicationCon != null) {
+            replicationCon.close();
+        }
         jdbcConnection.close();
     }
 

diff --git a/...res-cdc/src/test/java/com/ververica/cdc/connectors/postgres/testutils/UniqueDatabase.java b/...res-cdc/src/test/java/com/ververica/cdc/connectors/postgres/testutils/UniqueDatabase.java
@@ -23,7 +23,6 @@
 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 void createAndInitialize() {
         }
     }
 
-    public Connection getJdbcConnection() throws SQLException {
-        return DriverManager.getConnection(container.getJdbcUrl(), username, password);
-    }
-
     private String convertSQL(final String sql) {
         return sql.replace("$DBNAME$", schemaName);
     }