[postgres] close replication connection when close incremental Source…

…SplitReader(apache#2425)
loserwang1024 · Aug 23, 2023 · 6ec5705 · 6ec5705
1 parent 7adb037
commit 6ec5705
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 6 deletions.
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);
     }