This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 9d93f1bac81a367e31fb901cd1e69e029a0c61db Author: Alex <[email protected]> AuthorDate: Fri Sep 18 19:34:14 2026 +0800 [core] Fix OOM error handling in MergeTreeWriter and improve resource cleanup (#9933) --- .../apache/paimon/io/RollingFileWriterImpl.java | 11 +++-- .../apache/paimon/io/RollingFileWriterTest.java | 57 ++++++++++++++++++++++ 2 files changed, 65 insertions(+), 3 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/io/RollingFileWriterImpl.java b/paimon-core/src/main/java/org/apache/paimon/io/RollingFileWriterImpl.java index 11c332b90f..1d8e0fc158 100644 --- a/paimon-core/src/main/java/org/apache/paimon/io/RollingFileWriterImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/io/RollingFileWriterImpl.java @@ -49,6 +49,7 @@ public class RollingFileWriterImpl<T, R> implements RollingFileWriter<T, R> { private long recordCount = 0; private long currentFileRecordCount = 0; private boolean closed = false; + private boolean aborted = false; public RollingFileWriterImpl( Supplier<? extends SingleFileWriter<T, R>> writerFactory, @@ -177,8 +178,12 @@ public class RollingFileWriterImpl<T, R> implements RollingFileWriter<T, R> { @Override public void abort() { - if (currentWriter != null) { - currentWriter.abort(); + SingleFileWriter<T, R> writer = currentWriter; + currentWriter = null; + aborted = true; + + if (writer != null) { + writer.abort(); } for (FileWriterAbortExecutor abortExecutor : closedWriters) { abortExecutor.abort(); @@ -201,7 +206,7 @@ public class RollingFileWriterImpl<T, R> implements RollingFileWriter<T, R> { @Override public void close() throws IOException { - if (closed) { + if (closed || aborted) { return; } diff --git a/paimon-core/src/test/java/org/apache/paimon/io/RollingFileWriterTest.java b/paimon-core/src/test/java/org/apache/paimon/io/RollingFileWriterTest.java index 1676d7ebd4..175fc09a7e 100644 --- a/paimon-core/src/test/java/org/apache/paimon/io/RollingFileWriterTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/io/RollingFileWriterTest.java @@ -24,6 +24,8 @@ import org.apache.paimon.data.InternalRow; import org.apache.paimon.fileindex.FileIndexOptions; import org.apache.paimon.format.FileFormat; import org.apache.paimon.format.FormatReaderContext; +import org.apache.paimon.format.FormatWriter; +import org.apache.paimon.format.FormatWriterFactory; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.manifest.FileSource; @@ -48,8 +50,11 @@ import java.util.Arrays; import java.util.Collections; import java.util.Iterator; import java.util.List; +import java.util.function.Function; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Test for {@link RollingFileWriterImpl}. */ public class RollingFileWriterTest { @@ -140,6 +145,44 @@ public class RollingFileWriterTest { } } + @Test + public void testCloseAfterWriteFailure() throws IOException { + Path path = new Path(tempDir.resolve("failed-file").toString()); + OutOfMemoryError expected = new OutOfMemoryError("expected"); + RollingFileWriterImpl<InternalRow, Void> writer = + new RollingFileWriterImpl<>( + () -> + new TestSingleFileWriter( + LocalFileIO.create(), + (out, compression) -> + new FormatWriter() { + @Override + public void addElement(InternalRow element) { + throw expected; + } + + @Override + public boolean reachTargetSize( + boolean suggestedCheck, + long targetSize) { + return false; + } + + @Override + public void close() {} + }, + path), + Long.MAX_VALUE, + Long.MAX_VALUE); + + assertThatThrownBy(() -> writer.write(GenericRow.of(1))).isSameAs(expected); + assertThat(LocalFileIO.create().exists(path)).isFalse(); + assertThatCode(writer::abort).doesNotThrowAnyException(); + assertThatCode(writer::close).doesNotThrowAnyException(); + assertThatThrownBy(writer::result).isInstanceOf(IllegalStateException.class); + assertThatThrownBy(writer::drainAbortExecutors).isInstanceOf(IllegalStateException.class); + } + @Test public void testRollingByRows() throws IOException { // Huge byte target so only the row-count limit triggers rolling. @@ -313,6 +356,20 @@ public class RollingFileWriterTest { assertThat(file.valueStats().minValues().getFieldCount()).isEqualTo(SCHEMA.getFieldCount()); } + private static class TestSingleFileWriter extends SingleFileWriter<InternalRow, Void> { + + private TestSingleFileWriter( + LocalFileIO fileIO, FormatWriterFactory writerFactory, Path path) { + super(fileIO, writerFactory, path, Function.identity(), null, false); + } + + @Override + public Void result() throws IOException { + fileIO.getFileSize(path); + return null; + } + } + private static class SingleUseBundleRecords implements BundleRecords { private final List<InternalRow> rows;
