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..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())); } } - } finally { pos = out.getPos(); - 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"); + } + } + } +}