apache · luoyuxia · Apr 24, 2026 · Apr 24, 2026
diff --git a/...-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/mergetree/MergeTreeWriter.java b/...-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