[core] Abort single file writers on unchecked close failures

paimon-core/src/main/java/org/apache/paimon/io/FormatTableSingleFileWriter.java

@@ -168,9 +168,13 @@ public void close() throws IOException {
                 committer = ((TwoPhaseOutputStream) out).closeForCommit();
                 out = null;
             }
-        } catch (IOException e) {
+        } catch (Throwable e) {
             LOG.warn("Exception occurs when closing file {}. Cleaning up.", path, e);
-            abort();
+            try {
+                abort();
+            } catch (Throwable t) {
+                e.addSuppressed(t);
+            }
             throw e;
         } finally {
             closed = true;

paimon-core/src/main/java/org/apache/paimon/io/SingleFileWriter.java

@@ -247,9 +247,13 @@ public void close() throws IOException {
                 out.close();
                 out = null;
             }
-        } catch (IOException e) {
+        } catch (Throwable e) {
             LOG.warn("Exception occurs when closing file {}. Cleaning up.", path, e);
-            abort();
+            try {
+                abort();
+            } catch (Throwable t) {
+                e.addSuppressed(t);
+            }
             throw e;
         } finally {
             closed = true;

paimon-core/src/test/java/org/apache/paimon/io/FormatTableSingleFileWriterTest.java

@@ -88,6 +88,48 @@ public void testExistingFileIsKeptWhenOpeningFails() throws IOException {
         assertThat(fileIO.readFileUtf8(path)).isEqualTo("keep me");
     }
 
+    @Test
+    public void testRuntimeExceptionWhileClosingLeavesNoFileBehind() throws IOException {
+        // several format writers wrap IO failures in unchecked exceptions on the close path
+        FormatTableSingleFileWriter writer =
+                newWriter(
+                        (out, compression) ->
+                                new ThrowingCloseWriter(new IllegalStateException("cannot close")));
+
+        assertThatThrownBy(writer::close)
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessage("cannot close");
+
+        assertThat(fileIO.listFiles(new Path(tempDir.toString()), true)).isEmpty();
+    }
+
+    @Test
+    public void testIOExceptionWhileClosingLeavesNoFileBehind() throws IOException {
+        FormatTableSingleFileWriter writer =
+                newWriter((out, compression) -> new ThrowingCloseWriter(new IOException("boom")));
+
+        // the checked failure must still reach the caller unwrapped
+        assertThatThrownBy(writer::close).isInstanceOf(IOException.class).hasMessage("boom");
+
+        assertThat(fileIO.listFiles(new Path(tempDir.toString()), true)).isEmpty();
+    }
+
+    @Test
+    public void testCleanupFailureDoesNotReplaceOriginalException() {
+        FormatTableSingleFileWriter writer =
+                new FormatTableSingleFileWriter(
+                        new DeleteFailingFileIO(),
+                        (out, compression) ->
+                                new ThrowingCloseWriter(new IllegalStateException("cannot close")),
+                        path,
+                        "zstd");
+
+        assertThatThrownBy(writer::close)
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessage("cannot close")
+                .hasSuppressedException(new RuntimeException("cannot delete"));
+    }
+
     private FormatTableSingleFileWriter newWriter(FormatWriterFactory factory) {
         return new FormatTableSingleFileWriter(fileIO, factory, path, "zstd");
     }
@@ -105,4 +147,37 @@ public boolean reachTargetSize(boolean suggestedCheck, long targetSize) {
         @Override
         public void close() {}
     }
+
+    private static class DeleteFailingFileIO extends LocalFileIO {
+
+        @Override
+        public boolean delete(Path f, boolean recursive) {
+            throw new RuntimeException("cannot delete");
+        }
+    }
+
+    private static class ThrowingCloseWriter implements FormatWriter {
+
+        private final Throwable failure;
+
+        private ThrowingCloseWriter(Throwable failure) {
+            this.failure = failure;
+        }
+
+        @Override
+        public void addElement(InternalRow element) {}
+
+        @Override
+        public boolean reachTargetSize(boolean suggestedCheck, long targetSize) {
+            return false;
+        }
+
+        @Override
+        public void close() throws IOException {
+            if (failure instanceof IOException) {
+                throw (IOException) failure;
+            }
+            throw (RuntimeException) failure;
+        }
+    }
 }

