Disallow snapshot deletion while a v2 snapshot is in progress (#16430)

---------

Signed-off-by: Gaurav Bafna <gbbafna@amazon.com>
(cherry picked from commit bb1359f)
Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

server/src/internalClusterTest/java/org/opensearch/snapshots/ConcurrentSnapshotsV2IT.java

@@ -529,14 +529,13 @@ public void testDeleteWhileV2CreateOngoing() throws Exception {
         awaitNumberOfSnapshotsInProgress(1);
 
         ActionFuture<AcknowledgedResponse> a = startDeleteSnapshot(repoName, "snapshot-v1");
+        expectThrows(ConcurrentSnapshotExecutionException.class, a::actionGet);
 
         unblockNode(repoName, clusterManagerName);
         CreateSnapshotResponse csr = snapshotFuture.actionGet();
         assertTrue(csr.getSnapshotInfo().getPinnedTimestamp() != 0);
-        assertTrue(a.actionGet().isAcknowledged());
         List<SnapshotInfo> snapInfo = client().admin().cluster().prepareGetSnapshots(repoName).get().getSnapshots();
-        assertEquals(1, snapInfo.size());
-        assertThat(snapInfo, contains(csr.getSnapshotInfo()));
+        assertEquals(2, snapInfo.size());
     }
 
     @AwaitsFix(bugUrl = "https://github.com/opensearch-project/OpenSearch/issues/16205")

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

@@ -802,10 +802,8 @@ public void onResponse(RepositoryData repositoryData) {
                             }
                             cleanOrphanTimestamp(repositoryName, repositoryData);
                             logger.info("created snapshot-v2 [{}] in repository [{}]", repositoryName, snapshotName);
+                            leaveRepoLoop(repositoryName);
                             listener.onResponse(snapshotInfo);
-                            // For snapshot-v2, we don't allow concurrent snapshots . But meanwhile non-v2 snapshot operations
-                            // can get queued . This is triggering them.
-                            runNextQueuedOperation(repositoryData, repositoryName, true);
                         }
 
                         @Override
@@ -1204,10 +1202,8 @@ public void onResponse(RepositoryData repositoryData) {
                                         return;
                                     }
                                     logger.info("snapshot-v2 clone [{}] completed successfully", snapshot);
+                                    leaveRepoLoop(repositoryName);
                                     listener.onResponse(null);
-                                    // For snapshot-v2, we don't allow concurrent snapshots . But meanwhile non-v2 snapshot operations
-                                    // can get queued . This is triggering them.
-                                    runNextQueuedOperation(repositoryData, repositoryName, true);
                                 }
 
                                 @Override
@@ -3032,6 +3028,19 @@ public ClusterState execute(ClusterState currentState) throws Exception {
                 }
                 final SnapshotsInProgress snapshots = currentState.custom(SnapshotsInProgress.TYPE, SnapshotsInProgress.EMPTY);
                 final List<SnapshotsInProgress.Entry> snapshotEntries = findInProgressSnapshots(snapshots, snapshotNames, repoName);
+                boolean isSnapshotV2 = SHALLOW_SNAPSHOT_V2.get(repository.getMetadata().settings());
+                boolean remoteStoreIndexShallowCopy = remoteStoreShallowCopyEnabled(repository);
+                List<SnapshotsInProgress.Entry> entriesForThisRepo = snapshots.entries()
+                    .stream()
+                    .filter(entry -> Objects.equals(entry.repository(), repoName))
+                    .collect(Collectors.toList());
+                if (isSnapshotV2 && remoteStoreIndexShallowCopy && entriesForThisRepo.isEmpty() == false) {
+                    throw new ConcurrentSnapshotExecutionException(
+                        repoName,
+                        String.join(",", snapshotNames),
+                        "cannot delete snapshots in v2 repo while a snapshot is in progress"
+                    );
+                }
                 final List<SnapshotId> snapshotIds = matchingSnapshotIds(
                     snapshotEntries.stream().map(e -> e.snapshot().getSnapshotId()).collect(Collectors.toList()),
                     repositoryData,