Merge remote-tracking branch 'elastic/main' into all-in-one-query-change

Author: original-brownbear

docs/changelog/120807.yaml

@@ -0,0 +1,5 @@
+pr: 121240
+summary: Implement runtime skip_unavailable=true
+area: ES|QL
+type: enhancement
+issues: []

docs/changelog/121240.yaml

@@ -0,0 +1,5 @@
+pr: 121256
+summary: Run `TransportEnrichStatsAction` on local node
+area: Ingest Node
+type: enhancement
+issues: []

docs/changelog/121256.yaml

@@ -122,15 +122,9 @@ tests:
 - class: org.elasticsearch.test.rest.yaml.CcsCommonYamlTestSuiteIT
   method: test {p0=search.highlight/50_synthetic_source/text multi unified from vectors}
   issue: https://github.com/elastic/elasticsearch/issues/117815
-- class: org.elasticsearch.xpack.esql.plugin.ClusterRequestTests
-  method: testFallbackIndicesOptions
-  issue: https://github.com/elastic/elasticsearch/issues/117937
 - 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
@@ -215,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
@@ -437,6 +429,18 @@ 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
+- class: org.elasticsearch.xpack.esql.action.CrossClustersCancellationIT
+  method: testCancel
+  issue: https://github.com/elastic/elasticsearch/issues/121632
+- class: org.elasticsearch.xpack.esql.action.CrossClustersCancellationIT
+  method: testCancelSkipUnavailable
+  issue: https://github.com/elastic/elasticsearch/issues/121631
 
 # Examples:
 #

muted-tests.yml

@@ -18,7 +18,6 @@
 import java.util.List;
 import java.util.Map;
 
-import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 
 public class SourceModeRollingUpgradeIT extends AbstractRollingUpgradeTestCase {
@@ -83,20 +82,10 @@ public void testConfigureStoredSourceWhenIndexIsCreatedLegacy() throws IOExcepti
     private void assertDeprecationWarningForTemplate(String templateName) throws IOException {
         var request = new Request("GET", "/_migration/deprecations");
         var response = entityAsMap(client().performRequest(request));
-        if (response.containsKey("templates")) {
-            // Check the newer version of the deprecation API that contains the templates section
-            Map<?, ?> issuesByTemplate = (Map<?, ?>) response.get("templates");
-            assertThat(issuesByTemplate.containsKey(templateName), equalTo(true));
-            var templateIssues = (List<?>) issuesByTemplate.get(templateName);
-            assertThat(((Map<?, ?>) templateIssues.getFirst()).get("message"), equalTo(SourceFieldMapper.DEPRECATION_WARNING));
-        } else {
-            // Bwc version with 8.18 until https://github.com/elastic/elasticsearch/pull/120505/ gets backported, clean up after backport
-            var nodeSettings = (Map<?, ?>) ((List<?>) response.get("node_settings")).getFirst();
-            assertThat(nodeSettings.get("message"), equalTo(SourceFieldMapper.DEPRECATION_WARNING));
-            assertThat(
-                (String) nodeSettings.get("details"),
-                containsString(SourceFieldMapper.DEPRECATION_WARNING + " Affected component templates: [" + templateName + "]")
-            );
-        }
+        assertThat(response.containsKey("templates"), equalTo(true));
+        Map<?, ?> issuesByTemplate = (Map<?, ?>) response.get("templates");
+        assertThat(issuesByTemplate.containsKey(templateName), equalTo(true));
+        var templateIssues = (List<?>) issuesByTemplate.get(templateName);
+        assertThat(((Map<?, ?>) templateIssues.getFirst()).get("message"), equalTo(SourceFieldMapper.DEPRECATION_WARNING));
     }
 }

qa/rolling-upgrade/src/javaRestTest/java/org/elasticsearch/upgrades/SourceModeRollingUpgradeIT.java

@@ -20,7 +20,7 @@
     "params": {
       "master_timeout":{
         "type":"time",
-        "description":"Timeout for processing on master node"
+        "description":"Timeout for waiting for new cluster state in case it is blocked"
       }
     }
   }

rest-api-spec/src/main/resources/rest-api-spec/api/enrich.stats.json

@@ -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;
                     }

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

@@ -23,8 +23,6 @@
 import org.elasticsearch.cluster.ClusterStateTaskExecutor;
 import org.elasticsearch.cluster.ClusterStateTaskListener;
 import org.elasticsearch.cluster.NotMasterException;
-import org.elasticsearch.cluster.block.ClusterBlock;
-import org.elasticsearch.cluster.block.ClusterBlocks;
 import org.elasticsearch.cluster.coordination.FailedToCommitClusterStateException;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.Metadata;
@@ -72,7 +70,6 @@
 
 import static org.apache.logging.log4j.Level.DEBUG;
 import static org.apache.logging.log4j.Level.ERROR;
-import static org.elasticsearch.cluster.metadata.IndexMetadata.INDEX_REFRESH_BLOCK;
 import static org.elasticsearch.cluster.service.MasterService.isPublishFailureException;
 import static org.elasticsearch.core.Strings.format;
 
