This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 622ad78cf45ac2bb390571c32f695343402effd5 Author: fhan <[email protected]> AuthorDate: Tue Jun 9 12:05:09 2026 +0800 fix(common): Close log writer output stream on append failure (#18909) * fix(common): Close log writer output stream on append failure * simplify the exception handling --------- Co-authored-by: fhan <[email protected]> Co-authored-by: danny0405 <[email protected]> (cherry picked from commit 782bf654ee4cac9da7e1244da03d83b298bbbe3a) --- .../org/apache/hudi/exception/ExceptionUtil.java | 18 +- .../apache/hudi/exception/TestExceptionUtil.java | 20 +++ .../common/table/log/HoodieLogFormatWriter.java | 184 +++++++++++++-------- .../table/log/TestHoodieLogFormatWriter.java | 180 ++++++++++++++++++++ 4 files changed, 333 insertions(+), 69 deletions(-) diff --git a/hudi-common/src/main/java/org/apache/hudi/exception/ExceptionUtil.java b/hudi-common/src/main/java/org/apache/hudi/exception/ExceptionUtil.java index 92e2e3cc5356..40a39542c7a7 100644 --- a/hudi-common/src/main/java/org/apache/hudi/exception/ExceptionUtil.java +++ b/hudi-common/src/main/java/org/apache/hudi/exception/ExceptionUtil.java @@ -23,6 +23,8 @@ import org.apache.hudi.common.util.StringUtils; import javax.annotation.Nonnull; +import java.io.IOException; + /** * Util class for exception analysis. */ @@ -31,7 +33,7 @@ public final class ExceptionUtil { } /** - * Returns true if error message is contained in any nested exception of provided {@link Throwable}. + * Returns true if error message is contained in any nested exception to provided {@link Throwable}. */ public static boolean validateErrorMsg(@Nonnull Throwable t, String errorMsg) { if (StringUtils.isNullOrEmpty(errorMsg)) { @@ -48,4 +50,18 @@ public final class ExceptionUtil { return false; } + + /** + * Throws the provided exception as-is when it is an {@link IOException} or + * {@link RuntimeException}, otherwise wraps it in an {@link IOException}. + */ + public static void throwAsIOExceptionOrRuntimeException(Throwable exception) throws IOException { + if (exception instanceof IOException) { + throw (IOException) exception; + } + if (exception instanceof RuntimeException) { + throw (RuntimeException) exception; + } + throw new IOException(exception); + } } diff --git a/hudi-common/src/test/java/org/apache/hudi/exception/TestExceptionUtil.java b/hudi-common/src/test/java/org/apache/hudi/exception/TestExceptionUtil.java index dba850b1b907..c2244896d855 100644 --- a/hudi-common/src/test/java/org/apache/hudi/exception/TestExceptionUtil.java +++ b/hudi-common/src/test/java/org/apache/hudi/exception/TestExceptionUtil.java @@ -23,8 +23,11 @@ import org.junit.jupiter.api.Test; import java.io.IOException; +import static org.apache.hudi.exception.ExceptionUtil.throwAsIOExceptionOrRuntimeException; import static org.apache.hudi.exception.ExceptionUtil.validateErrorMsg; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; class TestExceptionUtil { @@ -54,4 +57,21 @@ class TestExceptionUtil { // Empty string should not be found in any message (including null) assertFalse(validateErrorMsg(exceptionWithoutMessage, "")); } + + @Test + void testThrowAsIOExceptionOrRuntimeException() { + IOException ioException = new IOException("io"); + IOException thrownIOException = assertThrows(IOException.class, () -> throwAsIOExceptionOrRuntimeException(ioException)); + assertSame(ioException, thrownIOException); + + RuntimeException runtimeException = new RuntimeException("runtime"); + RuntimeException thrownRuntimeException = + assertThrows(RuntimeException.class, () -> throwAsIOExceptionOrRuntimeException(runtimeException)); + assertSame(runtimeException, thrownRuntimeException); + + Exception checkedException = new Exception("checked"); + IOException wrappedException = + assertThrows(IOException.class, () -> throwAsIOExceptionOrRuntimeException(checkedException)); + assertSame(checkedException, wrappedException.getCause()); + } } diff --git a/hudi-hadoop-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormatWriter.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormatWriter.java index 2aed1d7dd87a..fe24bd600f1e 100644 --- a/hudi-hadoop-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormatWriter.java +++ b/hudi-hadoop-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormatWriter.java @@ -22,6 +22,7 @@ package org.apache.hudi.common.table.log; import org.apache.hudi.common.table.HoodieTableVersion; import org.apache.hudi.common.table.log.block.HoodieLogBlock; import org.apache.hudi.common.util.VisibleForTesting; +import org.apache.hudi.exception.ExceptionUtil; import org.apache.hudi.exception.HoodieIOException; import org.apache.hudi.storage.HoodieStorage; import org.apache.hudi.storage.StoragePath; @@ -125,68 +126,73 @@ public class HoodieLogFormatWriter extends HoodieLogFormat.Writer { @Override public AppendResult appendBlocks(List<HoodieLogBlock> blocks) throws IOException { - // Find current version - HoodieLogFormat.LogFormatVersion currentLogFormatVersion = - new HoodieLogFormatVersion(HoodieLogFormat.CURRENT_VERSION); - - FSDataOutputStream originalOutputStream = getOutputStream(); - long startPos = originalOutputStream.getPos(); - long sizeWritten = 0; - // HUDI-2655. here we wrap originalOutputStream to ensure huge blocks can be correctly written - FSDataOutputStream outputStream = new FSDataOutputStream(originalOutputStream, new FileSystem.Statistics(storage.getScheme()), startPos); - for (HoodieLogBlock block: blocks) { - long startSize = outputStream.size(); - - // 1. Write the magic header for the start of the block - outputStream.write(HoodieLogFormat.MAGIC); - - // bytes for header - byte[] headerBytes = HoodieLogBlock.getHeaderMetadataBytes(block.getLogBlockHeader()); - // content bytes - ByteArrayOutputStream content = block.getContentBytes(storage); - // bytes for footer - byte[] footerBytes = HoodieLogBlock.getFooterMetadataBytes(block.getLogBlockFooter()); - - // 2. Write the total size of the block (excluding Magic) - outputStream.writeLong(getLogBlockLength(content.size(), headerBytes.length, footerBytes.length)); - - // 3. Write the version of this log block - outputStream.writeInt(currentLogFormatVersion.getVersion()); - // 4. Write the block type - outputStream.writeInt(block.getBlockType().ordinal()); - - // 5. Write the headers for the log block - outputStream.write(headerBytes); - // 6. Write the size of the content block - outputStream.writeLong(content.size()); - // 7. Write the contents of the data block - content.writeTo(outputStream); - // 8. Write the footers for the log block - outputStream.write(footerBytes); - // 9. Write the total size of the log block (including magic) which is everything written - // until now (for reverse pointer) - // Update: this information is now used in determining if a block is corrupt by comparing to the - // block size in header. This change assumes that the block size will be the last data written - // to a block. Read will break if any data is written past this point for a block. - outputStream.writeLong(outputStream.size() - startSize); - - // Fetch the size again, so it accounts also (9). - - // HUDI-2655. Check the size written to avoid log blocks whose size overflow. - if (outputStream.size() == Integer.MAX_VALUE) { - throw new HoodieIOException("Blocks appended may overflow. Please decrease log block size or log block amount"); + try { + // Find current version + HoodieLogFormat.LogFormatVersion currentLogFormatVersion = + new HoodieLogFormatVersion(HoodieLogFormat.CURRENT_VERSION); + + FSDataOutputStream originalOutputStream = getOutputStream(); + long startPos = originalOutputStream.getPos(); + long sizeWritten = 0; + // HUDI-2655. here we wrap originalOutputStream to ensure huge blocks can be correctly written + FSDataOutputStream outputStream = new FSDataOutputStream(originalOutputStream, new FileSystem.Statistics(storage.getScheme()), startPos); + for (HoodieLogBlock block: blocks) { + long startSize = outputStream.size(); + + // 1. Write the magic header for the start of the block + outputStream.write(HoodieLogFormat.MAGIC); + + // bytes for header + byte[] headerBytes = HoodieLogBlock.getHeaderMetadataBytes(block.getLogBlockHeader()); + // content bytes + ByteArrayOutputStream content = block.getContentBytes(storage); + // bytes for footer + byte[] footerBytes = HoodieLogBlock.getFooterMetadataBytes(block.getLogBlockFooter()); + + // 2. Write the total size of the block (excluding Magic) + outputStream.writeLong(getLogBlockLength(content.size(), headerBytes.length, footerBytes.length)); + + // 3. Write the version of this log block + outputStream.writeInt(currentLogFormatVersion.getVersion()); + // 4. Write the block type + outputStream.writeInt(block.getBlockType().ordinal()); + + // 5. Write the headers for the log block + outputStream.write(headerBytes); + // 6. Write the size of the content block + outputStream.writeLong(content.size()); + // 7. Write the contents of the data block + content.writeTo(outputStream); + // 8. Write the footers for the log block + outputStream.write(footerBytes); + // 9. Write the total size of the log block (including magic) which is everything written + // until now (for reverse pointer) + // Update: this information is now used in determining if a block is corrupt by comparing to the + // block size in header. This change assumes that the block size will be the last data written + // to a block. Read will break if any data is written past this point for a block. + outputStream.writeLong(outputStream.size() - startSize); + + // Fetch the size again, so it accounts also (9). + + // HUDI-2655. Check the size written to avoid log blocks whose size overflow. + if (outputStream.size() == Integer.MAX_VALUE) { + throw new HoodieIOException("Blocks appended may overflow. Please decrease log block size or log block amount"); + } + sizeWritten += outputStream.size() - startSize; } - sizeWritten += outputStream.size() - startSize; + // No flush/hsync here: append-time visibility is not part of the contract. + // Downstream readers only need commit-level visibility, which is provided + // when the writer is closed (see closeStream) or when callers explicitly + // invoke sync(). + + AppendResult result = new AppendResult(logFile, startPos, sizeWritten); + // roll over if size is past the threshold + rolloverIfNeeded(); + return result; + } catch (IOException | RuntimeException e) { + closeOutputStreamOnAppendFailure(e); + throw e; } - // No flush/hsync here: append-time visibility is not part of the contract. - // Downstream readers only need commit-level visibility, which is provided - // when the writer is closed (see closeStream) or when callers explicitly - // invoke sync(). - - AppendResult result = new AppendResult(logFile, startPos, sizeWritten); - // roll over if size is past the threshold - rolloverIfNeeded(); - return result; } /** @@ -237,21 +243,63 @@ public class HoodieLogFormatWriter extends HoodieLogFormat.Writer { @Override public void close() throws IOException { - closeStream(); - // remove the shutdown hook after closing the stream to avoid memory leaks - if (null != shutdownThread) { - Runtime.getRuntime().removeShutdownHook(shutdownThread); + try { + closeStream(); + } finally { + // remove the shutdown hook after closing the stream to avoid memory leaks + if (null != shutdownThread) { + Runtime.getRuntime().removeShutdownHook(shutdownThread); + shutdownThread = null; + } } } private void closeStream() throws IOException { - if (outputStream != null) { + if (outputStream == null) { + return; + } + + Throwable failure = null; + try { // Persist all buffered data to DataNodes before closing so downstream // readers can observe a fully-written log file at commit-level visibility. sync(); - outputStream.close(); - outputStream = null; - closed = true; + } catch (IOException | RuntimeException e) { + failure = e; + } + + try { + closeOutputStream(); + } catch (IOException | RuntimeException closeException) { + if (failure != null) { + failure.addSuppressed(closeException); + } else { + failure = closeException; + } + } + + if (failure != null) { + ExceptionUtil.throwAsIOExceptionOrRuntimeException(failure); + } + } + + private void closeOutputStreamOnAppendFailure(Throwable failure) { + try { + closeOutputStream(); + } catch (IOException | RuntimeException closeException) { + failure.addSuppressed(closeException); + log.warn("Failed to close output stream after append failure for log file {}", logFile, closeException); + } + } + + private void closeOutputStream() throws IOException { + if (outputStream != null) { + try { + outputStream.close(); + } finally { + outputStream = null; + closed = true; + } } } diff --git a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/log/TestHoodieLogFormatWriter.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/log/TestHoodieLogFormatWriter.java new file mode 100644 index 000000000000..15316012a97a --- /dev/null +++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/log/TestHoodieLogFormatWriter.java @@ -0,0 +1,180 @@ +/* + * 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.hudi.common.table.log; + +import org.apache.hudi.common.model.HoodieLogFile; +import org.apache.hudi.common.table.log.block.HoodieCommandBlock; +import org.apache.hudi.common.table.log.block.HoodieLogBlock; +import org.apache.hudi.common.testutils.HoodieTestUtils; +import org.apache.hudi.storage.HoodieStorage; +import org.apache.hudi.storage.StoragePath; + +import org.apache.hadoop.fs.FSDataOutputStream; +import org.apache.hadoop.fs.FileSystem; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.io.OutputStream; +import java.util.HashMap; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class TestHoodieLogFormatWriter { + + private static final String WRITE_FAIL = "write-fail"; + private static final String CLOSE_FAIL = "close-fail"; + private static final String SYNC_FAIL = "sync-fail"; + + @TempDir + java.nio.file.Path tempDir; + + @Test + void testCloseOutputOnAppendWriteException() throws IOException { + HoodieStorage storage = HoodieTestUtils.getStorage(tempDir.toString()); + HoodieLogFormatWriter writer = newWriter(storage); + try { + CloseTrackingOutputStream outputStream = new CloseTrackingOutputStream(true, false); + writer.withOutputStream(newFSDataOutputStream(outputStream, storage)); + + IOException exception = assertThrows(IOException.class, () -> writer.appendBlock(commandBlock())); + + assertEquals(WRITE_FAIL, exception.getMessage()); + assertTrue(outputStream.isClosed()); + assertThrows(IllegalStateException.class, writer::getCurrentSize); + } finally { + writer.close(); + } + } + + @Test + void testPreserveAppendExceptionWhenCloseFails() throws IOException { + HoodieStorage storage = HoodieTestUtils.getStorage(tempDir.toString()); + HoodieLogFormatWriter writer = newWriter(storage); + try { + CloseTrackingOutputStream outputStream = new CloseTrackingOutputStream(true, true); + writer.withOutputStream(newFSDataOutputStream(outputStream, storage)); + + IOException exception = assertThrows(IOException.class, () -> writer.appendBlock(commandBlock())); + + assertEquals(WRITE_FAIL, exception.getMessage()); + assertTrue(outputStream.isClosed()); + assertEquals(1, exception.getSuppressed().length); + assertEquals(CLOSE_FAIL, exception.getSuppressed()[0].getMessage()); + assertThrows(IllegalStateException.class, writer::getCurrentSize); + } finally { + writer.close(); + } + } + + @Test + void testCloseOutputWhenSyncFailsOnClose() throws IOException { + HoodieStorage storage = HoodieTestUtils.getStorage(tempDir.toString()); + HoodieLogFormatWriter writer = newWriter(storage); + try { + CloseTrackingOutputStream outputStream = new CloseTrackingOutputStream(false, false); + writer.withOutputStream(new SyncFailingFSDataOutputStream(outputStream, storage)); + + IOException exception = assertThrows(IOException.class, writer::close); + + assertEquals(SYNC_FAIL, exception.getMessage()); + assertTrue(outputStream.isClosed()); + assertThrows(IllegalStateException.class, writer::getCurrentSize); + } finally { + writer.close(); + } + } + + private HoodieLogFormatWriter newWriter(HoodieStorage storage) throws IOException { + return HoodieLogFormatWriter.builder() + .withParentPath(new StoragePath(tempDir.toString())) + .withFileExtension(HoodieLogFile.DELTA_EXTENSION) + .withLogFileId("test-fileid") + .withInstantTime("100") + .withLogVersion(1) + .withStorage(storage) + .build(); + } + + private HoodieCommandBlock commandBlock() { + Map<HoodieLogBlock.HeaderMetadataType, String> header = new HashMap<>(); + header.put(HoodieLogBlock.HeaderMetadataType.COMMAND_BLOCK_TYPE, + String.valueOf(HoodieCommandBlock.HoodieCommandBlockTypeEnum.ROLLBACK_BLOCK.ordinal())); + return new HoodieCommandBlock(header); + } + + private FSDataOutputStream newFSDataOutputStream(CloseTrackingOutputStream outputStream, HoodieStorage storage) + throws IOException { + return new FSDataOutputStream(outputStream, new FileSystem.Statistics(storage.getScheme())); + } + + private static class CloseTrackingOutputStream extends OutputStream { + + private final boolean failOnWrite; + private final boolean failOnClose; + private boolean closed; + + private CloseTrackingOutputStream(boolean failOnWrite, boolean failOnClose) { + this.failOnWrite = failOnWrite; + this.failOnClose = failOnClose; + } + + @Override + public void write(int b) throws IOException { + if (failOnWrite) { + throw new IOException(WRITE_FAIL); + } + } + + @Override + public void write(byte[] b, int off, int len) throws IOException { + if (failOnWrite) { + throw new IOException(WRITE_FAIL); + } + } + + @Override + public void close() throws IOException { + closed = true; + if (failOnClose) { + throw new IOException(CLOSE_FAIL); + } + } + + private boolean isClosed() { + return closed; + } + } + + private static class SyncFailingFSDataOutputStream extends FSDataOutputStream { + + private SyncFailingFSDataOutputStream(CloseTrackingOutputStream outputStream, HoodieStorage storage) + throws IOException { + super(outputStream, new FileSystem.Statistics(storage.getScheme())); + } + + @Override + public void hsync() throws IOException { + throw new IOException(SYNC_FAIL); + } + } +}
