(seatunnel) branch dev updated: [Hotfix][Connector-V2] Close the tableWrite when task is close (#6897)

[fanjia]

This is an automated email from the ASF dual-hosted git repository.

fanjia pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git


The following commit(s) were added to refs/heads/dev by this push:
     new 23a744b9b2 [Hotfix][Connector-V2] Close the tableWrite when task is 
close (#6897)
23a744b9b2 is described below

commit 23a744b9b2526d3701db1bb688474c767d670da1
Author: dailai <dai...@chinatelecom.cn>
AuthorDate: Sat May 25 14:04:18 2024 +0800

    [Hotfix][Connector-V2] Close the tableWrite when task is close (#6897)
---
 .../connectors/seatunnel/paimon/sink/PaimonSinkWriter.java   | 12 +++++++++++-
 1 file changed, 11 insertions(+), 1 deletion(-)

diff --git 
a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/PaimonSinkWriter.java
 
b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/PaimonSinkWriter.java
index 164c6a0cd9..acadf99990 100644
--- 
a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/PaimonSinkWriter.java
+++ 
b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/PaimonSinkWriter.java
@@ -22,6 +22,7 @@ import org.apache.seatunnel.api.sink.SinkWriter;
 import org.apache.seatunnel.api.sink.SupportMultiTableSinkWriter;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.common.utils.SeaTunnelException;
 import 
org.apache.seatunnel.connectors.seatunnel.paimon.config.PaimonHadoopConfiguration;
 import 
org.apache.seatunnel.connectors.seatunnel.paimon.exception.PaimonConnectorErrorCode;
 import 
org.apache.seatunnel.connectors.seatunnel.paimon.exception.PaimonConnectorException;
@@ -182,5 +183,14 @@ public class PaimonSinkWriter
     public void abortPrepare() {}
 
     @Override
-    public void close() throws IOException {}
+    public void close() throws IOException {
+        if (Objects.nonNull(tableWrite)) {
+            try {
+                tableWrite.close();
+            } catch (Exception e) {
+                log.error("Failed to close table writer in paimon sink 
writer.", e);
+                throw new SeaTunnelException(e);
+            }
+        }
+    }
 }