Merge branch 'main' into ES-10641_add_leak_detection_to_store

Author: nicktindall

muted-tests.yml

@@ -125,9 +125,6 @@ tests:
 - class: org.elasticsearch.xpack.ml.integration.RegressionIT
   method: testTwoJobsWithSameRandomizeSeedUseSameTrainingSet
   issue: https://github.com/elastic/elasticsearch/issues/117805
-- class: org.elasticsearch.xpack.esql.action.EsqlActionTaskIT
-  method: testCancelRequestWhenFailingFetchingPages
-  issue: https://github.com/elastic/elasticsearch/issues/118193
 - class: org.elasticsearch.packaging.test.ArchiveTests
   method: test44AutoConfigurationNotTriggeredOnNotWriteableConfDir
   issue: https://github.com/elastic/elasticsearch/issues/118208
@@ -212,8 +209,6 @@ tests:
 - class: org.elasticsearch.xpack.esql.action.CrossClusterAsyncEnrichStopIT
   method: testEnrichAfterStop
   issue: https://github.com/elastic/elasticsearch/issues/120757
-- class: org.elasticsearch.search.fieldcaps.FieldCapabilitiesIT
-  issue: https://github.com/elastic/elasticsearch/issues/120772
 - class: org.elasticsearch.xpack.test.rest.XPackRestIT
   method: test {p0=ml/3rd_party_deployment/Test start deployment fails while model download in progress}
   issue: https://github.com/elastic/elasticsearch/issues/120810
@@ -434,6 +429,12 @@ tests:
 - class: org.elasticsearch.xpack.esql.action.CrossClusterQueryUnavailableRemotesIT
   method: testRemoteOnlyCCSAgainstDisconnectedRemoteWithSkipUnavailableTrue
   issue: https://github.com/elastic/elasticsearch/issues/121578
+- class: org.elasticsearch.xpack.esql.action.CrossClustersCancellationIT
+  method: testTasks
+  issue: https://github.com/elastic/elasticsearch/issues/121626
+- class: org.elasticsearch.xpack.esql.action.CrossClustersCancellationIT
+  method: testCloseSkipUnavailable
+  issue: https://github.com/elastic/elasticsearch/issues/121627
 
 # Examples:
 #

server/src/internalClusterTest/java/org/elasticsearch/search/fieldcaps/FieldCapabilitiesIT.java

@@ -28,6 +28,7 @@
 import org.elasticsearch.client.Cancellable;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
+import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.routing.allocation.command.MoveAllocationCommand;
 import org.elasticsearch.cluster.routing.allocation.decider.EnableAllocationDecider;
@@ -73,6 +74,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CancellationException;
@@ -591,21 +593,31 @@ public void testNoActiveCopy() throws Exception {
 
     private void moveOrCloseShardsOnNodes(String nodeName) throws Exception {
         final IndicesService indicesService = internalCluster().getInstance(IndicesService.class, nodeName);
+        final ClusterState clusterState = clusterService().state();
         for (IndexService indexService : indicesService) {
             for (IndexShard indexShard : indexService) {
                 if (randomBoolean()) {
                     closeShardNoCheck(indexShard, randomBoolean());
                 } else if (randomBoolean()) {
                     final ShardId shardId = indexShard.shardId();
-
+                    final var assignedNodes = new HashSet<>();
+                    clusterState.routingTable().shardRoutingTable(shardId).allShards().forEach(shr -> {
+                        if (shr.currentNodeId() != null) {
+                            assignedNodes.add(shr.currentNodeId());
+                        }
+                        if (shr.relocatingNodeId() != null) {
+                            assignedNodes.add(shr.relocatingNodeId());
+                        }
+                    });
                     final var targetNodes = new ArrayList<String>();
                     for (final var targetIndicesService : internalCluster().getInstances(IndicesService.class)) {
                         final var targetNode = targetIndicesService.clusterService().localNode();
-                        if (targetNode.canContainData() && targetIndicesService.getShardOrNull(shardId) == null) {
+                        if (targetNode.canContainData()
+                            && targetIndicesService.getShardOrNull(shardId) == null
+                            && assignedNodes.contains(targetNode.getId()) == false) {
                             targetNodes.add(targetNode.getId());
                         }
                     }
-
                     if (targetNodes.isEmpty()) {
                         continue;
                     }

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/exchange/ExchangeServiceTests.java

@@ -436,8 +436,8 @@ public void testExchangeSourceContinueOnFailure() {
             exchangeSink
         );
         assertThat(actualSeqNos, equalTo(expectedSeqNos));
+        safeGet(sourceCompletionFuture);
         assertThat(completedSinks.get() + failedSinks.get(), equalTo(totalSinks.get()));
-        sourceCompletionFuture.actionGet();
         if (failedRequests.get() > 0) {
             assertThat(failedSinks.get(), greaterThan(0));
         } else {

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

@@ -462,7 +462,9 @@ protected void doRun() throws Exception {
             }
             Exception failure = expectThrows(Exception.class, () -> future.actionGet().close());
             EsqlTestUtils.assertEsqlFailure(failure);
-            assertThat(failure.getMessage(), containsString("failed to fetch pages"));
+            Throwable cause = ExceptionsHelper.unwrap(failure, IOException.class);
+            assertNotNull(cause);
+            assertThat(cause.getMessage(), containsString("failed to fetch pages"));
             // If we proceed without waiting for pages, we might cancel the main request before starting the data-node request.
             // As a result, the exchange sinks on data-nodes won't be removed until the inactive_timeout elapses, which is
             // longer than the assertBusy timeout.