Fix RemoteClusterSecurityRestIT by waiting for tasks to finish (#128777)

Author: benchaplin

muted-tests.yml

@@ -417,9 +417,6 @@ tests:
 - class: org.elasticsearch.packaging.test.DockerTests
   method: test025SyncPluginsUsingProxy
   issue: https://github.com/elastic/elasticsearch/issues/127138
-- class: org.elasticsearch.xpack.remotecluster.RemoteClusterSecurityRestIT
-  method: testTaskCancellation
-  issue: https://github.com/elastic/elasticsearch/issues/128009
 - class: org.elasticsearch.xpack.esql.action.CrossClusterQueryWithPartialResultsIT
   method: testOneRemoteClusterPartial
   issue: https://github.com/elastic/elasticsearch/issues/124055

x-pack/plugin/security/qa/multi-cluster/src/javaRestTest/java/org/elasticsearch/xpack/remotecluster/RemoteClusterSecurityRestIT.java

@@ -155,6 +155,7 @@ public void testTaskCancellation() throws Exception {
         final String indexName = "index_fulfilling";
         final String roleName = "taskCancellationRoleName";
         final String userName = "taskCancellationUsername";
+        String asyncSearchOpaqueId = "async-search-opaque-id-" + randomUUID();
         try {
             // create some index on the fulfilling cluster, to be searched from the querying cluster
             {
@@ -206,13 +207,12 @@ public void testTaskCancellation() throws Exception {
                         {
                           "name": "*:*",
                           "error_type": "exception",
-                          "stall_time_seconds": 60
+                          "stall_time_seconds": 10
                         }
                       ]
                     }
                   }
                 }""");
-            String asyncSearchOpaqueId = "async-search-opaque-id-" + randomUUID();
             submitAsyncSearchRequest.setOptions(
                 RequestOptions.DEFAULT.toBuilder()
                     .addHeader("Authorization", headerFromRandomAuthMethod(userName, PASS))
@@ -308,6 +308,25 @@ public void testTaskCancellation() throws Exception {
             assertOK(adminClient().performRequest(new Request("DELETE", "/_security/user/" + userName)));
             assertOK(adminClient().performRequest(new Request("DELETE", "/_security/role/" + roleName)));
             assertOK(performRequestAgainstFulfillingCluster(new Request("DELETE", indexName)));
+            // wait for search related tasks to finish on the query cluster
+            assertBusy(() -> {
+                try {
+                    Response queryingClusterTasks = adminClient().performRequest(new Request("GET", "/_tasks"));
+                    assertOK(queryingClusterTasks);
+                    Map<String, Object> responseMap = XContentHelper.convertToMap(
+                        JsonXContent.jsonXContent,
+                        EntityUtils.toString(queryingClusterTasks.getEntity()),
+                        false
+                    );
+                    selectTasksWithOpaqueId(responseMap, asyncSearchOpaqueId, task -> {
+                        if (task.get("action") instanceof String action && action.contains("indices:data/read/search")) {
+                            fail("there are still search tasks running");
+                        }
+                    });
+                } catch (ResponseException e) {
+                    fail(e.getMessage());
+                }
+            });
         }
     }