Fetch all the locks for a shard to avoid multiple remote store calls

server/src/main/java/org/opensearch/index/store/RemoteSegmentStoreDirectory.java

@@ -749,20 +749,16 @@ public void deleteStaleSegments(int lastNMetadataFilesToKeep) throws IOException
             lastNMetadataFilesToKeep,
             sortedMetadataFileList.size()
         );
-        List<String> metadataFilesToBeDeleted = metadataFilesEligibleToDelete.stream().filter(metadataFile -> {
-            try {
-                return !isLockAcquired(metadataFile);
-            } catch (IOException e) {
-                logger.error(
-                    "skipping metadata file ("
-                        + metadataFile
-                        + ") deletion for this run,"
-                        + " as checking lock for metadata is failing with error: "
-                        + e
-                );
-                return false;
-            }
-        }).collect(Collectors.toList());
+        Set<String> allLockFiles;
+        try {
+            allLockFiles = new HashSet<>(mdLockManager.listAllLocks(MetadataFilenameUtils.METADATA_PREFIX));
+        } catch (Exception e) {
+            logger.error("Exception while fetching segment metadata lock files, skipping deleteStaleSegments", e);
+            return;
+        }
+        List<String> metadataFilesToBeDeleted = metadataFilesEligibleToDelete.stream()
+            .filter(metadataFile -> allLockFiles.contains(metadataFile) == false)
+            .collect(Collectors.toList());
 
         sortedMetadataFileList.removeAll(metadataFilesToBeDeleted);
         logger.debug(

server/src/main/java/org/opensearch/index/store/lockmanager/RemoteStoreLockManager.java

@@ -11,6 +11,7 @@
 import org.opensearch.common.annotation.PublicApi;
 
 import java.io.IOException;
+import java.util.Collection;
 
 /**
  * An Interface that defines Remote Store Lock Manager.
@@ -43,6 +44,8 @@ public interface RemoteStoreLockManager {
      */
     Boolean isAcquired(LockInfo lockInfo) throws IOException;
 
+    public Collection<String> listAllLocks(String prefix) throws IOException;
+
     /**
      * Acquires lock on the file mentioned in originalLockInfo for acquirer mentioned in clonedLockInfo.
      * There can occur a race condition where the original file is deleted before we can use it to acquire lock for the new acquirer. Until we have a

server/src/main/java/org/opensearch/index/store/lockmanager/RemoteStoreMetadataLockManager.java

@@ -101,6 +101,10 @@ public Boolean isAcquired(LockInfo lockInfo) throws IOException {
         return !lockFiles.isEmpty();
     }
 
+    public Collection<String> listAllLocks(String prefix) throws IOException {
+        return lockDirectory.listFilesByPrefix(prefix);
+    }
+
     /**
      * Acquires lock on the file mentioned in originalLockInfo for acquirer mentioned in clonedLockInfo.
      * Snapshot layer enforces thread safety by having checks in place to ensure that the source snapshot is not being deleted before proceeding