Stop reusing cluster for CrossClusterAsyncQueryStopIT and unmute (#136429) (#136439)

* Stop reusing cluster & unmute

(cherry picked from commit 620c5a3)

# Conflicts:
#	muted-tests.yml

Author: smalyshev

muted-tests.yml

@@ -429,9 +429,6 @@ tests:
 - class: org.elasticsearch.xpack.esql.qa.multi_node.EsqlClientYamlIT
   method: test {p0=esql/60_usage/*}
   issue: https://github.com/elastic/elasticsearch/issues/133461
-- class: org.elasticsearch.xpack.esql.action.CrossClusterAsyncQueryStopIT
-  method: testStopQueryLocal
-  issue: https://github.com/elastic/elasticsearch/issues/133481
 - class: org.elasticsearch.xpack.esql.inference.rerank.RerankOperatorTests
   method: testSimpleCircuitBreaking
   issue: https://github.com/elastic/elasticsearch/issues/133619

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

@@ -8,6 +8,7 @@
 package org.elasticsearch.xpack.esql.action;
 
 import org.elasticsearch.Build;
+import org.elasticsearch.ElasticsearchTimeoutException;
 import org.elasticsearch.action.ActionFuture;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.core.Tuple;
@@ -16,6 +17,8 @@
 import org.elasticsearch.tasks.TaskInfo;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.core.async.AsyncStopRequest;
+import org.elasticsearch.xpack.core.async.DeleteAsyncResultRequest;
+import org.elasticsearch.xpack.core.async.TransportDeleteAsyncResultAction;
 import org.elasticsearch.xpack.esql.plugin.EsqlPlugin;
 import org.elasticsearch.xpack.esql.plugin.QueryPragmas;
 
@@ -28,7 +31,6 @@
 
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.elasticsearch.xpack.esql.action.AbstractEsqlIntegTestCase.randomIncludeCCSMetadata;
-import static org.elasticsearch.xpack.esql.action.EsqlAsyncTestUtils.deleteAsyncId;
 import static org.elasticsearch.xpack.esql.action.EsqlAsyncTestUtils.getAsyncResponse;
 import static org.elasticsearch.xpack.esql.action.EsqlAsyncTestUtils.startAsyncQuery;
 import static org.elasticsearch.xpack.esql.action.EsqlAsyncTestUtils.startAsyncQueryWithPragmas;
@@ -43,6 +45,11 @@ public class CrossClusterAsyncQueryStopIT extends AbstractCrossClusterTestCase {
 
     private static final Logger LOGGER = LogManager.getLogger(CrossClusterAsyncQueryStopIT.class);
 
+    @Override
+    protected boolean reuseClusters() {
+        return false;
+    }
+
     public void testStopQuery() throws Exception {
         assumeTrue("Pragma does not work in release builds", Build.current().isSnapshot());
         Map<String, Object> testClusterInfo = setupClusters(3);
@@ -125,7 +132,7 @@ public void testStopQuery() throws Exception {
         } finally {
             // Ensure proper cleanup if the test fails
             CountingPauseFieldPlugin.allowEmitting.countDown();
-            assertAcked(deleteAsyncId(client(), asyncExecutionId));
+            deleteAsyncId(client(), asyncExecutionId);
         }
     }
 
@@ -222,7 +229,7 @@ public void testStopQueryLocal() throws Exception {
             }
         } finally {
             SimplePauseFieldPlugin.allowEmitting.countDown();
-            assertAcked(deleteAsyncId(client, asyncExecutionId));
+            deleteAsyncId(client, asyncExecutionId);
         }
     }
 
@@ -265,7 +272,7 @@ public void testStopQueryLocalNoRemotes() throws Exception {
             }
         } finally {
             SimplePauseFieldPlugin.allowEmitting.countDown();
-            assertAcked(deleteAsyncId(client(), asyncExecutionId));
+            deleteAsyncId(client(), asyncExecutionId);
         }
     }
 
@@ -351,7 +358,16 @@ public void testStopQueryInlineStats() throws Exception {
         } finally {
             // Ensure proper cleanup if the test fails
             CountingPauseFieldPlugin.allowEmitting.countDown();
-            assertAcked(deleteAsyncId(client(), asyncExecutionId));
+            deleteAsyncId(client(), asyncExecutionId);
+        }
+    }
+
+    public void deleteAsyncId(Client client, String id) {
+        try {
+            DeleteAsyncResultRequest request = new DeleteAsyncResultRequest(id);
+            assertAcked(client.execute(TransportDeleteAsyncResultAction.TYPE, request).actionGet(30, TimeUnit.SECONDS));
+        } catch (ElasticsearchTimeoutException e) {
+            LOGGER.warn("timeout waiting for DELETE response: {}: {}", id, e);
         }
     }