Merge remote-tracking branch 'elastic/main' into batched-exec-short

Author: original-brownbear

muted-tests.yml

@@ -373,9 +373,6 @@ tests:
 - class: org.elasticsearch.search.CrossClusterSearchUnavailableClusterIT
   method: testSearchSkipUnavailable
   issue: https://github.com/elastic/elasticsearch/issues/121497
-- class: org.elasticsearch.xpack.esql.action.CrossClusterAsyncQueryStopIT
-  method: testStopQueryLocal
-  issue: https://github.com/elastic/elasticsearch/issues/121672
 - class: org.elasticsearch.xpack.esql.qa.multi_node.EsqlSpecIT
   issue: https://github.com/elastic/elasticsearch/issues/121411
 - class: org.elasticsearch.smoketest.DocsClientYamlTestSuiteIT
@@ -386,6 +383,26 @@ tests:
 - class: org.elasticsearch.xpack.searchablesnapshots.FrozenSearchableSnapshotsIntegTests
   method: testCreateAndRestorePartialSearchableSnapshot
   issue: https://github.com/elastic/elasticsearch/issues/121927
+- class: org.elasticsearch.analysis.common.CommonAnalysisClientYamlTestSuiteIT
+  method: test {yaml=analysis-common/40_token_filters/stemmer_override file access}
+  issue: https://github.com/elastic/elasticsearch/issues/121625
+- class: org.elasticsearch.test.rest.ClientYamlTestSuiteIT
+  method: test {yaml=update/100_synthetic_source/stored text}
+  issue: https://github.com/elastic/elasticsearch/issues/121964
+- class: org.elasticsearch.test.rest.ClientYamlTestSuiteIT
+  method: test {yaml=update/100_synthetic_source/keyword}
+  issue: https://github.com/elastic/elasticsearch/issues/121965
+- class: org.elasticsearch.xpack.esql.plugin.DataNodeRequestSenderTests
+  method: testDoNotRetryOnRequestLevelFailure
+  issue: https://github.com/elastic/elasticsearch/issues/121966
+- class: org.elasticsearch.xpack.searchablesnapshots.hdfs.SecureHdfsSearchableSnapshotsIT
+  issue: https://github.com/elastic/elasticsearch/issues/121967
+- class: org.elasticsearch.smoketest.SmokeTestMultiNodeClientYamlTestSuiteIT
+  method: test {yaml=update/100_synthetic_source/stored text}
+  issue: https://github.com/elastic/elasticsearch/issues/121991
+- class: org.elasticsearch.smoketest.SmokeTestMultiNodeClientYamlTestSuiteIT
+  method: test {yaml=update/100_synthetic_source/keyword}
+  issue: https://github.com/elastic/elasticsearch/issues/121992
 
 # Examples:
 #

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -174,13 +174,18 @@ static TransportVersion def(int id) {
     public static final TransportVersion ML_INFERENCE_IBM_WATSONX_RERANK_ADDED = def(8_840_0_00);
     public static final TransportVersion COHERE_BIT_EMBEDDING_TYPE_SUPPORT_ADDED_BACKPORT_8_X = def(8_840_0_01);
     public static final TransportVersion ELASTICSEARCH_9_0 = def(9_000_0_00);
+    public static final TransportVersion REMOVE_SNAPSHOT_FAILURES_90 = def(9_000_0_01);
+    public static final TransportVersion TRANSPORT_STATS_HANDLING_TIME_REQUIRED_90 = def(9_000_0_02);
+    public static final TransportVersion REMOVE_DESIRED_NODE_VERSION_90 = def(9_000_0_03);
+    public static final TransportVersion ESQL_DRIVER_TASK_DESCRIPTION_90 = def(9_000_0_04);
     public static final TransportVersion COHERE_BIT_EMBEDDING_TYPE_SUPPORT_ADDED = def(9_001_0_00);
     public static final TransportVersion REMOVE_SNAPSHOT_FAILURES = def(9_002_0_00);
     public static final TransportVersion TRANSPORT_STATS_HANDLING_TIME_REQUIRED = def(9_003_0_00);
     public static final TransportVersion REMOVE_DESIRED_NODE_VERSION = def(9_004_0_00);
     public static final TransportVersion ESQL_DRIVER_TASK_DESCRIPTION = def(9_005_0_00);
     public static final TransportVersion ESQL_RETRY_ON_SHARD_LEVEL_FAILURE = def(9_006_0_00);
-    public static final TransportVersion BATCHED_QUERY_PHASE_VERSION = def(9_007_0_00);
+    public static final TransportVersion ESQL_PROFILE_ASYNC_NANOS = def(9_007_00_0);
+    public static final TransportVersion BATCHED_QUERY_PHASE_VERSION = def(9_008_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/action/admin/cluster/desirednodes/TransportUpdateDesiredNodesAction.java

@@ -22,14 +22,12 @@
 import org.elasticsearch.cluster.desirednodes.VersionConflictException;
 import org.elasticsearch.cluster.metadata.DesiredNodes;
 import org.elasticsearch.cluster.metadata.DesiredNodesMetadata;
-import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.routing.RerouteService;
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.cluster.service.MasterServiceTaskQueue;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
-import org.elasticsearch.features.FeatureService;
 import org.elasticsearch.injection.guice.Inject;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -42,18 +40,15 @@
 public class TransportUpdateDesiredNodesAction extends TransportMasterNodeAction<UpdateDesiredNodesRequest, UpdateDesiredNodesResponse> {
     private static final Logger logger = LogManager.getLogger(TransportUpdateDesiredNodesAction.class);
 
-    private final FeatureService featureService;
     private final MasterServiceTaskQueue<UpdateDesiredNodesTask> taskQueue;
 
     @Inject
     public TransportUpdateDesiredNodesAction(
         TransportService transportService,
         ClusterService clusterService,
         RerouteService rerouteService,
-        FeatureService featureService,
         ThreadPool threadPool,
         ActionFilters actionFilters,
-        IndexNameExpressionResolver indexNameExpressionResolver,
         AllocationService allocationService
     ) {
         super(
@@ -67,7 +62,6 @@ public TransportUpdateDesiredNodesAction(
             UpdateDesiredNodesResponse::new,
             EsExecutors.DIRECT_EXECUTOR_SERVICE
         );
-        this.featureService = featureService;
         this.taskQueue = clusterService.createTaskQueue(
             "update-desired-nodes",
             Priority.URGENT,

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/get/GetSnapshotsResponse.java

@@ -49,7 +49,8 @@ public GetSnapshotsResponse(List<SnapshotInfo> snapshots, @Nullable String next,
 
     public GetSnapshotsResponse(StreamInput in) throws IOException {
         this.snapshots = in.readCollectionAsImmutableList(SnapshotInfo::readFrom);
-        if (in.getTransportVersion().before(TransportVersions.REMOVE_SNAPSHOT_FAILURES)) {
+        if (in.getTransportVersion().before(TransportVersions.REMOVE_SNAPSHOT_FAILURES)
+            && in.getTransportVersion().isPatchFrom(TransportVersions.REMOVE_SNAPSHOT_FAILURES_90) == false) {
             // Deprecated `failures` field
             in.readMap(StreamInput::readException);
         }
@@ -83,7 +84,8 @@ public int remaining() {
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         out.writeCollection(snapshots);
-        if (out.getTransportVersion().before(TransportVersions.REMOVE_SNAPSHOT_FAILURES)) {
+        if (out.getTransportVersion().before(TransportVersions.REMOVE_SNAPSHOT_FAILURES)
+            && out.getTransportVersion().isPatchFrom(TransportVersions.REMOVE_SNAPSHOT_FAILURES_90) == false) {
             // Deprecated `failures` field
             out.writeMap(Map.of(), StreamOutput::writeException);
         }

server/src/main/java/org/elasticsearch/action/search/AbstractSearchAsyncAction.java

@@ -246,32 +246,6 @@ void skipShard(SearchShardIterator iterator) {
         successfulShardExecution();
     }
 
-    private static boolean assertExecuteOnStartThread() {
-        // Ensure that the current code has the following stacktrace:
-        // AbstractSearchAsyncAction#start -> AbstractSearchAsyncAction#executePhase -> AbstractSearchAsyncAction#performPhaseOnShard
-        final StackTraceElement[] stackTraceElements = Thread.currentThread().getStackTrace();
-        assert stackTraceElements.length >= 6 : stackTraceElements;
-        int index = 0;
-        assert stackTraceElements[index++].getMethodName().equals("getStackTrace");
-        assert stackTraceElements[index++].getMethodName().equals("assertExecuteOnStartThread");
-        assert stackTraceElements[index++].getMethodName().equals("failOnUnavailable");
-        if (stackTraceElements[index].getMethodName().equals("performPhaseOnShard")) {
-            assert stackTraceElements[index].getClassName().endsWith("CanMatchPreFilterSearchPhase");
-            index++;
-        }
-        assert stackTraceElements[index].getClassName().endsWith("AbstractSearchAsyncAction");
-        assert stackTraceElements[index++].getMethodName().equals("run");
-
-        assert stackTraceElements[index].getClassName().endsWith("AbstractSearchAsyncAction");
-        assert stackTraceElements[index++].getMethodName().equals("executePhase");
-
-        assert stackTraceElements[index].getClassName().endsWith("AbstractSearchAsyncAction");
-        assert stackTraceElements[index++].getMethodName().equals("start");
-
-        assert stackTraceElements[index].getClassName().endsWith("AbstractSearchAsyncAction") == false;
-        return true;
-    }
-
     private void performPhaseOnShard(final int shardIndex, final SearchShardIterator shardIt, final SearchShardTarget shard) {
         if (throttleConcurrentRequests) {
             var pendingExecutions = pendingExecutionsPerNode.computeIfAbsent(
@@ -313,7 +287,6 @@ public void onFailure(Exception e) {
     }
 
     private void failOnUnavailable(int shardIndex, SearchShardIterator shardIt) {
-        assert assertExecuteOnStartThread();
         SearchShardTarget unassignedShard = new SearchShardTarget(null, shardIt.shardId(), shardIt.getClusterAlias());
         onShardFailure(shardIndex, unassignedShard, shardIt, new NoShardAvailableActionException(shardIt.shardId()));
     }

server/src/main/java/org/elasticsearch/cluster/coordination/NodeJoinExecutor.java

@@ -371,7 +371,7 @@ private static void blockForbiddenVersions(TransportVersion joiningTransportVers
      * that are also present across the whole cluster as a result.
      */
     private Set<String> calculateEffectiveClusterFeatures(DiscoveryNodes nodes, Map<String, Set<String>> nodeFeatures) {
-        if (featureService.featuresCanBeAssumedForNodes(nodes)) {
+        if (FeatureService.featuresCanBeAssumedForNodes(nodes)) {
             Set<String> assumedFeatures = featureService.getNodeFeatures()
                 .values()
                 .stream()
@@ -382,7 +382,7 @@ private Set<String> calculateEffectiveClusterFeatures(DiscoveryNodes nodes, Map<
             // add all assumed features to the featureset of all nodes of the next major version
             nodeFeatures = new HashMap<>(nodeFeatures);
             for (var node : nodes.getNodes().entrySet()) {
-                if (featureService.featuresCanBeAssumedForNode(node.getValue())) {
+                if (FeatureService.featuresCanBeAssumedForNode(node.getValue())) {
                     assert nodeFeatures.containsKey(node.getKey()) : "Node " + node.getKey() + " does not have any features";
                     nodeFeatures.computeIfPresent(node.getKey(), (k, v) -> {
                         var newFeatures = new HashSet<>(v);
@@ -525,7 +525,7 @@ private Set<String> enforceNodeFeatureBarrier(DiscoveryNode node, Set<String> ef
             return newNodeFeatures;
         }
 
-        if (featureService.featuresCanBeAssumedForNode(node)) {
+        if (FeatureService.featuresCanBeAssumedForNode(node)) {
             // it might still be ok for this node to join if this node can have assumed features,
             // and all the missing features are assumed
             // we can get the NodeFeature object direct from this node's registered features

server/src/main/java/org/elasticsearch/cluster/metadata/DesiredNode.java

@@ -161,7 +161,8 @@ public static DesiredNode readFrom(StreamInput in) throws IOException {
         }
         final var memory = ByteSizeValue.readFrom(in);
         final var storage = ByteSizeValue.readFrom(in);
-        if (in.getTransportVersion().before(TransportVersions.REMOVE_DESIRED_NODE_VERSION)) {
+        if (in.getTransportVersion().before(TransportVersions.REMOVE_DESIRED_NODE_VERSION)
+            && in.getTransportVersion().isPatchFrom(TransportVersions.REMOVE_DESIRED_NODE_VERSION_90) == false) {
             in.readOptionalString();
         }
         return new DesiredNode(settings, processors, processorsRange, memory, storage);
@@ -180,7 +181,8 @@ public void writeTo(StreamOutput out) throws IOException {
         }
         memory.writeTo(out);
         storage.writeTo(out);
-        if (out.getTransportVersion().before(TransportVersions.REMOVE_DESIRED_NODE_VERSION)) {
+        if (out.getTransportVersion().before(TransportVersions.REMOVE_DESIRED_NODE_VERSION)
+            && out.getTransportVersion().isPatchFrom(TransportVersions.REMOVE_DESIRED_NODE_VERSION_90) == false) {
             out.writeOptionalString(null);
         }
     }

server/src/main/java/org/elasticsearch/features/FeatureService.java

@@ -53,14 +53,14 @@ public Map<String, NodeFeature> getNodeFeatures() {
     /**
      * Returns {@code true} if {@code node} can have assumed features.
      */
-    public boolean featuresCanBeAssumedForNode(DiscoveryNode node) {
+    public static boolean featuresCanBeAssumedForNode(DiscoveryNode node) {
         return ClusterFeatures.featuresCanBeAssumedForNode(node);
     }
 
     /**
     * Returns {@code true} if one or more nodes in {@code nodes} can have assumed features.
     */
-    public boolean featuresCanBeAssumedForNodes(DiscoveryNodes nodes) {
+    public static boolean featuresCanBeAssumedForNodes(DiscoveryNodes nodes) {
         return ClusterFeatures.featuresCanBeAssumedForNodes(nodes);
     }
 

server/src/main/java/org/elasticsearch/health/metadata/HealthMetadataService.java

@@ -26,7 +26,6 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Tuple;
-import org.elasticsearch.features.FeatureService;
 import org.elasticsearch.gateway.GatewayService;
 
 import java.util.List;
@@ -50,7 +49,6 @@ public class HealthMetadataService {
     private static final Logger logger = LogManager.getLogger(HealthMetadataService.class);
 
     private final ClusterService clusterService;
-    private final FeatureService featureService;
     private final ClusterStateListener clusterStateListener;
     private final MasterServiceTaskQueue<UpsertHealthMetadataTask> taskQueue;
     private volatile boolean enabled;
@@ -64,17 +62,16 @@ public class HealthMetadataService {
     // ClusterState to maintain an up-to-date version of it across the cluster.
     private volatile HealthMetadata localHealthMetadata;
 
-    private HealthMetadataService(ClusterService clusterService, FeatureService featureService, Settings settings) {
+    private HealthMetadataService(ClusterService clusterService, Settings settings) {
         this.clusterService = clusterService;
-        this.featureService = featureService;
         this.clusterStateListener = this::updateOnClusterStateChange;
         this.enabled = ENABLED_SETTING.get(settings);
         this.localHealthMetadata = initialHealthMetadata(settings);
         this.taskQueue = clusterService.createTaskQueue("health metadata service", Priority.NORMAL, new Executor());
     }
 
-    public static HealthMetadataService create(ClusterService clusterService, FeatureService featureService, Settings settings) {
-        HealthMetadataService healthMetadataService = new HealthMetadataService(clusterService, featureService, settings);
+    public static HealthMetadataService create(ClusterService clusterService, Settings settings) {
+        HealthMetadataService healthMetadataService = new HealthMetadataService(clusterService, settings);
         healthMetadataService.registerListeners();
         return healthMetadataService;
     }

server/src/main/java/org/elasticsearch/health/node/DiskHealthIndicatorService.java

@@ -18,7 +18,6 @@
 import org.elasticsearch.cluster.routing.RoutingNodes;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.util.set.Sets;
-import org.elasticsearch.features.FeatureService;
 import org.elasticsearch.health.Diagnosis;
 import org.elasticsearch.health.HealthIndicatorDetails;
 import org.elasticsearch.health.HealthIndicatorImpact;
@@ -73,11 +72,9 @@ public class DiskHealthIndicatorService implements HealthIndicatorService {
     private static final String IMPACT_CLUSTER_FUNCTIONALITY_UNAVAILABLE_ID = "cluster_functionality_unavailable";
 
     private final ClusterService clusterService;
-    private final FeatureService featureService;
 
-    public DiskHealthIndicatorService(ClusterService clusterService, FeatureService featureService) {
+    public DiskHealthIndicatorService(ClusterService clusterService) {
         this.clusterService = clusterService;
-        this.featureService = featureService;
     }
 
     @Override