diff --git a/server/src/main/java/org/opensearch/repositories/blobstore/RemoteStoreShardCleanupTask.java b/server/src/main/java/org/opensearch/repositories/blobstore/RemoteStoreShardCleanupTask.java index df61c1ca3263b..71f2606af0ce3 100644 --- a/server/src/main/java/org/opensearch/repositories/blobstore/RemoteStoreShardCleanupTask.java +++ b/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 ongoingRemoteDirectoryCleanups = newConcurrentSet(); - final static Map 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. - // - 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); } } } diff --git a/server/src/test/java/org/opensearch/repositories/blobstore/BlobStoreRepositoryTests.java b/server/src/test/java/org/opensearch/repositories/blobstore/BlobStoreRepositoryTests.java index b76e01d6d4c82..2445cad01574c 100644 --- a/server/src/test/java/org/opensearch/repositories/blobstore/BlobStoreRepositoryTests.java +++ b/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); } }