[tiering] Fix Paimon IOManager leak in MergeTreeWriter causing /tmp disk exhaustion

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/mergetree/MergeTreeWriter.java

@@ -45,33 +45,62 @@ public class MergeTreeWriter extends RecordWriter<KeyValue> {
 
     private final RowKeyExtractor rowKeyExtractor;
 
+    private final IOManager ioManager;
+
     public MergeTreeWriter(
             FileStoreTable fileStoreTable,
             TableBucket tableBucket,
             @Nullable String partition,
             List<String> partitionKeys) {
+        this(
+                fileStoreTable,
+                createIOManager(fileStoreTable),
+                tableBucket,
+                partition,
+                partitionKeys);
+    }
+
+    MergeTreeWriter(
+            FileStoreTable fileStoreTable,
+            IOManager ioManager,
+            TableBucket tableBucket,
+            @Nullable String partition,
+            List<String> partitionKeys) {
         super(
-                createTableWrite(fileStoreTable),
+                createTableWrite(fileStoreTable, ioManager),
                 fileStoreTable.rowType(),
                 tableBucket,
                 partition,
                 partitionKeys);
         this.rowKeyExtractor = fileStoreTable.createRowKeyExtractor();
+        this.ioManager = ioManager;
     }
 
-    private static TableWriteImpl<KeyValue> createTableWrite(FileStoreTable fileStoreTable) {
+    private static IOManager createIOManager(FileStoreTable fileStoreTable) {
         // we allow users to configure the temporary directory used by fluss tiering
         // since the default java.io.tmpdir may not be suitable.
         // currently, we don't expose the option, as a workaround way, maybe in the future we can
         // expose it if it's needed
         Map<String, String> props = fileStoreTable.options();
         String tmpDir =
                 props.getOrDefault(FLUSS_TIERING_TMP_DIR_KEY, System.getProperty("java.io.tmpdir"));
+        return IOManager.create(tmpDir);
+    }
+
+    private static TableWriteImpl<KeyValue> createTableWrite(
+            FileStoreTable fileStoreTable, IOManager ioManager) {
         //noinspection unchecked
         return (TableWriteImpl<KeyValue>)
-                fileStoreTable
-                        .newWrite(FLUSS_LAKE_TIERING_COMMIT_USER)
-                        .withIOManager(IOManager.create(tmpDir));
+                fileStoreTable.newWrite(FLUSS_LAKE_TIERING_COMMIT_USER).withIOManager(ioManager);
+    }
+
+    @Override
+    public void close() throws Exception {
+        try {
+            super.close();
+        } finally {
+            ioManager.close();
+        }
     }
 
     @Override