[8.18] Improve CrossClusterAsyncEnrichStopIT test (elastic#122432) (elastic#122441) (elastic#122453)

* [8.x] Improve CrossClusterAsyncEnrichStopIT test (elastic#122432) (elastic#122441)

* Improve CrossClusterAsyncEnrichStopIT test (elastic#122432)

(cherry picked from commit 88550f6)

* fix backport

(cherry picked from commit 2f6de7d)

# Conflicts:
#	muted-tests.yml
#	x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/AbstractCrossClusterTestCase.java
#	x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterAsyncQueryStopIT.java

* fix backport

Author: smalyshev

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterAsyncEnrichStopIT.java

@@ -8,9 +8,12 @@
 package org.elasticsearch.xpack.esql.action;
 
 import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest;
+import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.xcontent.XContentHelper;
+import org.elasticsearch.compute.operator.DriverTaskRunner;
 import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.tasks.TaskInfo;
 import org.elasticsearch.xcontent.json.JsonXContent;
 import org.elasticsearch.xpack.core.async.AsyncStopRequest;
 import org.elasticsearch.xpack.esql.plan.logical.Enrich;
@@ -31,7 +34,9 @@
 import static org.elasticsearch.xpack.esql.action.EsqlAsyncTestUtils.deleteAsyncId;
 import static org.elasticsearch.xpack.esql.action.EsqlAsyncTestUtils.startAsyncQuery;
 import static org.elasticsearch.xpack.esql.action.EsqlAsyncTestUtils.waitForCluster;
+import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.not;
 
 // This tests if enrich after stop works correctly
 public class CrossClusterAsyncEnrichStopIT extends AbstractEnrichBasedCrossClusterTestCase {
@@ -87,10 +92,23 @@ public void testEnrichAfterStop() throws Exception {
         // wait until c1 is done
         waitForCluster(client(), "c1", asyncExecutionId);
         waitForCluster(client(), LOCAL_CLUSTER, asyncExecutionId);
+        // wait until remote reduce task starts on c2
+        assertBusy(() -> {
+            List<TaskInfo> tasks = getDriverTasks(client(REMOTE_CLUSTER_2));
+            List<TaskInfo> reduceTasks = tasks.stream().filter(t -> t.description().contains("_LuceneSourceOperator") == false).toList();
+            assertThat(reduceTasks, not(empty()));
+        });
 
         // Run the stop request
         var stopRequest = new AsyncStopRequest(asyncExecutionId);
         var stopAction = client().execute(EsqlAsyncStopAction.INSTANCE, stopRequest);
+        // wait until remote reduce tasks are gone
+        assertBusy(() -> {
+            List<TaskInfo> tasks = getDriverTasks(client(REMOTE_CLUSTER_2));
+            List<TaskInfo> reduceTasks = tasks.stream().filter(t -> t.description().contains("_LuceneSourceOperator") == false).toList();
+            assertThat(reduceTasks, empty());
+        });
+
         // Allow the processing to proceed
         SimplePauseFieldPlugin.allowEmitting.countDown();
 
@@ -153,4 +171,9 @@ record Event(long timestamp, String user, String host) {}
         }
         client.admin().indices().prepareRefresh("events").get();
     }
+
+    static List<TaskInfo> getDriverTasks(Client client) {
+        return client.admin().cluster().prepareListTasks().setActions(DriverTaskRunner.ACTION_NAME).setDetailed(true).get().getTasks();
+    }
+
 }