[core] close files in CompactRewriter to avoid file leaks.

apache · May 27, 2024 · 6f7c9d7 · 6f7c9d7
1 parent a2363e6
commit 6f7c9d7
Show file tree

Hide file tree

Showing 3 changed files with 377 additions and 14 deletions.
diff --git a/...on-core/src/main/java/org/apache/paimon/mergetree/compact/ChangelogMergeTreeRewriter.java b/...on-core/src/main/java/org/apache/paimon/mergetree/compact/ChangelogMergeTreeRewriter.java
@@ -30,7 +30,9 @@
 import org.apache.paimon.mergetree.MergeSorter;
 import org.apache.paimon.mergetree.SortedRun;
 import org.apache.paimon.utils.CloseableIterator;
+import org.apache.paimon.utils.ExceptionUtils;
 import org.apache.paimon.utils.FieldsComparator;
+import org.apache.paimon.utils.IOUtils;
 
 import javax.annotation.Nullable;
 
@@ -123,6 +125,7 @@ private CompactResult rewriteOrProduceChangelog(
         CloseableIterator<ChangelogResult> iterator = null;
         RollingFileWriter<KeyValue, DataFileMeta> compactFileWriter = null;
         RollingFileWriter<KeyValue, DataFileMeta> changelogFileWriter = null;
+        Exception collectedExceptions = null;
 
         try {
             iterator =
@@ -151,18 +154,22 @@ private CompactResult rewriteOrProduceChangelog(
                     }
                 }
             }
+        } catch (Exception e) {
+            collectedExceptions = e;
         } finally {
-            if (iterator != null) {
-                iterator.close();
-            }
-            if (compactFileWriter != null) {
-                compactFileWriter.close();
-            }
-            if (changelogFileWriter != null) {
-                changelogFileWriter.close();
+            try {
+                IOUtils.closeAll(iterator, compactFileWriter, changelogFileWriter);
+            } catch (Exception e) {
+                collectedExceptions = ExceptionUtils.firstOrSuppressed(e, collectedExceptions);
             }
         }
 
+        if (null != collectedExceptions) {
+            compactFileWriter.abort();
+            changelogFileWriter.abort();
+            throw collectedExceptions;
+        }
+
         List<DataFileMeta> before = extractFilesFromSections(sections);
         List<DataFileMeta> after =
                 compactFileWriter != null

diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactRewriter.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactRewriter.java
@@ -32,7 +32,9 @@
 import org.apache.paimon.mergetree.SortedRun;
 import org.apache.paimon.reader.RecordReader;
 import org.apache.paimon.reader.RecordReaderIterator;
+import org.apache.paimon.utils.ExceptionUtils;
 import org.apache.paimon.utils.FieldsComparator;
+import org.apache.paimon.utils.IOUtils;
 
 import javax.annotation.Nullable;
 
@@ -75,13 +77,31 @@ protected CompactResult rewriteCompaction(
             int outputLevel, boolean dropDelete, List<List<SortedRun>> sections) throws Exception {
         RollingFileWriter<KeyValue, DataFileMeta> writer =
                 writerFactory.createRollingMergeTreeFileWriter(outputLevel, FileSource.COMPACT);
-        RecordReader<KeyValue> reader =
-                readerForMergeTree(sections, new ReducerMergeFunctionWrapper(mfFactory.create()));
-        if (dropDelete) {
-            reader = new DropDeleteReader(reader);
+        RecordReader<KeyValue> reader = null;
+        Exception collectedExceptions = null;
+        try {
+            reader =
+                    readerForMergeTree(
+                            sections, new ReducerMergeFunctionWrapper(mfFactory.create()));
+            if (dropDelete) {
+                reader = new DropDeleteReader(reader);
+            }
+            writer.write(new RecordReaderIterator<>(reader));
+        } catch (Exception e) {
+            collectedExceptions = e;
+        } finally {
+            try {
+                IOUtils.closeAll(reader, writer);
+            } catch (Exception e) {
+                collectedExceptions = ExceptionUtils.firstOrSuppressed(e, collectedExceptions);
+            }
         }
-        writer.write(new RecordReaderIterator<>(reader));
-        writer.close();
+
+        if (null != collectedExceptions) {
+            writer.abort();
+            throw collectedExceptions;
+        }
+
         List<DataFileMeta> before = extractFilesFromSections(sections);
         notifyRewriteCompactBefore(before);
         return new CompactResult(before, writer.result());