Fix intermittent failures in CrossClusterAsyncSearchIT (#142781) (#142866)

(cherry picked from commit cf1cf4c)

# Conflicts:
#	muted-tests.yml

Author: smalyshev

muted-tests.yml

@@ -272,9 +272,6 @@ tests:
 - class: org.elasticsearch.xpack.security.authc.AuthenticationServiceTests
   method: testInvalidToken
   issue: https://github.com/elastic/elasticsearch/issues/133328
-- class: org.elasticsearch.xpack.search.CrossClusterAsyncSearchIT
-  method: testCancelViaExpirationOnRemoteResultsWithMinimizeRoundtrips
-  issue: https://github.com/elastic/elasticsearch/issues/127302
 - class: org.elasticsearch.xpack.esql.qa.multi_node.EsqlClientYamlIT
   method: test {p0=esql/60_usage/*}
   issue: https://github.com/elastic/elasticsearch/issues/133461

x-pack/plugin/async-search/src/internalClusterTest/java/org/elasticsearch/xpack/search/CrossClusterAsyncSearchIT.java

@@ -10,6 +10,7 @@
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.ResourceNotFoundException;
 import org.elasticsearch.action.ActionFuture;
+import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
 import org.elasticsearch.action.admin.cluster.node.tasks.cancel.CancelTasksRequest;
 import org.elasticsearch.action.admin.cluster.node.tasks.get.GetTaskRequest;
 import org.elasticsearch.action.admin.cluster.node.tasks.get.GetTaskResponse;
@@ -1946,14 +1947,19 @@ private Map<String, Object> setupTwoClusters() {
                 .setSettings(Settings.builder().put(remoteSettings.build()))
                 .setMapping("@timestamp", "type=date", "f", "type=text")
         );
-        assertFalse(
-            client(REMOTE_CLUSTER).admin()
-                .cluster()
-                .prepareHealth(TEST_REQUEST_TIMEOUT, remoteIndex)
-                .setWaitForYellowStatus()
-                .setTimeout(TimeValue.timeValueSeconds(10))
-                .get()
-                .isTimedOut()
+        // Wait for yellow and no initializing shards so all assigned shards are STARTED before tests run searches.
+        ClusterHealthResponse healthResponse = client(REMOTE_CLUSTER).admin()
+            .cluster()
+            .prepareHealth(TEST_REQUEST_TIMEOUT, remoteIndex)
+            .setWaitForYellowStatus()
+            .setWaitForNoInitializingShards(true)
+            .setTimeout(TimeValue.timeValueSeconds(10))
+            .get();
+        assertFalse("remote index health check timed out", healthResponse.isTimedOut());
+        assertEquals(
+            "remote index should have no initializing shards, had: " + healthResponse.getInitializingShards(),
+            0,
+            healthResponse.getInitializingShards()
         );
         indexDocs(client(REMOTE_CLUSTER), remoteIndex);