You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by yu...@apache.org on 2026/04/24 09:25:36 UTC
(fluss) branch release-0.9 updated: [tiering] Fix Paimon IOManager leak in MergeTreeWriter causing /tmp disk exhaustion (#3193)
This is an automated email from the ASF dual-hosted git repository.
luoyuxia pushed a commit to branch release-0.9
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/release-0.9 by this push:
new 5c6f6b1b8 [tiering] Fix Paimon IOManager leak in MergeTreeWriter causing /tmp disk exhaustion (#3193)
5c6f6b1b8 is described below
commit 5c6f6b1b86e1f23fd16c9456139578c143782ee0
Author: yuxia Luo <lu...@alumni.sjtu.edu.cn>
AuthorDate: Fri Apr 24 17:25:31 2026 +0800
[tiering] Fix Paimon IOManager leak in MergeTreeWriter causing /tmp disk exhaustion (#3193)
---
.../paimon/tiering/mergetree/MergeTreeWriter.java | 39 +++++++++++++++++++---
1 file changed, 34 insertions(+), 5 deletions(-)
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/mergetree/MergeTreeWriter.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/mergetree/MergeTreeWriter.java
index 95a527518..5923b50ab 100644
--- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/mergetree/MergeTreeWriter.java
+++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/mergetree/MergeTreeWriter.java
@@ -45,21 +45,38 @@ 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
@@ -67,11 +84,23 @@ public class MergeTreeWriter extends RecordWriter<KeyValue> {
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