Merge branch 'main' into lucene_snapshot

Author: javanna

docs/changelog/116346.yaml

@@ -0,0 +1,5 @@
+pr: 116346
+summary: "[ESQL] Fix Binary Comparisons on Date Nanos"
+area: ES|QL
+type: bug
+issues: []

docs/reference/connector/docs/_connectors-create-native.asciidoc

@@ -22,7 +22,7 @@ PUT _connector/my-{service-name-stub}-connector
   "index_name": "my-elasticsearch-index",
   "name": "Content synced from {service-name}",
   "service_type": "{service-name-stub}",
-  "is_native": "true"
+  "is_native": true
 }
 ----
 // TEST[skip:can't test in isolation]

docs/reference/search/retriever.asciidoc

@@ -554,8 +554,8 @@ It then re-ranks the results based on semantic similarity to the text in the `in
 [[rule-retriever]]
 ==== Query Rules Retriever
 
-The `rule` retriever enables fine-grained control over search results by applying contextual <<query-rules>> to pin or exclude documents for specific queries.
-This retriever has similar functionality to the <<query-dsl-rule-query>>, but works out of the box with other retrievers.
+The `rule` retriever enables fine-grained control over search results by applying contextual <<query-rules,query rules>> to pin or exclude documents for specific queries.
+This retriever has similar functionality to the <<query-dsl-rule-query, rule query>>, but works out of the box with other retrievers.
 
 ===== Prerequisites
 

muted-tests.yml

@@ -237,18 +237,12 @@ tests:
 - class: org.elasticsearch.search.basic.SearchWithRandomDisconnectsIT
   method: testSearchWithRandomDisconnects
   issue: https://github.com/elastic/elasticsearch/issues/116175
-- class: org.elasticsearch.indexing.IndexActionIT
-  method: testAutoGenerateIdNoDuplicates
-  issue: https://github.com/elastic/elasticsearch/issues/115716
 - class: org.elasticsearch.xpack.test.rest.XPackRestIT
   method: test {p0=ml/start_stop_datafeed/Test start datafeed given index pattern with no matching indices}
   issue: https://github.com/elastic/elasticsearch/issues/116220
 - class: org.elasticsearch.search.basic.SearchWhileRelocatingIT
   method: testSearchAndRelocateConcurrentlyRandomReplicas
   issue: https://github.com/elastic/elasticsearch/issues/116145
-- class: org.elasticsearch.reservedstate.service.FileSettingsServiceTests
-  method: testProcessFileChanges
-  issue: https://github.com/elastic/elasticsearch/issues/115280
 - class: org.elasticsearch.xpack.test.rest.XPackRestIT
   method: test {p0=ml/filter_crud/Test update filter}
   issue: https://github.com/elastic/elasticsearch/issues/116271
@@ -261,18 +255,6 @@ tests:
 - class: org.elasticsearch.xpack.security.operator.OperatorPrivilegesIT
   method: testEveryActionIsEitherOperatorOnlyOrNonOperator
   issue: https://github.com/elastic/elasticsearch/issues/102992
-- class: org.elasticsearch.datastreams.DataStreamsClientYamlTestSuiteIT
-  issue: https://github.com/elastic/elasticsearch/issues/116291
-- class: org.elasticsearch.xpack.test.rest.XPackRestIT
-  issue: https://github.com/elastic/elasticsearch/issues/114723
-- class: org.elasticsearch.xpack.search.AsyncSearchSecurityIT
-  issue: https://github.com/elastic/elasticsearch/issues/116293
-- class: org.elasticsearch.xpack.downsample.DownsampleRestIT
-  issue: https://github.com/elastic/elasticsearch/issues/116326
-- class: org.elasticsearch.xpack.downsample.DownsampleWithBasicRestIT
-  issue: https://github.com/elastic/elasticsearch/issues/116327
-- class: org.elasticsearch.validation.DotPrefixClientYamlTestSuiteIT
-  issue: https://github.com/elastic/elasticsearch/issues/116328
 - class: org.elasticsearch.action.search.SearchQueryThenFetchAsyncActionTests
   method: testBottomFieldSort
   issue: https://github.com/elastic/elasticsearch/issues/116249
@@ -288,6 +270,9 @@ tests:
 - class: org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshotsIntegTests
   method: testCreateAndRestoreSearchableSnapshot
   issue: https://github.com/elastic/elasticsearch/issues/116377
+- class: org.elasticsearch.threadpool.SimpleThreadPoolIT
+  method: testThreadPoolMetrics
+  issue: https://github.com/elastic/elasticsearch/issues/108320
 
 # Examples:
 #

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

@@ -48,8 +48,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.Executor;
-import java.util.concurrent.LinkedTransferQueue;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -751,7 +751,7 @@ protected final ShardSearchRequest buildShardSearchRequest(SearchShardIterator s
 
     private static final class PendingExecutions {
         private final Semaphore semaphore;
-        private final LinkedTransferQueue<Consumer<Releasable>> queue = new LinkedTransferQueue<>();
+        private final ConcurrentLinkedQueue<Consumer<Releasable>> queue = new ConcurrentLinkedQueue<>();
 
         PendingExecutions(int permits) {
             assert permits > 0 : "not enough permits: " + permits;
@@ -770,11 +770,10 @@ void submit(Consumer<Releasable> task) {
                     }
                 }
             }
-
         }
 
         private void executeAndRelease(Consumer<Releasable> task) {
-            while (task != null) {
+            do {
                 final SubscribableListener<Void> onDone = new SubscribableListener<>();
                 task.accept(() -> onDone.onResponse(null));
                 if (onDone.isDone()) {
@@ -797,13 +796,21 @@ public void onFailure(Exception e) {
                     });
                     return;
                 }
-            }
+            } while (task != null);
         }
 
         private Consumer<Releasable> pollNextTaskOrReleasePermit() {
             var task = queue.poll();
             if (task == null) {
                 semaphore.release();
+                while (queue.peek() != null && semaphore.tryAcquire()) {
+                    task = queue.poll();
+                    if (task == null) {
+                        semaphore.release();
+                    } else {
+                        return task;
+                    }
+                }
             }
             return task;
         }

server/src/main/java/org/elasticsearch/cluster/ClusterState.java

@@ -47,6 +47,7 @@
 import org.elasticsearch.common.xcontent.ChunkedToXContentHelper;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.SuppressForbidden;
+import org.elasticsearch.core.UpdateForV9;
 import org.elasticsearch.index.shard.IndexLongFieldRange;
 import org.elasticsearch.indices.SystemIndexDescriptor;
 import org.elasticsearch.xcontent.ToXContent;
@@ -1025,52 +1026,25 @@ public static ClusterState readFrom(StreamInput in, DiscoveryNode localNode) thr
         builder.metadata = Metadata.readFrom(in);
         builder.routingTable = RoutingTable.readFrom(in);
         builder.nodes = DiscoveryNodes.readFrom(in, localNode);
-        if (in.getTransportVersion().onOrAfter(TransportVersions.V_8_8_0)) {
-            builder.nodeIdsToCompatibilityVersions(in.readMap(CompatibilityVersions::readVersion));
-        } else {
-            // this clusterstate is from a pre-8.8.0 node
-            // infer the versions from discoverynodes for now
-            // leave mappings versions empty
-            builder.nodes()
-                .getNodes()
-                .values()
-                .forEach(n -> builder.putCompatibilityVersions(n.getId(), inferTransportVersion(n), Map.of()));
-        }
-        if (in.getTransportVersion().onOrAfter(TransportVersions.V_8_12_0)) {
-            builder.nodeFeatures(ClusterFeatures.readFrom(in));
-        }
+        builder.nodeIdsToCompatibilityVersions(in.readMap(CompatibilityVersions::readVersion));
+        builder.nodeFeatures(ClusterFeatures.readFrom(in));
         builder.blocks = ClusterBlocks.readFrom(in);
         int customSize = in.readVInt();
         for (int i = 0; i < customSize; i++) {
             Custom customIndexMetadata = in.readNamedWriteable(Custom.class);
             builder.putCustom(customIndexMetadata.getWriteableName(), customIndexMetadata);
         }
-        if (in.getTransportVersion().before(TransportVersions.V_8_0_0)) {
-            in.readVInt(); // used to be minimumMasterNodesOnPublishingMaster, which was used in 7.x for BWC with 6.x
-        }
         return builder.build();
     }
 
     /**
      * If the cluster state does not contain transport version information, this is the version
      * that is inferred for all nodes on version 8.8.0 or above.
      */
+    @UpdateForV9(owner = UpdateForV9.Owner.CORE_INFRA)
     public static final TransportVersion INFERRED_TRANSPORT_VERSION = TransportVersions.V_8_8_0;
-
     public static final Version VERSION_INTRODUCING_TRANSPORT_VERSIONS = Version.V_8_8_0;
 
-    private static TransportVersion inferTransportVersion(DiscoveryNode node) {
-        TransportVersion tv;
-        if (node.getVersion().before(VERSION_INTRODUCING_TRANSPORT_VERSIONS)) {
-            // 1-to-1 mapping between Version and TransportVersion
-            tv = TransportVersion.fromId(node.getPre811VersionId().getAsInt());
-        } else {
-            // use the lowest value it could be for now
-            tv = INFERRED_TRANSPORT_VERSION;
-        }
-        return tv;
-    }
-
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         clusterName.writeTo(out);
@@ -1079,17 +1053,10 @@ public void writeTo(StreamOutput out) throws IOException {
         metadata.writeTo(out);
         routingTable.writeTo(out);
         nodes.writeTo(out);
-        if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_8_0)) {
-            out.writeMap(compatibilityVersions, StreamOutput::writeWriteable);
-        }
-        if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_12_0)) {
-            clusterFeatures.writeTo(out);
-        }
+        out.writeMap(compatibilityVersions, StreamOutput::writeWriteable);
+        clusterFeatures.writeTo(out);
         blocks.writeTo(out);
         VersionedNamedWriteable.writeVersionedWritables(out, customs);
-        if (out.getTransportVersion().before(TransportVersions.V_8_0_0)) {
-            out.writeVInt(-1); // used to be minimumMasterNodesOnPublishingMaster, which was used in 7.x for BWC with 6.x
-        }
     }
 
     private static class ClusterStateDiff implements Diff<ClusterState> {
@@ -1106,7 +1073,6 @@ private static class ClusterStateDiff implements Diff<ClusterState> {
 
         private final Diff<DiscoveryNodes> nodes;
 
-        @Nullable
         private final Diff<Map<String, CompatibilityVersions>> versions;
         private final Diff<ClusterFeatures> features;
 
@@ -1142,26 +1108,13 @@ private static class ClusterStateDiff implements Diff<ClusterState> {
             toVersion = in.readLong();
             routingTable = RoutingTable.readDiffFrom(in);
             nodes = DiscoveryNodes.readDiffFrom(in, localNode);
-            if (in.getTransportVersion().onOrAfter(TransportVersions.V_8_8_0) && in.readBoolean()) {
-                versions = DiffableUtils.readJdkMapDiff(
-                    in,
-                    DiffableUtils.getStringKeySerializer(),
-                    COMPATIBILITY_VERSIONS_VALUE_SERIALIZER
-                );
-            } else {
-                versions = null;   // infer at application time
-            }
-            if (in.getTransportVersion().onOrAfter(TransportVersions.V_8_12_0)) {
-                features = ClusterFeatures.readDiffFrom(in);
-            } else {
-                features = null;    // fill in when nodes re-register with a master that understands features
-            }
+            boolean versionPresent = in.readBoolean();
+            if (versionPresent == false) throw new IOException("ClusterStateDiff stream must have versions");
+            versions = DiffableUtils.readJdkMapDiff(in, DiffableUtils.getStringKeySerializer(), COMPATIBILITY_VERSIONS_VALUE_SERIALIZER);
+            features = ClusterFeatures.readDiffFrom(in);
             metadata = Metadata.readDiffFrom(in);
             blocks = ClusterBlocks.readDiffFrom(in);
             customs = DiffableUtils.readJdkMapDiff(in, DiffableUtils.getStringKeySerializer(), CUSTOM_VALUE_SERIALIZER);
-            if (in.getTransportVersion().before(TransportVersions.V_8_0_0)) {
-                in.readVInt(); // used to be minimumMasterNodesOnPublishingMaster, which was used in 7.x for BWC with 6.x
-            }
         }
 
         @Override
@@ -1172,18 +1125,12 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeLong(toVersion);
             routingTable.writeTo(out);
             nodes.writeTo(out);
-            if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_8_0)) {
-                out.writeOptionalWriteable(versions);
-            }
-            if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_12_0)) {
-                features.writeTo(out);
-            }
+            out.writeBoolean(true);
+            versions.writeTo(out);
+            features.writeTo(out);
             metadata.writeTo(out);
             blocks.writeTo(out);
             customs.writeTo(out);
-            if (out.getTransportVersion().before(TransportVersions.V_8_0_0)) {
-                out.writeVInt(-1); // used to be minimumMasterNodesOnPublishingMaster, which was used in 7.x for BWC with 6.x
-            }
         }
 
         @Override
@@ -1200,19 +1147,8 @@ public ClusterState apply(ClusterState state) {
             builder.version(toVersion);
             builder.routingTable(routingTable.apply(state.routingTable));
             builder.nodes(nodes.apply(state.nodes));
-            if (versions != null) {
-                builder.nodeIdsToCompatibilityVersions(this.versions.apply(state.compatibilityVersions));
-            } else {
-                // infer the versions from discoverynodes for now
-                // leave mappings versions empty
-                builder.nodes()
-                    .getNodes()
-                    .values()
-                    .forEach(n -> builder.putCompatibilityVersions(n.getId(), inferTransportVersion(n), Map.of()));
-            }
-            if (features != null) {
-                builder.nodeFeatures(this.features.apply(state.clusterFeatures));
-            }
+            builder.nodeIdsToCompatibilityVersions(this.versions.apply(state.compatibilityVersions));
+            builder.nodeFeatures(this.features.apply(state.clusterFeatures));
             builder.metadata(metadata.apply(state.metadata));
             builder.blocks(blocks.apply(state.blocks));
             builder.customs(customs.apply(state.customs));

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

@@ -8,16 +8,13 @@
  */
 package org.elasticsearch.cluster.coordination;
 
-import org.elasticsearch.TransportVersion;
-import org.elasticsearch.TransportVersions;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.version.CompatibilityVersions;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.transport.TransportRequest;
 
 import java.io.IOException;
-import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
@@ -72,21 +69,8 @@ public JoinRequest(
     public JoinRequest(StreamInput in) throws IOException {
         super(in);
         sourceNode = new DiscoveryNode(in);
-        if (in.getTransportVersion().onOrAfter(TransportVersions.V_8_8_0)) {
-            compatibilityVersions = CompatibilityVersions.readVersion(in);
-        } else {
-            // there's a 1-1 mapping from Version to TransportVersion before 8.8.0
-            // no known mapping versions here
-            compatibilityVersions = new CompatibilityVersions(
-                TransportVersion.fromId(sourceNode.getPre811VersionId().getAsInt()),
-                Map.of()
-            );
-        }
-        if (in.getTransportVersion().onOrAfter(TransportVersions.V_8_12_0)) {
-            features = in.readCollectionAsSet(StreamInput::readString);
-        } else {
-            features = Set.of();
-        }
+        compatibilityVersions = CompatibilityVersions.readVersion(in);
+        features = in.readCollectionAsSet(StreamInput::readString);
         minimumTerm = in.readLong();
         optionalJoin = Optional.ofNullable(in.readOptionalWriteable(Join::new));
     }
@@ -95,12 +79,8 @@ public JoinRequest(StreamInput in) throws IOException {
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         sourceNode.writeTo(out);
-        if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_8_0)) {
-            compatibilityVersions.writeTo(out);
-        }
-        if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_12_0)) {
-            out.writeCollection(features, StreamOutput::writeString);
-        }
+        compatibilityVersions.writeTo(out);
+        out.writeCollection(features, StreamOutput::writeString);
         out.writeLong(minimumTerm);
         out.writeOptionalWriteable(optionalJoin.orElse(null));
     }