[core] Check the commit lease after the snapshot existence check

paimon-core/src/main/java/org/apache/paimon/catalog/RenamingSnapshotCommit.java

@@ -94,6 +94,11 @@ protected boolean commitImpl(
                 };
         // fs.rename may not return false if the target file already exists, or even be atomic.
         // The bound lock lets us check existence before publishing.
-        return !fileIO.exists(newSnapshotPath) && callable.call();
+        if (fileIO.exists(newSnapshotPath)) {
+            return false;
+        }
+        // Checking existence may outlast the lease, so check it again before writing.
+        ensureValid();
+        return callable.call();
     }
 }

paimon-core/src/test/java/org/apache/paimon/catalog/RenamingSnapshotCommitTest.java

@@ -19,21 +19,43 @@
 package org.apache.paimon.catalog;
 
 import org.apache.paimon.Snapshot;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.fs.local.LocalFileIO;
 import org.apache.paimon.operation.Lock;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.schema.FileSystemSchemaManager;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.CatalogEnvironment;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.types.DataTypes;
 import org.apache.paimon.utils.HintFileUtils;
 import org.apache.paimon.utils.SnapshotManager;
 
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 
+import javax.annotation.Nullable;
+
 import java.io.IOException;
 import java.nio.file.Files;
+import java.nio.file.StandardCopyOption;
+import java.util.ArrayList;
 import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
 /**
@@ -146,6 +168,129 @@ public void testCommitRenameThrowsAndTargetMissing(@TempDir java.nio.file.Path t
         assertThat(fileIO.exists(snapshotManager.snapshotPath(snapshot.id()))).isFalse();
     }
 
+    /**
+     * The lease of the commit lock expires while the commit checks whether the snapshot file
+     * already exists. The commit must not publish the snapshot after that.
+     */
+    @Test
+    public void testCommitRechecksLeaseAfterCheckingSnapshot(@TempDir java.nio.file.Path tmp)
+            throws Exception {
+        AtomicBoolean expired = new AtomicBoolean();
+        Path tablePath = new Path(tmp.toUri());
+        SnapshotManager snapshotManager =
+                new SnapshotManager(new LocalFileIO(), tablePath, null, null, null);
+        Path snapshotPath = snapshotManager.snapshotPath(5L);
+        FileIO fileIO =
+                new LocalFileIO() {
+                    @Override
+                    public boolean exists(Path path) throws IOException {
+                        boolean exists = super.exists(path);
+                        if (path.equals(snapshotPath)) {
+                            expired.set(true);
+                        }
+                        return exists;
+                    }
+                };
+        snapshotManager = new SnapshotManager(fileIO, tablePath, null, null, null);
+        Lock lock =
+                new Lock() {
+                    @Override
+                    public AutoCloseable acquire() {
+                        return () -> {};
+                    }
+
+                    @Override
+                    public void ensureValid() {
+                        if (expired.get()) {
+                            throw new IllegalStateException("Lease expired");
+                        }
+                    }
+
+                    @Override
+                    public void close() {}
+                };
+
+        RenamingSnapshotCommit commit = new RenamingSnapshotCommit(snapshotManager, lock);
+        Snapshot snapshot = createSnapshot(5L);
+        // thrown by the check in commitImpl, not wrapped like a failure to acquire the lock
+        assertThatThrownBy(() -> commit.commit(null, snapshot, "main", Collections.emptyList()))
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessage("Lease expired");
+
+        assertThat(new LocalFileIO().exists(snapshotPath)).isFalse();
+    }
+
+    /**
+     * Two writers share a lease lock on a file system whose rename overwrites the target, as some
+     * object stores do. Writer A's lease expires while it checks whether snapshot 2 exists, and
+     * writer B, which gets the lock then, commits snapshot 2. Writer A must not overwrite it.
+     */
+    @Test
+    public void testExpiredLeaseDoesNotOverwriteAnotherWritersSnapshot(
+            @TempDir java.nio.file.Path tmp) throws Exception {
+        Path tablePath = new Path(tmp.toUri());
+        OverwritingRenameLocalFileIO fileIO = new OverwritingRenameLocalFileIO();
+        new FileSystemSchemaManager(fileIO, tablePath)
+                .createTable(Schema.newBuilder().column("id", DataTypes.INT()).build());
+        LeaseLockFactory locks = new LeaseLockFactory();
+        FileStoreTable writerA = createTable(fileIO, tablePath, locks);
+        FileStoreTable writerB = createTable(fileIO, tablePath, locks);
+        commitRow(writerA, 1);
+
+        fileIO.onCommitExists(
+                writerA.snapshotManager().snapshotPath(2),
+                () -> {
+                    locks.expireHolder();
+                    Thread writer = new Thread(() -> commitRow(writerB, 2));
+                    writer.start();
+                    writer.join();
+                });
+        commitRow(writerA, 3);
+
+        assertThat(readIds(FileStoreTableFactory.create(fileIO, tablePath)))
+                .containsExactlyInAnyOrder(1, 2, 3);
+    }
+
+    private static FileStoreTable createTable(
+            FileIO fileIO, Path tablePath, CatalogLockFactory lockFactory) {
+        CatalogEnvironment env =
+                new CatalogEnvironment(
+                        Identifier.create("db", "t"),
+                        null,
+                        null,
+                        lockFactory,
+                        null,
+                        null,
+                        false,
+                        false);
+        return FileStoreTableFactory.create(
+                fileIO,
+                tablePath,
+                new FileSystemSchemaManager(fileIO, tablePath).latest().get(),
+                env);
+    }
+
+    private static void commitRow(FileStoreTable table, int id) {
+        BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+        try (BatchTableWrite write = writeBuilder.newWrite();
+                BatchTableCommit commit = writeBuilder.newCommit()) {
+            write.write(GenericRow.of(id));
+            commit.commit(write.prepareCommit());
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private static List<Integer> readIds(FileStoreTable table) throws IOException {
+        ReadBuilder readBuilder = table.newReadBuilder();
+        List<Integer> ids = new ArrayList<>();
+        try (RecordReader<InternalRow> reader =
+                readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+            reader.forEachRemaining(row -> ids.add(row.getInt(0)));
+        }
+        return ids;
+    }
+
     private static Snapshot createSnapshot(long id) throws IOException {
         long schemaId = 1L;
         String baseManifestList = "manifest-list-base";
@@ -217,6 +362,106 @@ public boolean rename(Path src, Path dst) throws IOException {
         }
     }
 
+    /**
+     * A LocalFileIO whose rename replaces an existing target, and that can run an action once when
+     * a snapshot commit checks whether a file exists.
+     */
+    private static class OverwritingRenameLocalFileIO extends LocalFileIO {
+
+        private volatile Path trigger;
+        private volatile ThrowingRunnable action;
+
+        void onCommitExists(Path path, ThrowingRunnable action) {
+            this.action = action;
+            this.trigger = path;
+        }
+
+        @Override
+        public boolean exists(Path path) throws IOException {
+            boolean exists = super.exists(path);
+            if (path.equals(trigger) && calledFromCommitImpl()) {
+                trigger = null;
+                try {
+                    action.run();
+                } catch (Exception e) {
+                    throw new IOException(e);
+                }
+            }
+            return exists;
+        }
+
+        @Override
+        public boolean rename(Path src, Path dst) throws IOException {
+            java.nio.file.Path target = java.nio.file.Paths.get(dst.toUri());
+            Files.createDirectories(target.getParent());
+            Files.move(
+                    java.nio.file.Paths.get(src.toUri()),
+                    target,
+                    StandardCopyOption.REPLACE_EXISTING);
+            return true;
+        }
+
+        private static boolean calledFromCommitImpl() {
+            for (StackTraceElement e : Thread.currentThread().getStackTrace()) {
+                if (e.getClassName().startsWith(RenamingSnapshotCommit.class.getName())
+                        && e.getMethodName().equals("commitImpl")) {
+                    return true;
+                }
+            }
+            return false;
+        }
+    }
+
+    private interface ThrowingRunnable {
+        void run() throws Exception;
+    }
+
+    /**
+     * Hands out lease locks that exclude each other. {@link #expireHolder} ends the current
+     * holder's lease, so another lock can be acquired and the holder fails {@code ensureValid}.
+     */
+    private static class LeaseLockFactory implements CatalogLockFactory {
+
+        private final AtomicReference<Lock> holder = new AtomicReference<>();
+
+        void expireHolder() {
+            holder.set(null);
+        }
+
+        @Override
+        public String identifier() {
+            return "lease";
+        }
+
+        @Override
+        public Lock createLock(
+                CatalogLockContext context,
+                Identifier identifier,
+                @Nullable String tableUuid,
+                @Nullable String commitUser,
+                Options tableOptions) {
+            return new Lock() {
+                @Override
+                public AutoCloseable acquire() {
+                    if (!holder.compareAndSet(null, this)) {
+                        throw new IllegalStateException("The lock is held by another writer");
+                    }
+                    return () -> holder.compareAndSet(this, null);
+                }
+
+                @Override
+                public void ensureValid() {
+                    if (holder.get() != this) {
+                        throw new IllegalStateException("Lease expired");
+                    }
+                }
+
+                @Override
+                public void close() {}
+            };
+        }
+    }
+
     /**
      * A FileIO based on LocalFileIO that always throws from rename().
      *