diff --git a/paimon-core/src/main/java/org/apache/paimon/io/FormatTableSingleFileWriter.java b/paimon-core/src/main/java/org/apache/paimon/io/FormatTableSingleFileWriter.java index 8a01787f4c89..6290d953919d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/io/FormatTableSingleFileWriter.java +++ b/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; diff --git a/paimon-core/src/main/java/org/apache/paimon/io/SingleFileWriter.java b/paimon-core/src/main/java/org/apache/paimon/io/SingleFileWriter.java index 29c4a448a767..8ae8f4bdc075 100644 --- a/paimon-core/src/main/java/org/apache/paimon/io/SingleFileWriter.java +++ b/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; diff --git a/paimon-core/src/test/java/org/apache/paimon/io/FormatTableSingleFileWriterTest.java b/paimon-core/src/test/java/org/apache/paimon/io/FormatTableSingleFileWriterTest.java index 0b81f3801b6d..7802b119c95e 100644 --- a/paimon-core/src/test/java/org/apache/paimon/io/FormatTableSingleFileWriterTest.java +++ b/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; + } + } } diff --git a/paimon-core/src/test/java/org/apache/paimon/io/SingleFileWriterTest.java b/paimon-core/src/test/java/org/apache/paimon/io/SingleFileWriterTest.java index 2f67f80ef4c6..1f90231b7a25 100644 --- a/paimon-core/src/test/java/org/apache/paimon/io/SingleFileWriterTest.java +++ b/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();