paimon-core/src/test/java/org/apache/paimon/io/SingleFileWriterTest.java

@@ -26,7 +26,9 @@
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.PositionOutputStreamWrapper;
 import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.utils.TraceableFileIO;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -145,6 +147,65 @@ public void testSubclassAbortIsNotCalledWhileOpeningOnIOException() throws IOExc
         assertThat(fileIO.exists(path)).isFalse();
     }
 
+    @Test
+    public void testRuntimeExceptionWhileClosingDeletesFile() throws IOException {
+        // several format writers wrap IO failures in unchecked exceptions on the close path
+        TestSingleFileWriter writer =
+                newWriter(
+                        (out, compression) ->
+                                new ThrowingCloseWriter(new IllegalStateException("cannot close")));
+
+        assertThatThrownBy(writer::close)
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessage("cannot close");
+
+        assertThat(fileIO.exists(path)).isFalse();
+    }
+
+    @Test
+    public void testIOExceptionWhileClosingDeletesFile() throws IOException {
+        TestSingleFileWriter writer =
+                newWriter((out, compression) -> new ThrowingCloseWriter(new IOException("boom")));
+
+        // the checked failure must still reach the caller unwrapped
+        assertThatThrownBy(writer::close).isInstanceOf(IOException.class).hasMessage("boom");
+
+        assertThat(fileIO.exists(path)).isFalse();
+    }
+
+    @Test
+    public void testRuntimeExceptionWhileFlushingClosesStream() throws IOException {
+        // the output stream can fail with an unchecked exception too, for example
+        // AsyncPositionOutputStream when the writing thread is interrupted
+        FileIO trackedFileIO = new FlushFailingFileIO();
+        TestSingleFileWriter writer =
+                new TestSingleFileWriter(
+                        trackedFileIO, (out, compression) -> new NoOpFormatWriter(), path, false);
+
+        assertThatThrownBy(writer::close)
+                .isExactlyInstanceOf(RuntimeException.class)
+                .hasMessage("cannot flush");
+
+        assertThat(TraceableFileIO.openOutputStreams(path::equals)).isEmpty();
+        assertThat(trackedFileIO.exists(path)).isFalse();
+    }
+
+    @Test
+    public void testCleanupFailureDoesNotReplaceOriginalException() {
+        TestSingleFileWriter writer =
+                new TestSingleFileWriter(
+                        new DeleteFailingFileIO(),
+                        (out, compression) ->
+                                new ThrowingCloseWriter(new IllegalStateException("cannot close")),
+                        path,
+                        false);
+
+        assertThatThrownBy(writer::close)
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessage("cannot close")
+                .hasSuppressedException(new RuntimeException("cannot delete"));
+    }
+
     @Test
     public void testSuccessfulOpenKeepsFile() throws IOException {
         NoOpFormatWriter formatWriter = new NoOpFormatWriter();
@@ -226,6 +287,52 @@ public void close() {
         }
     }
 
+    private static class ThrowingCloseWriter implements FormatWriter {
+
+        private final Throwable failure;
+
+        private ThrowingCloseWriter(Throwable failure) {
+            this.failure = failure;
+        }
+
+        @Override
+        public void addElement(InternalRow element) {}
+
+        @Override
+        public boolean reachTargetSize(boolean suggestedCheck, long targetSize) {
+            return false;
+        }
+
+        @Override
+        public void close() throws IOException {
+            if (failure instanceof IOException) {
+                throw (IOException) failure;
+            }
+            throw (RuntimeException) failure;
+        }
+    }
+
+    private static class DeleteFailingFileIO extends LocalFileIO {
+
+        @Override
+        public boolean delete(Path f, boolean recursive) {
+            throw new RuntimeException("cannot delete");
+        }
+    }
+
+    private static class FlushFailingFileIO extends TraceableFileIO {
+
+        @Override
+        public PositionOutputStream newOutputStream(Path f, boolean overwrite) throws IOException {
+            return new PositionOutputStreamWrapper(super.newOutputStream(f, overwrite)) {
+                @Override
+                public void flush() {
+                    throw new RuntimeException("cannot flush");
+                }
+            };
+        }
+    }
+
     private static class DirectWriteFactory implements FormatWriterFactory, SupportsDirectWrite {
 
         private final FileAwareWriter writer = new FileAwareWriter();