Retry in SnapshotIT Snapshot Abort (#54195) (#54251)

Retry here to work around the possible race between snapshot finalization
and deletion.

Closes #53509

client/rest-high-level/src/test/java/org/elasticsearch/client/SnapshotIT.java

@@ -131,7 +131,7 @@ public void testVerifyRepository() throws IOException {
         assertThat(response.getNodes().size(), equalTo(1));
     }
 
-    public void testCreateSnapshot() throws IOException {
+    public void testCreateSnapshot() throws Exception {
         String repository = "test_repository";
         assertTrue(createTestRepository(repository, FsRepository.TYPE, "{\"location\": \".\"}").isAcknowledged());
 
@@ -145,12 +145,23 @@ public void testCreateSnapshot() throws IOException {
         CreateSnapshotResponse response = createTestSnapshot(request);
         assertEquals(waitForCompletion ? RestStatus.OK : RestStatus.ACCEPTED, response.status());
         if (waitForCompletion == false) {
-            // If we don't wait for the snapshot to complete we have to cancel it to not leak the snapshot task
-            AcknowledgedResponse deleteResponse = execute(
-                new DeleteSnapshotRequest(repository, snapshot),
-                highLevelClient().snapshot()::delete, highLevelClient().snapshot()::deleteAsync
-            );
-            assertTrue(deleteResponse.isAcknowledged());
+            // busy assert on the delete because a known race condition could cause the delete request to not see
+            // the snapshot if delete and snapshot finalization happen at the same time
+            // See https://github.com/elastic/elasticsearch/issues/53509#issuecomment-603899620 for details
+            // TODO: Remove busy assert in 7.x+ once this race is fixed
+            assertBusy(() -> {
+                // If we don't wait for the snapshot to complete we have to cancel it to not leak the snapshot task
+                AcknowledgedResponse deleteResponse;
+                try {
+                    deleteResponse = execute(
+                        new DeleteSnapshotRequest(repository, snapshot),
+                        highLevelClient().snapshot()::delete, highLevelClient().snapshot()::deleteAsync
+                    );
+                } catch (Exception e) {
+                    throw new AssertionError(e);
+                }
+                assertTrue(deleteResponse.isAcknowledged());
+            });
         }
     }