[Hotfix][Connector-V2] Close the tableWrite when task is close (#6897)

apache · May 25, 2024 · 23a744b · 23a744b
1 parent 264934f
commit 23a744b
Showing 1 changed file with 11 additions and 1 deletion.
diff --git a/...src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/PaimonSinkWriter.java b/...src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/PaimonSinkWriter.java
@@ -22,6 +22,7 @@
 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 List<PaimonSinkState> snapshotState(long checkpointId) throws IOException
     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);
+            }
+        }
+    }
 }