[Bug Fix] Simplify remote directory cleanup after snapshot delete to …

…avoid concurrent cleanup task runs for same shard.

Signed-off-by: Harish Bhakuni <[email protected]>

server/src/main/java/org/opensearch/repositories/blobstore/RemoteStoreShardCleanupTask.java

@@ -12,10 +12,8 @@
 import org.apache.logging.log4j.Logger;
 import org.opensearch.core.index.shard.ShardId;
 
-import java.util.Map;
 import java.util.Set;
 
-import static org.opensearch.common.util.concurrent.ConcurrentCollections.newConcurrentMap;
 import static org.opensearch.common.util.concurrent.ConcurrentCollections.newConcurrentSet;
 
 /**
@@ -25,7 +23,6 @@ public class RemoteStoreShardCleanupTask implements Runnable {
     private final Runnable task;
     private final String shardIdentifier;
     final static Set<String> ongoingRemoteDirectoryCleanups = newConcurrentSet();
-    final static Map<String, Runnable> pendingRemoteDirectoryCleanups = newConcurrentMap();
     private static final Logger staticLogger = LogManager.getLogger(RemoteStoreShardCleanupTask.class);
 
     public RemoteStoreShardCleanupTask(Runnable task, String indexUUID, ShardId shardId) {
@@ -39,25 +36,16 @@ private static String indexShardIdentifier(String indexUUID, ShardId shardId) {
 
     @Override
     public void run() {
-        // TODO: this is the best effort at the moment since there is still a known race condition scenario in this
-        // method which needs to be handled where one of the thread just came out of while loop and removed the
-        // entry from ongoingRemoteDirectoryCleanup, and another thread added new pending task in the map.
-        // we need to introduce semaphores/locks to avoid that situation which introduces the overhead of lock object
-        // cleanups. however, there will be no scenario where two threads run cleanup for same shard at same time.
-        // <issue-link>
-        if (pendingRemoteDirectoryCleanups.put(shardIdentifier, task) == null) {
-            if (ongoingRemoteDirectoryCleanups.add(shardIdentifier)) {
-                while (pendingRemoteDirectoryCleanups.containsKey(shardIdentifier)) {
-                    Runnable newTask = pendingRemoteDirectoryCleanups.get(shardIdentifier);
-                    pendingRemoteDirectoryCleanups.remove(shardIdentifier);
-                    newTask.run();
-                }
-                ongoingRemoteDirectoryCleanups.remove(shardIdentifier);
-            } else {
-                staticLogger.debug("one task is already ongoing for shard {}, we can leave entry in pending", shardIdentifier);
+        // If there is already a same task ongoing for a shard, we need to skip the new task to avoid multiple
+        //  concurrent cleanup of same shard.
+        if (ongoingRemoteDirectoryCleanups.add(shardIdentifier)) {
+            try {
+                task.run();
+            } finally {
+            ongoingRemoteDirectoryCleanups.remove(shardIdentifier);
             }
         } else {
-            staticLogger.debug("one cleanup task for shard {} is already in pending, we can skip this task", shardIdentifier);
+            staticLogger.warn("one cleanup task for shard {} is already ongoing, need to skip this task", shardIdentifier);
         }
     }
 }

server/src/test/java/org/opensearch/repositories/blobstore/BlobStoreRepositoryTests.java

@@ -321,38 +321,31 @@ private RepositoryData addRandomSnapshotsToRepoData(RepositoryData repoData, boo
         return repoData;
     }
 
+    private String getShardIdentifier(String indexUUID, String shardId) {
+        return String.join("/", indexUUID, shardId);
+    }
+
     public void testRemoteStoreShardCleanupTask() {
-        // todo: move it to separate class and add more scenarios.
         AtomicBoolean executed1 = new AtomicBoolean(false);
         Runnable task1 = () -> executed1.set(true);
         String indexName = "test-idx";
         String testIndexUUID = "test-idx-uuid";
         ShardId shardId = new ShardId(new Index(indexName, testIndexUUID), 0);
 
-        // Scenario 1: pending = empty, ongoing = false => executed
+        // just adding random shards in ongoing cleanups.
+        RemoteStoreShardCleanupTask.ongoingRemoteDirectoryCleanups.add(getShardIdentifier(testIndexUUID, "1"));
+        RemoteStoreShardCleanupTask.ongoingRemoteDirectoryCleanups.add(getShardIdentifier(testIndexUUID, "2"));
+
+        // Scenario 1: ongoing = false => executed
         RemoteStoreShardCleanupTask remoteStoreShardCleanupTask = new RemoteStoreShardCleanupTask(task1, testIndexUUID, shardId);
         remoteStoreShardCleanupTask.run();
         assertTrue(executed1.get());
 
-        // Scenario 2: pending = empty, ongoing = true => pending = currentTask
+        // Scenario 2: ongoing = true => currentTask skipped.
         executed1.set(false);
-        String shardIdentifier = String.join("/", testIndexUUID, String.valueOf(shardId.id()));
-        RemoteStoreShardCleanupTask.ongoingRemoteDirectoryCleanups.add(shardIdentifier);
-
+        RemoteStoreShardCleanupTask.ongoingRemoteDirectoryCleanups.add(getShardIdentifier(testIndexUUID, "0"));
         remoteStoreShardCleanupTask = new RemoteStoreShardCleanupTask(task1, testIndexUUID, shardId);
         remoteStoreShardCleanupTask.run();
         assertFalse(executed1.get());
-        assertSame(RemoteStoreShardCleanupTask.pendingRemoteDirectoryCleanups.get(shardIdentifier), task1);
-
-        // Scenario3: pending = anotherTask, ongoing = true => pending = currentTask
-        AtomicBoolean executed2 = new AtomicBoolean(false);
-        Runnable task2 = () -> executed2.set(true);
-        RemoteStoreShardCleanupTask.pendingRemoteDirectoryCleanups.put(shardIdentifier, task1);
-        RemoteStoreShardCleanupTask.ongoingRemoteDirectoryCleanups.add(shardIdentifier);
-
-        remoteStoreShardCleanupTask = new RemoteStoreShardCleanupTask(task2, testIndexUUID, shardId);
-        remoteStoreShardCleanupTask.run();
-        assertFalse(executed1.get());
-        assertSame(RemoteStoreShardCleanupTask.pendingRemoteDirectoryCleanups.get(shardIdentifier), task2);
     }
 }