This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 1376e1a938 [common] Fix snapshot and tag reads failing when the file
is deleted concurrently (#9851)
1376e1a938 is described below
commit 1376e1a9382d60baca5b588b7c62873374f0887b
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);
