This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new fea8bfb9ad [core] Abort single file writers on unchecked close
failures (#9016)
fea8bfb9ad is described below
commit fea8bfb9add68036cc54406cf664b5b084b01446
Author: Vova Kolmakov <[email protected]>
AuthorDate: Fri Aug 7 12:20:03 2026 +0700
[core] Abort single file writers on unchecked close failures (#9016)
---
.../paimon/io/FormatTableSingleFileWriter.java | 8 +-
.../org/apache/paimon/io/SingleFileWriter.java | 8 +-
.../paimon/io/FormatTableSingleFileWriterTest.java | 75 +++++++++++++++
.../org/apache/paimon/io/SingleFileWriterTest.java | 107 +++++++++++++++++++++
4 files changed, 194 insertions(+), 4 deletions(-)
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 8a01787f4c..6290d95391 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 class FormatTableSingleFileWriter {
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 29c4a448a7..8ae8f4bdc0 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 abstract class SingleFileWriter<T, R> implements
FileWriter<T, R> {
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 0b81f3801b..7802b119c9 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 class FormatTableSingleFileWriterTest {
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 class FormatTableSingleFileWriterTest {
@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 2f67f80ef4..1f90231b7a 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.format.SupportsDirectWrite;
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 class SingleFileWriterTest {
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 class SingleFileWriterTest {
}
}
+ 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();
