[Improve][Connector-v2] Release resource in closeStatements even exce…

…ption occurred in executeBatch
apache · Aug 30, 2024 · ee3c20a · ee3c20a
1 parent 8ea9080
commit ee3c20a
Show file tree

Hide file tree

Showing 3 changed files with 29 additions and 14 deletions.
diff --git a/...nnel/connectors/seatunnel/jdbc/internal/executor/BufferReducedBatchStatementExecutor.java b/...nnel/connectors/seatunnel/jdbc/internal/executor/BufferReducedBatchStatementExecutor.java
@@ -96,11 +96,17 @@ public void executeBatch() throws SQLException {
 
     @Override
     public void closeStatements() throws SQLException {
-        if (!buffer.isEmpty()) {
-            executeBatch();
+        try {
+            if (!buffer.isEmpty()) {
+                executeBatch();
+            }
+        } finally {
+            if (!buffer.isEmpty()) {
+                buffer.clear();
+            }
+            upsertExecutor.closeStatements();
+            deleteExecutor.closeStatements();
         }
-        upsertExecutor.closeStatements();
-        deleteExecutor.closeStatements();
     }
 
     private boolean changeFlag(RowKind rowKind) {

diff --git a/...seatunnel/connectors/seatunnel/jdbc/internal/executor/BufferedBatchStatementExecutor.java b/...seatunnel/connectors/seatunnel/jdbc/internal/executor/BufferedBatchStatementExecutor.java
@@ -57,9 +57,15 @@ public void executeBatch() throws SQLException {
 
     @Override
     public void closeStatements() throws SQLException {
-        if (!buffer.isEmpty()) {
-            executeBatch();
+        try {
+            if (!buffer.isEmpty()) {
+                executeBatch();
+            }
+        } finally {
+            if (!buffer.isEmpty()) {
+                buffer.clear();
+            }
+            statementExecutor.closeStatements();
         }
-        statementExecutor.closeStatements();
     }
 }
diff --git a/...nel/connectors/seatunnel/jdbc/internal/executor/InsertOrUpdateBatchStatementExecutor.java b/...nel/connectors/seatunnel/jdbc/internal/executor/InsertOrUpdateBatchStatementExecutor.java
@@ -112,13 +112,16 @@ public void executeBatch() throws SQLException {
 
     @Override
     public void closeStatements() throws SQLException {
-        if (!submitted) {
-            executeBatch();
-        }
-        for (PreparedStatement statement :
-                Arrays.asList(existStatement, insertStatement, updateStatement)) {
-            if (statement != null) {
-                statement.close();
+        try {
+            if (!submitted) {
+                executeBatch();
+            }
+        } finally {
+            for (PreparedStatement statement :
+                    Arrays.asList(existStatement, insertStatement, updateStatement)) {
+                if (statement != null) {
+                    statement.close();
+                }
             }
         }
     }