@@ -622,7 +619,6 @@ public ClusterState execute(BatchExecutionContext<StartedShardUpdateTask> batchE
             List<TaskContext<StartedShardUpdateTask>> tasksToBeApplied = new ArrayList<>();
             List<ShardRouting> shardRoutingsToBeApplied = new ArrayList<>(batchExecutionContext.taskContexts().size());
             Set<ShardRouting> seenShardRoutings = new HashSet<>(); // to prevent duplicates
-            Set<Index> indicesWithUnpromotableShardsStarted = null;
             final Map<Index, ClusterStateTimeRanges> updatedTimestampRanges = new HashMap<>();
             final ClusterState initialState = batchExecutionContext.initialState();
             for (var taskContext : batchExecutionContext.taskContexts()) {
@@ -741,14 +737,6 @@ public ClusterState execute(BatchExecutionContext<StartedShardUpdateTask> batchE
                                     new ClusterStateTimeRanges(newTimestampMillisRange, newEventIngestedMillisRange)
                                 );
                             }
-
-                            if (matched.isPromotableToPrimary() == false
-                                && initialState.blocks().hasIndexBlock(index.getName(), INDEX_REFRESH_BLOCK)) {
-                                if (indicesWithUnpromotableShardsStarted == null) {
-                                    indicesWithUnpromotableShardsStarted = new HashSet<>();
-                                }
-                                indicesWithUnpromotableShardsStarted.add(index);
-                            }
                         }
                     }
                 }
@@ -772,10 +760,7 @@ public ClusterState execute(BatchExecutionContext<StartedShardUpdateTask> batchE
                     maybeUpdatedState = ClusterState.builder(maybeUpdatedState).metadata(metadataBuilder).build();
                 }
 
-                maybeUpdatedState = maybeRemoveIndexRefreshBlocks(maybeUpdatedState, indicesWithUnpromotableShardsStarted);
-
                 assert assertStartedIndicesHaveCompleteTimestampRanges(maybeUpdatedState);
-                assert assertRefreshBlockIsNotPresentWhenTheIndexIsSearchable(maybeUpdatedState);
 
                 for (final var taskContext : tasksToBeApplied) {
                     final var task = taskContext.getTask();
@@ -791,36 +776,6 @@ public ClusterState execute(BatchExecutionContext<StartedShardUpdateTask> batchE
             return maybeUpdatedState;
         }
 
-        private static ClusterState maybeRemoveIndexRefreshBlocks(
-            ClusterState clusterState,
-            @Nullable Set<Index> indicesWithUnpromotableShardsStarted
-        ) {
-            // The provided cluster state must include the newly STARTED unpromotable shards
-            if (indicesWithUnpromotableShardsStarted == null) {
-                return clusterState;
-            }
-
-            ClusterBlocks.Builder clusterBlocksBuilder = null;
-            for (Index indexWithUnpromotableShardsStarted : indicesWithUnpromotableShardsStarted) {
-                String indexName = indexWithUnpromotableShardsStarted.getName();
-                assert clusterState.blocks().hasIndexBlock(indexName, INDEX_REFRESH_BLOCK) : indexWithUnpromotableShardsStarted;
-
-                var indexRoutingTable = clusterState.routingTable().index(indexWithUnpromotableShardsStarted);
-                if (indexRoutingTable.readyForSearch()) {
-                    if (clusterBlocksBuilder == null) {
-                        clusterBlocksBuilder = ClusterBlocks.builder(clusterState.blocks());
-                    }
-                    clusterBlocksBuilder.removeIndexBlock(indexName, INDEX_REFRESH_BLOCK);
-                }
-            }
-
-            if (clusterBlocksBuilder == null) {
-                return clusterState;
-            }
-
-            return ClusterState.builder(clusterState).blocks(clusterBlocksBuilder).build();
-        }
-
         private static boolean assertStartedIndicesHaveCompleteTimestampRanges(ClusterState clusterState) {
             for (Map.Entry<String, IndexRoutingTable> cursor : clusterState.getRoutingTable().getIndicesRouting().entrySet()) {
                 assert cursor.getValue().allPrimaryShardsActive() == false
@@ -844,16 +799,6 @@ private static boolean assertStartedIndicesHaveCompleteTimestampRanges(ClusterSt
             return true;
         }
 
-        private static boolean assertRefreshBlockIsNotPresentWhenTheIndexIsSearchable(ClusterState clusterState) {
-            for (Map.Entry<String, Set<ClusterBlock>> indexBlock : clusterState.blocks().indices().entrySet()) {
-                if (indexBlock.getValue().contains(INDEX_REFRESH_BLOCK)) {
-                    assert clusterState.routingTable().index(indexBlock.getKey()).readyForSearch() == false
-                        : "Index [" + indexBlock.getKey() + "] is searchable but has an INDEX_REFRESH_BLOCK";
-                }
-            }
-            return true;
-        }
-
         @Override
         public void clusterStatePublished(ClusterState newClusterState) {
             rerouteService.reroute(

server/src/main/java/org/elasticsearch/cluster/action/shard/ShardStateAction.java

@@ -3209,7 +3209,12 @@ void checkIndex() throws IOException {
             try {
                 doCheckIndex();
             } catch (IOException e) {
-                store.markStoreCorrupted(e);
+                if (ExceptionsHelper.unwrap(e, AlreadyClosedException.class) != null) {
+                    // Cache-based read operations on Lucene files can throw an AlreadyClosedException wrapped into an IOException in case
+                    // of evictions. We don't want to mark the store as corrupted for this.
+                } else {
+                    store.markStoreCorrupted(e);
+                }
                 throw e;
             } finally {
                 store.decRef();