[Feature][Connectors-v2] Clean up temporary files for paimon sink (#9819)

Author: hawk9821

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/PaimonSinkWriter.java

@@ -49,6 +49,7 @@
 import org.apache.paimon.CoreOptions;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.disk.IOManagerImpl;
 import org.apache.paimon.schema.TableSchema;
 import org.apache.paimon.table.BucketMode;
 import org.apache.paimon.table.FileStoreTable;
@@ -83,6 +84,8 @@ public class PaimonSinkWriter
 
     private FileStoreTable paimonTable;
 
+    private final IOManagerImpl ioManager;
+
     private TableWrite tableWrite;
 
     private final List<CommitMessage> committables = new ArrayList<>();
@@ -150,6 +153,9 @@ public PaimonSinkWriter(
         this.taskIndex = context.getIndexOfSubtask();
         this.paimonSinkConfig = paimonSinkConfig;
         this.sinkPaimonTableSchema = this.paimonTable.schema();
+        this.ioManager =
+                (IOManagerImpl)
+                        IOManager.create(splitPaths(paimonSinkConfig.getChangelogTmpPath()));
         this.newTableWrite();
         BucketMode bucketMode = this.paimonTable.bucketMode();
         // https://paimon.apache.org/docs/master/primary-key-table/data-distribution/#dynamic-bucket
@@ -275,12 +281,7 @@ private void reOpenTableWrite() {
     private void newTableWrite() {
         TableWrite oldTableWrite = this.tableWrite;
         tableWriteClose(oldTableWrite);
-        this.tableWrite =
-                this.paimonTable
-                        .newWrite(commitUser)
-                        .withIOManager(
-                                IOManager.create(
-                                        splitPaths(paimonSinkConfig.getChangelogTmpPath())));
+        this.tableWrite = this.paimonTable.newWrite(commitUser).withIOManager(ioManager);
     }
 
     @Override
@@ -329,6 +330,11 @@ public void close() throws IOException {
             if (Objects.nonNull(paimonCatalog)) {
                 paimonCatalog.close();
             }
+            try {
+                ioManager.close();
+            } catch (Exception e) {
+                log.warn("Failed to close io manager in paimon sink writer.", e);
+            }
         }
     }
 

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paimon-e2e/src/test/java/org/apache/seatunnel/e2e/connector/paimon/PaimonIT.java

@@ -17,6 +17,7 @@
 
 package org.apache.seatunnel.e2e.connector.paimon;
 
+import org.apache.seatunnel.common.utils.FileUtils;
 import org.apache.seatunnel.connectors.seatunnel.paimon.config.PaimonBaseOptions;
 import org.apache.seatunnel.e2e.common.TestResource;
 import org.apache.seatunnel.e2e.common.TestSuiteBase;
@@ -43,6 +44,7 @@
 import org.testcontainers.containers.Container;
 import org.testcontainers.utility.MountableFile;
 
+import java.io.File;
 import java.io.IOException;
 import java.nio.file.Path;
 import java.util.ArrayList;
@@ -53,10 +55,10 @@
         disabledReason =
                 "Paimon does not support flink 1.13, Spark 2.4.6 has a jar package(zstd-jni-version.jar) version compatibility issue.")
 public class PaimonIT extends TestSuiteBase implements TestResource {
-    private String rootUser = "root";
-    private String rootPassword = "123456";
-    private String paimonUser = "paimon";
-    private String paimonUserPassword = "123456";
+    private final String rootUser = "root";
+    private final String rootPassword = "123456";
+    private final String paimonUser = "paimon";
+    private final String paimonUserPassword = "123456";
 
     private PrivilegedCatalog privilegedCatalog;
     private final String DATABASE_NAME = "default";
@@ -184,4 +186,22 @@ public void privilegeEnabledPaimonSourceUnAuthorized(TestContainer container) th
                 container.executeJob("/paimon_to_paimon_privilege1.conf");
         Assertions.assertEquals(1, execResult1.getExitCode());
     }
+
+    @TestTemplate
+    public void jobFinishedCleanTmpFiles(TestContainer container) throws Exception {
+        // fake to paimon
+        Container.ExecResult execResult =
+                container.executeJob("/fake_to_paimon_with_change_log_tmp.conf");
+        Assertions.assertEquals(0, execResult.getExitCode());
+        // check job finished clean up tmp files
+        String hostName = System.getProperty("user.name");
+        boolean isWindows =
+                System.getProperties().getProperty("os.name").toUpperCase().contains("WINDOWS");
+        String tmpDir =
+                isWindows
+                        ? String.format("C:/Users/%s/tmp/seatunnel_mnt/paimon_tmp", hostName)
+                        : "/tmp/seatunnel_mnt/paimon_tmp";
+        List<File> files = FileUtils.listFile(tmpDir);
+        Assertions.assertTrue(CollectionUtils.isEmpty(files));
+    }
 }

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paimon-e2e/src/test/resources/fake_to_paimon_with_change_log_tmp.conf

@@ -0,0 +1,72 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+env {
+  parallelism = 1
+  job.mode = "BATCH"
+
+  # You can set spark configuration here
+  spark.app.name = "SeaTunnel"
+  spark.executor.instances = 2
+  spark.executor.cores = 1
+  spark.executor.memory = "1g"
+  spark.master = local
+}
+
+source {
+  FakeSource {
+    auto.increment.enabled = true
+    auto.increment.start = 1
+    row.num = 100000
+    schema = {
+      fields {
+        pk_id = bigint
+        c_map = "map<string, string>"
+        c_array = "array<int>"
+        c_string = string
+        c_boolean = boolean
+        c_tinyint = tinyint
+        c_smallint = smallint
+        c_int = int
+        c_bigint = bigint
+        c_float = float
+        c_double = double
+        c_decimal = "decimal(30, 8)"
+        c_bytes = bytes
+        c_date = date
+        c_timestamp = timestamp
+        c_time = time
+      }
+      primaryKey {
+        name = "pk_id"
+        columnNames = [pk_id]
+      }
+    }
+    plugin_output = "fake"
+  }
+}
+
+sink {
+  Paimon {
+    warehouse = "/tmp/seatunnel_mnt/paimon"
+    database = "default"
+    table = "st_test"
+    paimon.table.write-props = {
+      changelog-tmp-path = "/tmp/seatunnel_mnt/paimon_tmp"
+    }
+  }
+}