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 c20e2c70bb18..c6e8d02a46ce 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.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 @@ default String readFileUtf8(Path path) throws IOException { 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 96e023d1eb46..6642f7b62a92 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.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 c6dddc6dcab9..fb09951952fd 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 void testSnapshotExistsRestoresInterruptedStatus() throws IOException { } } + @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 ede52e6ef3ab..4a899950ea3f 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.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 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 void setup() throws Exception { 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);