Add Changes in Snapshot Delete Flow for remote store interoperability.

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

@@ -3093,11 +3093,13 @@ private ShardSnapshotMetaDeleteResult deleteFromShardSnapshotMeta(
             // Using survivingSnapshots instead of newSnapshotsList as shallow snapshots can be present which won't be part of
             // newSnapshotsList
             if (survivingSnapshots.isEmpty()) {
+                // No shallow copy or full copy snapshot is surviving.
                 return new ShardSnapshotMetaDeleteResult(indexId, snapshotShardId, ShardGenerations.DELETED_SHARD_GEN, blobs);
             } else {
                 final BlobStoreIndexShardSnapshots updatedSnapshots;
                 // If we have surviving non shallow snapshots, update index- file.
                 if (newSnapshotsList.size() > 0) {
+                    // Some full copy snapshots are surviving.
                     updatedSnapshots = new BlobStoreIndexShardSnapshots(newSnapshotsList);
                     if (indexGeneration < 0L) {
                         writtenGeneration = UUIDs.randomBase64UUID();
@@ -3107,8 +3109,11 @@ private ShardSnapshotMetaDeleteResult deleteFromShardSnapshotMeta(
                         writeShardIndexBlobAtomic(shardContainer, indexGeneration, updatedSnapshots);
                     }
                 } else {
+                    // Some shallow copy snapshots are surviving. In this case, since no full copy snapshots are present, we use
+                    // EMPTY BlobStoreIndexShardSnapshots for updatedSnapshots which is used in unusedBlobs to compute stale files,
+                    // and use DELETED_SHARD_GEN since index-N file would not be present anymore.
                     updatedSnapshots = BlobStoreIndexShardSnapshots.EMPTY;
-                    writtenGeneration = ShardGenerations.NEW_SHARD_GEN;
+                    writtenGeneration = ShardGenerations.DELETED_SHARD_GEN;
                 }
                 final Set<String> survivingSnapshotUUIDs = survivingSnapshots.stream().map(SnapshotId::getUUID).collect(Collectors.toSet());
                 return new ShardSnapshotMetaDeleteResult(

server/src/main/java/org/opensearch/snapshots/SnapshotsService.java

@@ -2231,57 +2231,34 @@ private void deleteSnapshotsFromRepository(
             final List<SnapshotId> snapshotIds = deleteEntry.getSnapshots();
             assert deleteEntry.state() == SnapshotDeletionsInProgress.State.STARTED : "incorrect state for entry [" + deleteEntry + "]";
             final Repository repository = repositoriesService.repository(deleteEntry.repository());
-            StepListener<Boolean> checkForShallowSnapshotStep = new StepListener<Boolean>();
-
-            // This check is done to preserve the bwc with repository implementations that don't support shallow snapshot
-            // and use the existing deleteSnapshots API of Repository plugin. For other repositories, we expect them to implement
-            // new delete API - deleteSnapshotsAndReleaseLockFiles or use the one implemented by BlobStoreRepository.
-            // TODO This can be improved by having this information (whether the repository contains any shallow copy snapshot)
-            // in the RepositoryData instead of fetching snapshot info for each snapshot and verifying.
-            threadPool.executor(ThreadPool.Names.SNAPSHOT).execute(new AbstractRunnable() {
-                @Override
-                public void onFailure(Exception e) {
-                    checkForShallowSnapshotStep.onFailure(e);
-                }
-
-                @Override
-                protected void doRun() {
-                    boolean cleanupRemoteStoreLockFiles = false;
-                    final List<SnapshotId> allSnapshotIds = repositoryData.getSnapshotIds().stream().collect(Collectors.toList());
-                    for (SnapshotId snapshotId : allSnapshotIds) {
-                        if (Boolean.TRUE.equals(repository.getSnapshotInfo(snapshotId).isRemoteStoreIndexShallowCopyEnabled())) {
-                            cleanupRemoteStoreLockFiles = true;
-                            break;
-                        }
-                    }
-                    checkForShallowSnapshotStep.onResponse(cleanupRemoteStoreLockFiles);
-                }
-            });
 
-            checkForShallowSnapshotStep.whenComplete(cleanupRemoteStoreLockFiles -> {
-                if (cleanupRemoteStoreLockFiles) {
-                    repository.deleteSnapshotsAndReleaseLockFiles(
-                        snapshotIds,
-                        repositoryData.getGenId(),
-                        minCompatibleVersion(minNodeVersion, repositoryData, snapshotIds),
-                        remoteStoreLockManagerFactory,
-                        ActionListener.wrap(updatedRepoData -> {
-                            logger.info("snapshots {} deleted", snapshotIds);
-                            removeSnapshotDeletionFromClusterState(deleteEntry, null, updatedRepoData);
-                        }, ex -> removeSnapshotDeletionFromClusterState(deleteEntry, ex, repositoryData))
-                    );
-                } else {
-                    repository.deleteSnapshots(
-                        snapshotIds,
-                        repositoryData.getGenId(),
-                        minCompatibleVersion(minNodeVersion, repositoryData, snapshotIds),
-                        ActionListener.wrap(updatedRepoData -> {
-                            logger.info("snapshots {} deleted", snapshotIds);
-                            removeSnapshotDeletionFromClusterState(deleteEntry, null, updatedRepoData);
-                        }, ex -> removeSnapshotDeletionFromClusterState(deleteEntry, ex, repositoryData))
-                    );
-                }
-            }, ex -> removeSnapshotDeletionFromClusterState(deleteEntry, ex, repositoryData));
+            // TODO: Relying on repository flag to decide delete flow may lead to shallow snapshot blobs not being taken up for cleanup
+            // when the repository currently have the flag disabled and we try to delete the shallow snapshots taken prior to disabling
+            // the flag. This can be improved by having the info whether there ever were any shallow snapshot present in this repository
+            // or not in RepositoryData.
+            final boolean cleanupRemoteStoreLockFiles = REMOTE_STORE_INDEX_SHALLOW_COPY.get(repository.getMetadata().settings());
+            if (cleanupRemoteStoreLockFiles) {
+                repository.deleteSnapshotsAndReleaseLockFiles(
+                    snapshotIds,
+                    repositoryData.getGenId(),
+                    minCompatibleVersion(minNodeVersion, repositoryData, snapshotIds),
+                    remoteStoreLockManagerFactory,
+                    ActionListener.wrap(updatedRepoData -> {
+                        logger.info("snapshots {} deleted", snapshotIds);
+                        removeSnapshotDeletionFromClusterState(deleteEntry, null, updatedRepoData);
+                    }, ex -> removeSnapshotDeletionFromClusterState(deleteEntry, ex, repositoryData))
+                );
+            } else {
+                repository.deleteSnapshots(
+                    snapshotIds,
+                    repositoryData.getGenId(),
+                    minCompatibleVersion(minNodeVersion, repositoryData, snapshotIds),
+                    ActionListener.wrap(updatedRepoData -> {
+                        logger.info("snapshots {} deleted", snapshotIds);
+                        removeSnapshotDeletionFromClusterState(deleteEntry, null, updatedRepoData);
+                    }, ex -> removeSnapshotDeletionFromClusterState(deleteEntry, ex, repositoryData))
+                );
+            }
         }
     }