[Backport] Modify create snapshot v2 response when wait_for_completion is false (#15615)

Signed-off-by: Anshu Agarwal <anshukag@amazon.com>

Author: anshu1106

server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteRestoreSnapshotIT.java

@@ -845,18 +845,13 @@ public void testCreateSnapshotV2() throws Exception {
 
         String snapshotName2 = "test-create-snapshot2";
 
-        // verify even if waitForCompletion is not true, the request executes in a sync manner
-        CreateSnapshotResponse createSnapshotResponse2 = client().admin()
+        // verify response status if waitForCompletion is not true
+        RestStatus createSnapshotResponseStatus = client().admin()
             .cluster()
             .prepareCreateSnapshot(snapshotRepoName, snapshotName2)
-            .get();
-        snapshotInfo = createSnapshotResponse2.getSnapshotInfo();
-        assertThat(snapshotInfo.state(), equalTo(SnapshotState.SUCCESS));
-        assertThat(snapshotInfo.successfulShards(), greaterThan(0));
-        assertThat(snapshotInfo.successfulShards(), equalTo(snapshotInfo.totalShards()));
-        assertThat(snapshotInfo.snapshotId().getName(), equalTo(snapshotName2));
-        assertThat(snapshotInfo.getPinnedTimestamp(), greaterThan(0L));
-
+            .get()
+            .status();
+        assertEquals(RestStatus.ACCEPTED, createSnapshotResponseStatus);
     }
 
     public void testMixedSnapshotCreationWithV2RepositorySetting() throws Exception {
@@ -932,6 +927,7 @@ public void testMixedSnapshotCreationWithV2RepositorySetting() throws Exception
         CreateSnapshotResponse createSnapshotResponse2 = client().admin()
             .cluster()
             .prepareCreateSnapshot(snapshotRepoName, snapshotName2)
+            .setWaitForCompletion(true)
             .get();
         snapshotInfo = createSnapshotResponse2.getSnapshotInfo();
         assertThat(snapshotInfo.state(), equalTo(SnapshotState.SUCCESS));
@@ -993,6 +989,7 @@ public void testConcurrentSnapshotV2CreateOperation() throws InterruptedExceptio
                     CreateSnapshotResponse createSnapshotResponse2 = client().admin()
                         .cluster()
                         .prepareCreateSnapshot(snapshotRepoName, snapshotName)
+                        .setWaitForCompletion(true)
                         .get();
                     SnapshotInfo snapshotInfo = createSnapshotResponse2.getSnapshotInfo();
                     assertThat(snapshotInfo.state(), equalTo(SnapshotState.SUCCESS));
@@ -1068,6 +1065,7 @@ public void testCreateSnapshotV2WithRedIndex() throws Exception {
         CreateSnapshotResponse createSnapshotResponse2 = client().admin()
             .cluster()
             .prepareCreateSnapshot(snapshotRepoName, snapshotName1)
+            .setWaitForCompletion(true)
             .get();
         SnapshotInfo snapshotInfo = createSnapshotResponse2.getSnapshotInfo();
         assertThat(snapshotInfo.state(), equalTo(SnapshotState.SUCCESS));
@@ -1136,6 +1134,7 @@ public void testCreateSnapshotV2WithIndexingLoad() throws Exception {
         CreateSnapshotResponse createSnapshotResponse2 = client().admin()
             .cluster()
             .prepareCreateSnapshot(snapshotRepoName, snapshotName1)
+            .setWaitForCompletion(true)
             .get();
 
         SnapshotInfo snapshotInfo = createSnapshotResponse2.getSnapshotInfo();
@@ -1256,6 +1255,7 @@ public void testClusterManagerFailoverDuringSnapshotCreation() throws Exception
                 CreateSnapshotResponse createSnapshotResponse = client().admin()
                     .cluster()
                     .prepareCreateSnapshot(snapshotRepoName, snapshotName1)
+                    .setWaitForCompletion(true)
                     .get();
                 snapshotInfo[0] = createSnapshotResponse.getSnapshotInfo();
 

server/src/main/java/org/opensearch/action/admin/cluster/snapshots/create/TransportCreateSnapshotAction.java

@@ -111,18 +111,22 @@ protected void clusterManagerOperation(
         ClusterState state,
         final ActionListener<CreateSnapshotResponse> listener
     ) {
+
         if (state.nodes().getMinNodeVersion().before(SnapshotsService.NO_REPO_INITIALIZE_VERSION)) {
             if (request.waitForCompletion()) {
                 snapshotsService.executeSnapshotLegacy(request, ActionListener.map(listener, CreateSnapshotResponse::new));
             } else {
                 snapshotsService.createSnapshotLegacy(request, ActionListener.map(listener, snapshot -> new CreateSnapshotResponse()));
             }
+
         } else {
             Repository repository = repositoriesService.repository(request.repository());
             boolean isSnapshotV2 = SHALLOW_SNAPSHOT_V2.get(repository.getMetadata().settings());
 
-            if (request.waitForCompletion() || isSnapshotV2) {
+            if (request.waitForCompletion()) {
                 snapshotsService.executeSnapshot(request, ActionListener.map(listener, CreateSnapshotResponse::new));
+            } else if (isSnapshotV2) {
+                snapshotsService.executeSnapshot(request, ActionListener.map(listener, snapshot -> new CreateSnapshotResponse()));
             } else {
                 snapshotsService.createSnapshot(request, ActionListener.map(listener, snapshot -> new CreateSnapshotResponse()));
             }