This is an automated email from the ASF dual-hosted git repository. danny0405 pushed a commit to branch release-1.1.2 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 26a4648233bc816197df000d9a6b7bcd7ad0586d Author: Danny Chan <[email protected]> AuthorDate: Wed Aug 19 15:37:36 2026 +0800 fix(storage-format): ignore temporary LSM timeline manifests (#19659) * fix(storage-format): ignore temporary LSM timeline manifests * fix(storage): handle immutable file cleanup failures (cherry picked from commit 5a11a4fce20620afac143358bc771dfe2d14d3c8) --- .../hudi/common/table/timeline/LSMTimeline.java | 11 +- .../common/table/timeline/TestLSMTimeline.java | 58 +++++++++++ .../storage/hadoop/TestHoodieHadoopStorage.java | 114 +++++++++++++++++++++ .../org/apache/hudi/storage/HoodieStorage.java | 76 ++++++-------- .../hudi/io/storage/TestHoodieStorageBase.java | 19 ++++ 5 files changed, 232 insertions(+), 46 deletions(-) diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/LSMTimeline.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/LSMTimeline.java index c99dcb67f0bd..5e314c662a47 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/LSMTimeline.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/LSMTimeline.java @@ -110,8 +110,7 @@ public class LSMTimeline { private static final String VERSION_FILE_NAME = "_version_"; // _version_ private static final String MANIFEST_FILE_PREFIX = "manifest_"; // manifest_[N] - - private static final String TEMP_FILE_SUFFIX = ".tmp"; + private static final Pattern MANIFEST_FILE_PATTERN = Pattern.compile("^" + MANIFEST_FILE_PREFIX + "(\\d+)$"); private static final Pattern ARCHIVE_FILE_PATTERN = Pattern.compile("^(\\d+)_(\\d+)_(\\d)\\.parquet"); @@ -242,7 +241,11 @@ public class LSMTimeline { * Parse the snapshot version from the manifest file name. */ public static int getManifestVersion(String fileName) { - return Integer.parseInt(fileName.split("_")[1]); + Matcher fileMatcher = MANIFEST_FILE_PATTERN.matcher(fileName); + if (fileMatcher.matches()) { + return Integer.parseInt(fileMatcher.group(1)); + } + throw new HoodieException("Unexpected manifest file name: " + fileName); } /** @@ -298,6 +301,6 @@ public class LSMTimeline { * Returns a path filter for the manifest files. */ public static StoragePathFilter getManifestFilePathFilter() { - return path -> path.getName().startsWith(MANIFEST_FILE_PREFIX) && !path.getName().endsWith(TEMP_FILE_SUFFIX); + return path -> MANIFEST_FILE_PATTERN.matcher(path.getName()).matches(); } } diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/timeline/TestLSMTimeline.java b/hudi-common/src/test/java/org/apache/hudi/common/table/timeline/TestLSMTimeline.java index 009c926fa226..f8506962ad47 100644 --- a/hudi-common/src/test/java/org/apache/hudi/common/table/timeline/TestLSMTimeline.java +++ b/hudi-common/src/test/java/org/apache/hudi/common/table/timeline/TestLSMTimeline.java @@ -19,13 +19,30 @@ package org.apache.hudi.common.table.timeline; +import org.apache.hudi.common.table.HoodieTableMetaClient; import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.storage.HoodieStorage; +import org.apache.hudi.storage.StoragePath; +import org.apache.hudi.storage.StoragePathFilter; +import org.apache.hudi.storage.StoragePathInfo; import org.junit.jupiter.api.Test; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.util.Arrays; +import java.util.List; +import java.util.stream.Collectors; + import static org.hamcrest.CoreMatchers.is; import static org.hamcrest.MatcherAssert.assertThat; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; /** * Test cases for {@link LSMTimeline}. @@ -54,4 +71,45 @@ public class TestLSMTimeline { assertThat(layer, is(0)); assertThat("for invalid file name, returns 0", LSMTimeline.getFileLayer("invalid_file_name.parquet"), is(0)); } + + @Test + void testManifestFileValidation() { + StoragePathFilter filter = LSMTimeline.getManifestFilePathFilter(); + + assertTrue(filter.accept(new StoragePath("manifest_1"))); + assertTrue(filter.accept(new StoragePath("manifest_114"))); + assertThat(LSMTimeline.getManifestVersion("manifest_114"), is(114)); + + assertFalse(filter.accept(new StoragePath("manifest_114.tmp"))); + assertFalse(filter.accept(new StoragePath("manifest_114.a5e022a6-e5c9-4450-b9e5-9296262329b5"))); + assertFalse(filter.accept(new StoragePath("manifest_invalid"))); + assertFalse(filter.accept(new StoragePath("manifest_"))); + assertThrows(HoodieException.class, + () -> LSMTimeline.getManifestVersion("manifest_114.a5e022a6-e5c9-4450-b9e5-9296262329b5")); + } + + @Test + void testLatestSnapshotVersionFallbackIgnoresTemporaryManifest() throws IOException { + StoragePath archivePath = new StoragePath("/table/.hoodie/timeline/history"); + HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class); + HoodieStorage storage = mock(HoodieStorage.class); + List<StoragePathInfo> manifestFiles = Arrays.asList( + manifestFile(archivePath, "manifest_113"), + manifestFile(archivePath, "manifest_114"), + manifestFile(archivePath, "manifest_114.a5e022a6-e5c9-4450-b9e5-9296262329b5")); + + when(metaClient.getStorage()).thenReturn(storage); + when(storage.open(LSMTimeline.getVersionFilePath(archivePath))).thenThrow(new FileNotFoundException()); + when(storage.listDirectEntries(eq(archivePath), any(StoragePathFilter.class))).thenAnswer(invocation -> { + StoragePathFilter filter = invocation.getArgument(1); + return manifestFiles.stream().filter(file -> filter.accept(file.getPath())).collect(Collectors.toList()); + }); + + assertThat(LSMTimeline.latestSnapshotVersion(metaClient, archivePath), is(114)); + assertThat(LSMTimeline.allSnapshotVersions(metaClient, archivePath), is(Arrays.asList(113, 114))); + } + + private static StoragePathInfo manifestFile(StoragePath archivePath, String fileName) { + return new StoragePathInfo(new StoragePath(archivePath, fileName), 0, false, (short) 1, 0, 0); + } } diff --git a/hudi-hadoop-common/src/test/java/org/apache/hudi/storage/hadoop/TestHoodieHadoopStorage.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/storage/hadoop/TestHoodieHadoopStorage.java index 8d0f64d0c6bd..0b060e144ae1 100644 --- a/hudi-hadoop-common/src/test/java/org/apache/hudi/storage/hadoop/TestHoodieHadoopStorage.java +++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/storage/hadoop/TestHoodieHadoopStorage.java @@ -19,17 +19,28 @@ package org.apache.hudi.storage.hadoop; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.exception.HoodieIOException; import org.apache.hudi.hadoop.fs.HadoopFSUtils; import org.apache.hudi.io.storage.TestHoodieStorageBase; +import org.apache.hudi.storage.HoodieInstantWriter; import org.apache.hudi.storage.HoodieStorage; +import org.apache.hudi.storage.StoragePath; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.junit.jupiter.api.Test; +import java.io.FilterOutputStream; import java.io.IOException; +import java.io.OutputStream; +import static org.junit.jupiter.api.Assertions.assertEquals; +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; /** * Tests {@link HoodieHadoopStorage}. @@ -70,4 +81,107 @@ public class TestHoodieHadoopStorage extends TestHoodieStorageBase { assertSame(fileSystem, storage.getFileSystem()); assertSame(fileSystem, HadoopFSUtils.getFs(getTempDir(), conf, true)); } + + @Test + void testCreateImmutableFileCleansTemporaryFileAfterCloseFailure() throws IOException { + Configuration conf = new Configuration(); + FileSystem fileSystem = HadoopFSUtils.getFs(getTempDir(), conf, true); + HoodieStorage storage = new CloseFailingHoodieHadoopStorage(fileSystem); + StoragePath directory = new StoragePath(getTempDir(), "testImmutableFileCloseFailure"); + StoragePath path = new StoragePath(directory, "1.file"); + storage.createDirectory(directory); + + HoodieIOException exception = assertThrows(HoodieIOException.class, + () -> storage.createImmutableFileInPath(path, + Option.of(HoodieInstantWriter.convertByteArrayToWriter(new byte[] {42})))); + + assertTrue(exception.getMessage().startsWith("Failed to create immutable file ")); + assertFalse(storage.exists(path)); + assertTrue(storage.listDirectEntries(directory).isEmpty()); + } + + @Test + void testCreateImmutableFileCleansTemporaryFileAfterUncheckedRenameFailure() throws IOException { + Configuration conf = new Configuration(); + FileSystem fileSystem = HadoopFSUtils.getFs(getTempDir(), conf, true); + HoodieStorage storage = new RenameFailingHoodieHadoopStorage(fileSystem); + StoragePath directory = new StoragePath(getTempDir(), "testImmutableFileRenameFailure"); + StoragePath path = new StoragePath(directory, "1.file"); + storage.createDirectory(directory); + + HoodieException exception = assertThrows(HoodieException.class, + () -> storage.createImmutableFileInPath(path, + Option.of(HoodieInstantWriter.convertByteArrayToWriter(new byte[] {42})))); + + assertEquals("rename failure", exception.getMessage()); + assertFalse(storage.exists(path)); + assertTrue(storage.listDirectEntries(directory).isEmpty()); + } + + @Test + void testCreateImmutableFileSuppressesUncheckedCleanupFailure() throws IOException { + Configuration conf = new Configuration(); + FileSystem fileSystem = HadoopFSUtils.getFs(getTempDir(), conf, true); + CleanupFailingHoodieHadoopStorage storage = new CleanupFailingHoodieHadoopStorage(fileSystem); + StoragePath directory = new StoragePath(getTempDir(), "testImmutableFileCleanupFailure"); + StoragePath path = new StoragePath(directory, "1.file"); + storage.createDirectory(directory); + + HoodieIOException exception = assertThrows(HoodieIOException.class, + () -> storage.createImmutableFileInPath(path, Option.of(outputStream -> { + outputStream.write(42); + throw new IOException("write failure"); + }))); + + assertEquals("write failure", exception.getCause().getMessage()); + assertEquals(1, exception.getSuppressed().length); + assertEquals("cleanup failure", exception.getSuppressed()[0].getMessage()); + assertFalse(storage.exists(path)); + assertEquals(1, storage.listDirectEntries(directory).size()); + StoragePath temporaryPath = storage.listDirectEntries(directory).get(0).getPath(); + storage.deleteTemporaryFileAfterTest(temporaryPath); + } + + private static class CloseFailingHoodieHadoopStorage extends HoodieHadoopStorage { + CloseFailingHoodieHadoopStorage(FileSystem fileSystem) { + super(fileSystem); + } + + @Override + public OutputStream create(StoragePath path, boolean overwrite) throws IOException { + return new FilterOutputStream(super.create(path, overwrite)) { + @Override + public void close() throws IOException { + super.close(); + throw new IOException("close failure"); + } + }; + } + } + + private static class RenameFailingHoodieHadoopStorage extends HoodieHadoopStorage { + RenameFailingHoodieHadoopStorage(FileSystem fileSystem) { + super(fileSystem); + } + + @Override + public boolean rename(StoragePath oldPath, StoragePath newPath) { + throw new HoodieException("rename failure"); + } + } + + private static class CleanupFailingHoodieHadoopStorage extends HoodieHadoopStorage { + CleanupFailingHoodieHadoopStorage(FileSystem fileSystem) { + super(fileSystem); + } + + @Override + public boolean deleteFile(StoragePath path) { + throw new HoodieIOException("cleanup failure"); + } + + void deleteTemporaryFileAfterTest(StoragePath path) throws IOException { + super.deleteFile(path); + } + } } diff --git a/hudi-io/src/main/java/org/apache/hudi/storage/HoodieStorage.java b/hudi-io/src/main/java/org/apache/hudi/storage/HoodieStorage.java index 004d88b66183..2abdfb55d02b 100644 --- a/hudi-io/src/main/java/org/apache/hudi/storage/HoodieStorage.java +++ b/hudi-io/src/main/java/org/apache/hudi/storage/HoodieStorage.java @@ -23,6 +23,7 @@ import org.apache.hudi.ApiMaturityLevel; import org.apache.hudi.PublicAPIClass; import org.apache.hudi.PublicAPIMethod; import org.apache.hudi.common.util.Option; +import org.apache.hudi.exception.HoodieException; import org.apache.hudi.exception.HoodieIOException; import org.apache.hudi.io.SeekableDataInputStream; @@ -329,58 +330,49 @@ public abstract class HoodieStorage implements Closeable { public final void createImmutableFileInPath(StoragePath path, Option<HoodieInstantWriter> contentWriter, boolean needTempFile) throws HoodieIOException { - OutputStream fsout = null; StoragePath tmpPath = null; + StoragePath pathToCreate = path; + if (contentWriter.isPresent() && needTempFile) { + StoragePath parent = path.getParent(); + tmpPath = new StoragePath(parent, path.getName() + "." + UUID.randomUUID()); + pathToCreate = tmpPath; + } + boolean fileCreated = false; + HoodieException failure = null; try { - if (!contentWriter.isPresent()) { - fsout = create(path, false); - } - - if (contentWriter.isPresent() && needTempFile) { - StoragePath parent = path.getParent(); - tmpPath = new StoragePath(parent, path.getName() + "." + UUID.randomUUID()); - fsout = create(tmpPath, false); - contentWriter.get().writeToStream(fsout); + try (OutputStream fsout = create(pathToCreate, false)) { + if (contentWriter.isPresent()) { + contentWriter.get().writeToStream(fsout); + } } - - if (contentWriter.isPresent() && !needTempFile) { - fsout = create(path, false); - contentWriter.get().writeToStream(fsout); + fileCreated = tmpPath == null || rename(tmpPath, path); + } catch (Throwable t) { + if (t instanceof HoodieException) { + failure = (HoodieException) t; + } else if (t instanceof IOException) { + failure = new HoodieIOException("Failed to create immutable file " + path, (IOException) t); + } else { + failure = new HoodieException("Failed to create immutable file " + path, t); } - } catch (IOException e) { - String errorMsg = "Failed to create file " + (tmpPath != null ? tmpPath : path); - throw new HoodieIOException(errorMsg, e); + throw failure; } finally { - try { - if (null != fsout) { - fsout.close(); - } - } catch (IOException e) { - String errorMsg = "Failed to close file " + (needTempFile ? tmpPath : path); - throw new HoodieIOException(errorMsg, e); - } - - boolean renameSuccess = false; - try { - if (null != tmpPath) { - renameSuccess = rename(tmpPath, path); - } - } catch (IOException e) { - throw new HoodieIOException( - "Failed to rename " + tmpPath + " to the target " + path, - e); - } finally { - if (!renameSuccess && null != tmpPath) { - try { - deleteFile(tmpPath); - LOG.debug("Failed to rename {} to {}, target file exists: {}", tmpPath, path, exists(path)); - } catch (IOException e) { - throw new HoodieIOException("Failed to delete tmp file " + tmpPath, e); + if (tmpPath != null && !fileCreated) { + try { + deleteFile(tmpPath); + } catch (Throwable t) { + if (failure != null) { + failure.addSuppressed(t); + } else { + throw new HoodieException("Failed to delete temporary file " + tmpPath, t); } } } } + + if (!fileCreated) { + LOG.debug("Failed to rename {} to {}; target file may already exist", tmpPath, path); + } } /** diff --git a/hudi-io/src/test/java/org/apache/hudi/io/storage/TestHoodieStorageBase.java b/hudi-io/src/test/java/org/apache/hudi/io/storage/TestHoodieStorageBase.java index ee66a3d366bd..68cce569b41c 100644 --- a/hudi-io/src/test/java/org/apache/hudi/io/storage/TestHoodieStorageBase.java +++ b/hudi-io/src/test/java/org/apache/hudi/io/storage/TestHoodieStorageBase.java @@ -20,6 +20,7 @@ package org.apache.hudi.io.storage; import org.apache.hudi.common.util.Option; +import org.apache.hudi.exception.HoodieIOException; import org.apache.hudi.io.SeekableDataInputStream; import org.apache.hudi.io.util.IOUtils; import org.apache.hudi.storage.HoodieInstantWriter; @@ -151,6 +152,24 @@ public abstract class TestHoodieStorageBase { assertTrue(storage.createDirectory(path4)); } + @Test + public void testImmutableFileIsNotPublishedOnWriteFailure() throws IOException { + HoodieStorage storage = getStorage(); + StoragePath directory = new StoragePath(getTempDir(), "testImmutableFileWriteFailure"); + StoragePath path = new StoragePath(directory, "1.file"); + storage.createDirectory(directory); + + HoodieIOException exception = assertThrows(HoodieIOException.class, + () -> storage.createImmutableFileInPath(path, Option.of(outputStream -> { + outputStream.write(42); + throw new IOException("write failure"); + }))); + + assertEquals("write failure", exception.getCause().getMessage()); + assertFalse(storage.exists(path)); + assertTrue(storage.listDirectEntries(directory).isEmpty()); + } + @Test public void testSeekable() throws IOException { HoodieStorage storage = getStorage();
