From 47ed972780dd828230ecbb3d4a3bde3056e5ad68 Mon Sep 17 00:00:00 2001 From: Zihan Dai <99155080+PDGGK@users.noreply.github.com> Date: Sat, 15 Aug 2026 17:20:26 +1000 Subject: [PATCH 1/2] [core] Take the stream position outside the finally so close always runs writeWithoutRolling reads the position inside the finally, ahead of the close: } finally { pos = out.getPos(); out.close(); } PositionOutputStream#getPos is declared to throw IOException. When it does, out.close() is never reached and the stream leaks; and because a throw from a finally replaces whatever the try body was throwing -- these are not suppressed, unlike try-with-resources -- the original write failure is discarded and the caller sees the getPos error instead. The two failures are not independent. The try body usually throws because the underlying stream is already broken -- disk full, an object store rejecting the upload -- and getPos on that same stream is then likely to throw as well. So the case where the position read fails is largely the same case where losing the write error hurts most. pos is only used by the success-path return, so it moves there. On a successful write the value is identical: it is still read after the FormatWriter has been closed and its buffers flushed. On a failing write, close now runs and the original exception propagates. No test. Reaching the branch needs a FileIO whose streams fail on getPos, and the existing tests build their manifest files through table.store()...Factory().create(), which supplies the catalog's own FileIO with no seam to substitute one. Happy to add a test with a hand-built ObjectsFile if you would like it. BinaryIndexManifestEntryTest covers the success path through this method and still passes. --- .../src/main/java/org/apache/paimon/utils/ObjectsFile.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsFile.java b/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsFile.java index 7d1a5d483a07..a9afc28d80bd 100644 --- a/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsFile.java +++ b/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsFile.java @@ -208,8 +208,8 @@ protected Pair writeWithoutRolling(Iterator records) { writer.addElement(serializer.toRow(records.next())); } } - } finally { pos = out.getPos(); + } finally { out.close(); } return Pair.of(path.getName(), pos); From a8763314b80d6f6edeab064cce44eb58d1de55b6 Mon Sep 17 00:00:00 2001 From: Zihan Dai Date: Sat, 15 Aug 2026 22:23:12 +1000 Subject: [PATCH 2/2] Use try-with-resources so a failing close is suppressed, not promoted Review feedback from JingsongLi: the finally block was still able to replace the original failure, because a finally that throws discards the exception in flight. try-with-resources attaches the close failure as a suppressed exception instead. The two try blocks stay nested rather than becoming one resource list: the position must be read after the writer has flushed and before the stream is closed. Adds a test that pins the exception plumbing, which is what the earlier claim got wrong. --- .../org/apache/paimon/utils/ObjectsFile.java | 7 +- .../utils/ObjectsFileWriteFailureTest.java | 160 ++++++++++++++++++ 2 files changed, 163 insertions(+), 4 deletions(-) create mode 100644 paimon-core/src/test/java/org/apache/paimon/utils/ObjectsFileWriteFailureTest.java diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsFile.java b/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsFile.java index a9afc28d80bd..e6c923ef7c32 100644 --- a/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsFile.java +++ b/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsFile.java @@ -200,17 +200,16 @@ protected Pair writeWithoutRolling(Iterator records) { } return Pair.of(path.getName(), fileIO.getFileSize(path)); } else { - PositionOutputStream out = fileIO.newOutputStream(path, false); long pos; - try { + try (PositionOutputStream out = fileIO.newOutputStream(path, false)) { + // Nested rather than a single resource list: the position has to be read + // after the writer has flushed, and before the stream itself is closed. try (FormatWriter writer = writerFactory.create(out, compression)) { while (records.hasNext()) { writer.addElement(serializer.toRow(records.next())); } } pos = out.getPos(); - } finally { - out.close(); } return Pair.of(path.getName(), pos); } diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/ObjectsFileWriteFailureTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/ObjectsFileWriteFailureTest.java new file mode 100644 index 000000000000..5c06fe58ab28 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/utils/ObjectsFileWriteFailureTest.java @@ -0,0 +1,160 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.utils; + +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.format.FormatWriter; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.PositionOutputStream; +import org.apache.paimon.fs.local.LocalFileIO; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.util.Collections; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * When both the write and the closing of the output stream fail, {@link + * ObjectsFile#writeWithoutRolling} must report the write failure and keep the close failure as a + * suppressed exception rather than letting the latter replace the former. + */ +class ObjectsFileWriteFailureTest { + + @Test + void writeFailureSurvivesAFailingStreamClose(@TempDir java.nio.file.Path tempDir) { + FailingStream stream = new FailingStream(); + ObjectsFile file = objectsFile(tempDir, stream, true); + + assertThatThrownBy(() -> file.writeWithoutRolling(Collections.emptyIterator())) + .isInstanceOf(RuntimeException.class) + .cause() + .hasMessage("writer close failed") + .satisfies( + cause -> + assertThat(cause.getSuppressed()) + .extracting(Throwable::getMessage) + .containsExactly("stream close failed")); + + assertThat(stream.closed).isTrue(); + } + + /** The stream is closed even when the write succeeds and the position read is the last step. */ + @Test + void streamIsClosedOnTheSuccessPath(@TempDir java.nio.file.Path tempDir) throws Exception { + FailingStream stream = new FailingStream(); + stream.failOnClose = false; + ObjectsFile file = objectsFile(tempDir, stream, false); + + assertThat(file.writeWithoutRolling(Collections.emptyIterator()).getValue()).isEqualTo(7L); + assertThat(stream.closed).isTrue(); + } + + private static ObjectsFile objectsFile( + java.nio.file.Path tempDir, PositionOutputStream stream, boolean writerCloseFails) { + Path path = new Path(tempDir.toUri().toString(), "manifest-0"); + FileIO fileIO = + new LocalFileIO() { + @Override + public PositionOutputStream newOutputStream(Path file, boolean overwrite) { + return stream; + } + }; + return new ObjectsFile( + fileIO, + null, + null, + (f, size) -> { + throw new UnsupportedOperationException(); + }, + (out, compression) -> new StubWriter(writerCloseFails), + "none", + new PathFactory() { + @Override + public Path newPath() { + return path; + } + + @Override + public Path toPath(String fileName) { + return path; + } + }, + null) {}; + } + + private static class StubWriter implements FormatWriter { + + private final boolean failOnClose; + + private StubWriter(boolean failOnClose) { + this.failOnClose = failOnClose; + } + + @Override + public void addElement(InternalRow element) {} + + @Override + public boolean reachTargetSize(boolean suggestedCheck, long targetSize) { + return false; + } + + @Override + public void close() throws IOException { + if (failOnClose) { + throw new IOException("writer close failed"); + } + } + } + + private static class FailingStream extends PositionOutputStream { + + private boolean failOnClose = true; + private boolean closed = false; + + @Override + public long getPos() { + return 7L; + } + + @Override + public void write(int b) {} + + @Override + public void write(byte[] b) {} + + @Override + public void write(byte[] b, int off, int len) {} + + @Override + public void flush() {} + + @Override + public void close() throws IOException { + closed = true; + if (failOnClose) { + throw new IOException("stream close failed"); + } + } + } +}