Safer Wait for Snapshot Success in ClusterPrivilegeTests (elastic#40943)

* Safer Wait for Snapshot Success in ClusterPrivilegeTests

* The snapshot state returned by the API might become SUCCESS before it's fully removed from the cluster state.
  * We should fix this race in the transport API but it's not trivial and will be part of the incoming big round of refactoring the repository interaction, this added check fixes the test for now
* closes elastic#38030

x-pack/plugin/security/src/test/java/org/elasticsearch/integration/ClusterPrivilegeTests.java

@@ -6,12 +6,14 @@
 package org.elasticsearch.integration;
 
 import org.elasticsearch.action.admin.cluster.snapshots.status.SnapshotsStatusResponse;
+import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.cluster.SnapshotsInProgress;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.SecureString;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.xpack.core.security.authc.support.Hasher;
+import org.hamcrest.Matchers;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 
@@ -139,7 +141,6 @@ public void testThatClusterPrivilegesWorkAsExpectedViaHttp() throws Exception {
         assertAccessIsDenied("user_d", "PUT", "/_cluster/settings", "{ \"transient\" : { \"search.default_search_timeout\": \"1m\" } }");
     }
 
-    @AwaitsFix( bugUrl = "https://github.com/elastic/elasticsearch/issues/38030")
     public void testThatSnapshotAndRestore() throws Exception {
         String repoJson = Strings.toString(jsonBuilder().startObject().field("type", "fs").startObject("settings").field("location",
                 repositoryLocation.toString()).endObject().endObject());
@@ -203,6 +204,11 @@ private void waitForSnapshotToFinish(String repo, String snapshot) throws Except
         assertBusy(() -> {
             SnapshotsStatusResponse response = client().admin().cluster().prepareSnapshotStatus(repo).setSnapshots(snapshot).get();
             assertThat(response.getSnapshots().get(0).getState(), is(SnapshotsInProgress.State.SUCCESS));
+            // The status of the snapshot in the repository can become SUCCESS before it is fully finalized in the cluster state so wait for
+            // it to disappear from the cluster state as well
+            SnapshotsInProgress snapshotsInProgress =
+                client().admin().cluster().state(new ClusterStateRequest()).get().getState().custom(SnapshotsInProgress.TYPE);
+            assertThat(snapshotsInProgress.entries(), Matchers.empty());
         });
     }
 }