This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 5a11a4fce206 fix(storage-format): ignore temporary LSM timeline
manifests (#19659)
5a11a4fce206 is described below
commit 5a11a4fce20620afac143358bc771dfe2d14d3c8
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
---
.../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 4a6680904f0b..4e216f451524 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
@@ -109,8 +109,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");
@@ -241,7 +240,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);
}
/**
@@ -297,6 +300,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 2ec0d048b45d..13fdd1c932ad 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 090e3f956787..380967dea129 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.FileIOUtils;
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();
