[server] Added thread-safety to CoordinatorEventProcessor for Completed Snapshots (#2221)

Author: rionmonster

fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStore.java

@@ -26,6 +26,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.concurrent.ThreadSafe;
+
 import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -35,8 +37,10 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.Executor;
+import java.util.concurrent.locks.ReentrantLock;
 
 import static org.apache.fluss.utils.Preconditions.checkNotNull;
+import static org.apache.fluss.utils.concurrent.LockUtils.inLock;
 
 /* This file is based on source code of Apache Flink Project (https://flink.apache.org/), licensed by the Apache
  * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
@@ -47,6 +51,7 @@
  * managing the completed snapshots including store/subsume/get completed snapshots for single one
  * table bucket.
  */
+@ThreadSafe
 public class CompletedSnapshotStore {
 
     private static final Logger LOG = LoggerFactory.getLogger(CompletedSnapshotStore.class);
@@ -62,6 +67,8 @@ public class CompletedSnapshotStore {
     private final Executor ioExecutor;
     private final SnapshotsCleaner snapshotsCleaner;
 
+    private final ReentrantLock lock = new ReentrantLock();
+
     /**
      * Local copy of the completed snapshots in snapshot store. This is restored from snapshot
      * handel store when recovering.
@@ -84,15 +91,19 @@ public CompletedSnapshotStore(
     }
 
     public void add(final CompletedSnapshot completedSnapshot) throws Exception {
-        addSnapshotAndSubsumeOldestOne(completedSnapshot, snapshotsCleaner, () -> {});
+        inLock(
+                lock,
+                () ->
+                        addSnapshotAndSubsumeOldestOne(
+                                completedSnapshot, snapshotsCleaner, () -> {}));
     }
 
     public long getPhysicalStorageRemoteKvSize() {
         return sharedKvFileRegistry.getFileSize();
     }
 
     public long getNumSnapshots() {
-        return completedSnapshots.size();
+        return inLock(lock, () -> completedSnapshots.size());
     }
 
     /**
@@ -117,34 +128,43 @@ CompletedSnapshot addSnapshotAndSubsumeOldestOne(
                 snapshot.getTableBucket(), snapshot.getSnapshotID(), completedSnapshotHandle);
 
         // Now add the new one. If it fails, we don't want to lose existing data.
-        completedSnapshots.addLast(snapshot);
-
-        // Remove completed snapshot from queue and snapshotStateHandleStore, not discard.
-        Optional<CompletedSnapshot> subsume =
-                subsume(
-                        completedSnapshots,
-                        maxNumberOfSnapshotsToRetain,
-                        completedSnapshot -> {
-                            remove(
-                                    completedSnapshot.getTableBucket(),
-                                    completedSnapshot.getSnapshotID());
-                            snapshotsCleaner.addSubsumedSnapshot(completedSnapshot);
-                        });
-
-        findLowest(completedSnapshots)
-                .ifPresent(
-                        id -> {
-                            // unregister the unused kv file, which will then cause the kv file
-                            // deletion
-                            sharedKvFileRegistry.unregisterUnusedKvFile(id);
-                            snapshotsCleaner.cleanSubsumedSnapshots(
-                                    id, Collections.emptySet(), postCleanup, ioExecutor);
-                        });
-        return subsume.orElse(null);
+        return inLock(
+                lock,
+                () -> {
+                    completedSnapshots.addLast(snapshot);
+
+                    // Remove completed snapshot from queue and snapshotStateHandleStore, not
+                    // discard.
+                    Optional<CompletedSnapshot> subsume =
+                            subsume(
+                                    completedSnapshots,
+                                    maxNumberOfSnapshotsToRetain,
+                                    completedSnapshot -> {
+                                        remove(
+                                                completedSnapshot.getTableBucket(),
+                                                completedSnapshot.getSnapshotID());
+                                        snapshotsCleaner.addSubsumedSnapshot(completedSnapshot);
+                                    });
+
+                    findLowest(completedSnapshots)
+                            .ifPresent(
+                                    id -> {
+                                        // unregister the unused kv file, which will then cause the
+                                        // kv file
+                                        // deletion
+                                        sharedKvFileRegistry.unregisterUnusedKvFile(id);
+                                        snapshotsCleaner.cleanSubsumedSnapshots(
+                                                id,
+                                                Collections.emptySet(),
+                                                postCleanup,
+                                                ioExecutor);
+                                    });
+                    return subsume.orElse(null);
+                });
     }
 
     public List<CompletedSnapshot> getAllSnapshots() {
-        return new ArrayList<>(completedSnapshots);
+        return inLock(lock, () -> new ArrayList<>(completedSnapshots));
     }
 
     private static Optional<CompletedSnapshot> subsume(
@@ -211,7 +231,7 @@ protected static Optional<Long> findLowest(Deque<CompletedSnapshot> unSubsumedSn
      * added.
      */
     public Optional<CompletedSnapshot> getLatestSnapshot() {
-        return Optional.ofNullable(completedSnapshots.peekLast());
+        return inLock(lock, () -> Optional.ofNullable(completedSnapshots.peekLast()));
     }
 
     /**