This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 54c00628fbe3e00eff98eb0ec750c5ea9df1b8f1 Author: Dapeng Sun(孙大鹏) <[email protected]> AuthorDate: Thu Sep 17 14:39:53 2026 +0800 [common] Fix snapshot and tag reads failing when the file is deleted concurrently (#9851) --- .../src/main/java/org/apache/paimon/fs/FileIO.java | 28 ++++++ .../test/java/org/apache/paimon/fs/FileIOTest.java | 104 +++++++++++++++++++++ .../apache/paimon/utils/SnapshotManagerTest.java | 21 +++++ .../org/apache/paimon/utils/TagManagerTest.java | 22 +++++ 4 files changed, 175 insertions(+) diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java index c20e2c70bb..c6e8d02a46 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java +++ b/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java @@ -35,9 +35,11 @@ import java.io.Closeable; import java.io.FileNotFoundException; import java.io.IOException; import java.io.InputStreamReader; +import java.io.InterruptedIOException; import java.io.OutputStreamWriter; import java.io.Serializable; import java.net.URI; +import java.nio.channels.ClosedByInterruptException; import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.ArrayList; @@ -341,6 +343,32 @@ public interface FileIO extends Serializable, Closeable { builder.append(line); } return builder.toString(); + } catch (FileNotFoundException e) { + throw e; + } catch (InterruptedIOException | ClosedByInterruptException e) { + // An interrupted read says nothing about whether the file is still there. + throw e; + } catch (IOException e) { + // Some object stores throw a plain IOException for a file deleted during reading. + boolean missing; + try { + missing = !exists(path); + } catch (IOException | RuntimeException checkFailure) { + // Keep the original failure when the file cannot be confirmed to be gone. + if (checkFailure != e) { + e.addSuppressed(checkFailure); + } + throw e; + } + if (!missing) { + throw e; + } + LOG.debug( + "Read of {} failed and the file is gone, reporting it as not found.", path, e); + FileNotFoundException notFound = + new FileNotFoundException("File " + path + " does not exist."); + notFound.initCause(e); + throw notFound; } } diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java index 96e023d1eb..6642f7b62a 100644 --- a/paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java @@ -25,10 +25,13 @@ import org.apache.paimon.options.Options; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.mockito.Mockito; import java.io.File; import java.io.FileNotFoundException; import java.io.IOException; +import java.io.InterruptedIOException; +import java.nio.channels.ClosedByInterruptException; import java.nio.file.AccessDeniedException; import java.nio.file.DirectoryNotEmptyException; import java.nio.file.FileAlreadyExistsException; @@ -51,6 +54,107 @@ public class FileIOTest { @TempDir java.nio.file.Path tempDir; + @Test + public void testReadFileUtf8PropagatesInterruptedRead() throws IOException { + FileIO fileIO = Mockito.spy(LocalFileIO.create()); + Path file = new Path(tempDir.resolve("file").toUri()); + fileIO.writeFile(file, "content", false); + InterruptedIOException interrupted = new InterruptedIOException("Interrupted"); + Mockito.doAnswer( + invocation -> { + fileIO.deleteQuietly(file); + throw interrupted; + }) + .when(fileIO) + .newInputStream(file); + Mockito.clearInvocations(fileIO); + + assertThatThrownBy(() -> fileIO.readFileUtf8(file)).isSameAs(interrupted); + Mockito.verify(fileIO, Mockito.never()).exists(file); + + Path closed = new Path(tempDir.resolve("closed").toUri()); + fileIO.writeFile(closed, "content", false); + ClosedByInterruptException closedByInterrupt = new ClosedByInterruptException(); + Mockito.doAnswer( + invocation -> { + fileIO.deleteQuietly(closed); + throw closedByInterrupt; + }) + .when(fileIO) + .newInputStream(closed); + + assertThatThrownBy(() -> fileIO.readFileUtf8(closed)).isSameAs(closedByInterrupt); + } + + @Test + public void testReadFileUtf8KeepsReadFailureReusedByExistenceCheck() throws IOException { + FileIO fileIO = Mockito.spy(LocalFileIO.create()); + Path file = new Path(tempDir.resolve("file").toUri()); + fileIO.writeFile(file, "content", false); + IOException failure = new IOException("Cached failure"); + Mockito.doThrow(failure).when(fileIO).newInputStream(file); + Mockito.doThrow(failure).when(fileIO).exists(file); + + assertThatThrownBy(() -> fileIO.readFileUtf8(file)) + .isSameAs(failure) + .satisfies(e -> assertThat(e.getSuppressed()).isEmpty()); + } + + @Test + public void testReadFileUtf8ReportsFileDeletedDuringReadAsNotFound() throws IOException { + FileIO fileIO = Mockito.spy(LocalFileIO.create()); + Path file = new Path(tempDir.resolve("file").toUri()); + fileIO.writeFile(file, "content", false); + IOException readFailure = new IOException("404 Not Found"); + Mockito.doAnswer( + invocation -> { + fileIO.deleteQuietly(file); + throw readFailure; + }) + .when(fileIO) + .newInputStream(file); + + assertThatThrownBy(() -> fileIO.readFileUtf8(file)) + .isInstanceOf(FileNotFoundException.class) + .hasCause(readFailure); + } + + @Test + public void testReadFileUtf8KeepsReadFailureOfExistingFile() throws IOException { + FileIO fileIO = Mockito.spy(LocalFileIO.create()); + Path file = new Path(tempDir.resolve("file").toUri()); + fileIO.writeFile(file, "content", false); + IOException readFailure = new IOException("Read failure"); + Mockito.doThrow(readFailure).when(fileIO).newInputStream(file); + + assertThatThrownBy(() -> fileIO.readFileUtf8(file)).isSameAs(readFailure); + } + + @Test + public void testReadFileUtf8KeepsReadFailureWhenExistenceCheckFails() throws IOException { + FileIO fileIO = Mockito.spy(LocalFileIO.create()); + Path file = new Path(tempDir.resolve("file").toUri()); + fileIO.writeFile(file, "content", false); + IOException readFailure = new IOException("Read failure"); + Mockito.doThrow(readFailure).when(fileIO).newInputStream(file); + IOException checkFailure = new IOException("Exists failure"); + Mockito.doThrow(checkFailure).when(fileIO).exists(file); + + assertThatThrownBy(() -> fileIO.readFileUtf8(file)) + .isSameAs(readFailure) + .satisfies(e -> assertThat(e.getSuppressed()).containsExactly(checkFailure)); + + IOException secondReadFailure = new IOException("Read failure"); + Mockito.doThrow(secondReadFailure).when(fileIO).newInputStream(file); + RuntimeException uncheckedCheckFailure = new RuntimeException("Exists failure"); + Mockito.doThrow(uncheckedCheckFailure).when(fileIO).exists(file); + + assertThatThrownBy(() -> fileIO.readFileUtf8(file)) + .isSameAs(secondReadFailure) + .satisfies( + e -> assertThat(e.getSuppressed()).containsExactly(uncheckedCheckFailure)); + } + @Test public void testRequireOptions() throws IOException { Options options = new Options(); diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java index c6dddc6dca..fb09951952 100644 --- a/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java @@ -140,6 +140,27 @@ public class SnapshotManagerTest { } } + @Test + public void testLatestSnapshotOfUserStopsAtSnapshotDeletedDuringRead() throws IOException { + FileIO fileIO = Mockito.spy(LocalFileIO.create()); + SnapshotManager snapshotManager = newSnapshotManager(fileIO, new Path(tempDir.toString())); + for (long id = 1; id <= 3; id++) { + fileIO.tryToWriteAtomic( + snapshotManager.snapshotPath(id), + createSnapshotWithMillis(id, id * 1000).toJson()); + } + Path expiring = snapshotManager.snapshotPath(1); + Mockito.doAnswer( + invocation -> { + fileIO.deleteQuietly(expiring); + throw new IOException("404 Not Found"); + }) + .when(fileIO) + .newInputStream(expiring); + + assertThat(snapshotManager.latestSnapshotOfUser("currentCommitUser")).isEmpty(); + } + @ParameterizedTest @ValueSource(booleans = {true, false}) public void testEarliestSnapshot(boolean isRaceCondition) throws IOException { diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/TagManagerTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/TagManagerTest.java index ede52e6ef3..4a899950ea 100644 --- a/paimon-core/src/test/java/org/apache/paimon/utils/TagManagerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/utils/TagManagerTest.java @@ -40,7 +40,9 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.mockito.Mockito; +import java.io.IOException; import java.time.Duration; import java.util.Arrays; import java.util.Collections; @@ -51,6 +53,7 @@ import java.util.concurrent.ThreadLocalRandom; import static org.apache.paimon.operation.FileStoreTestUtils.commitData; import static org.apache.paimon.operation.FileStoreTestUtils.partitionedData; +import static org.apache.paimon.utils.SnapshotManagerTest.createSnapshotWithMillis; import static org.assertj.core.api.Assertions.assertThat; /** Tests for TagManager. */ @@ -71,6 +74,25 @@ public class TagManagerTest { tagManager = null; } + @Test + public void testGetTagDeletedDuringRead() throws IOException { + FileIO spyFileIO = Mockito.spy(LocalFileIO.create()); + tagManager = new TagManager(spyFileIO, new Path(root)); + Path path = tagManager.tagPath("tag"); + spyFileIO.tryToWriteAtomic( + path, + Tag.fromSnapshotAndTagTtl(createSnapshotWithMillis(1, 1000), null, null).toJson()); + Mockito.doAnswer( + invocation -> { + spyFileIO.deleteQuietly(path); + throw new IOException("404 Not Found"); + }) + .when(spyFileIO) + .newInputStream(path); + + assertThat(tagManager.get("tag")).isEmpty(); + } + @Test public void testCreateTagWithoutTimeRetained() throws Exception { TestFileStore store = createStore(TestKeyValueGenerator.GeneratorMode.NON_PARTITIONED, 4);
