Merge branch 'main' into 2025/09/09/cluster-applier-thread-watchdog

Author: DaveCTurner

build-tools-internal/src/main/java/org/elasticsearch/gradle/internal/InternalTestClustersPlugin.java

@@ -11,9 +11,7 @@
 
 import org.elasticsearch.gradle.VersionProperties;
 import org.elasticsearch.gradle.internal.info.GlobalBuildInfoPlugin;
-import org.elasticsearch.gradle.testclusters.ElasticsearchCluster;
 import org.elasticsearch.gradle.testclusters.TestClustersPlugin;
-import org.gradle.api.NamedDomainObjectContainer;
 import org.gradle.api.Plugin;
 import org.gradle.api.Project;
 
@@ -33,17 +31,5 @@ public void apply(Project project) {
             version -> (version.equals(VersionProperties.getElasticsearchVersion()) && buildParams.getSnapshotBuild() == false)
                 || buildParams.getBwcVersions().unreleasedInfo(version) == null
         );
-
-        NamedDomainObjectContainer<ElasticsearchCluster> testClusters = (NamedDomainObjectContainer<ElasticsearchCluster>) project
-            .getExtensions()
-            .getByName(TestClustersPlugin.EXTENSION_NAME);
-        // Limit the number of allocated processors for all nodes to 2 in the cluster by default.
-        // This is to ensure that the tests run consistently across different environments.
-        String processorCount = shouldConfigureTestClustersWithOneProcessor() ? "1" : "2";
-        testClusters.configureEach(elasticsearchCluster -> elasticsearchCluster.setting("node.processors", processorCount));
-    }
-
-    private boolean shouldConfigureTestClustersWithOneProcessor() {
-        return Boolean.parseBoolean(System.getProperty("tests.configure_test_clusters_with_one_processor", "false"));
     }
 }

build-tools/src/main/java/org/elasticsearch/gradle/testclusters/ElasticsearchNode.java

@@ -1404,6 +1404,15 @@ private void createConfiguration() {
             baseConfig.put("cluster.service.slow_master_task_logging_threshold", "5s");
         }
 
+        // Limit the number of allocated processors for all nodes in the cluster by default.
+        // This is to ensure that the tests run consistently across different environments.
+        String processorCount = shouldConfigureTestClustersWithOneProcessor() ? "1" : "2";
+        if (getVersion().onOrAfter("7.6.0")) {
+            baseConfig.put("node.processors", processorCount);
+        } else {
+            baseConfig.put("processors", processorCount);
+        }
+
         baseConfig.put("action.destructive_requires_name", "false");
 
         HashSet<String> overriden = new HashSet<>(baseConfig.keySet());
@@ -1789,4 +1798,8 @@ private static class LinkCreationException extends UncheckedIOException {
             super(message, cause);
         }
     }
+
+    private boolean shouldConfigureTestClustersWithOneProcessor() {
+        return Boolean.parseBoolean(System.getProperty("tests.configure_test_clusters_with_one_processor", "false"));
+    }
 }

docs/changelog/135734.yaml

@@ -0,0 +1,5 @@
+pr: 135734
+summary: Fill in topn values if competitive
+area: ES|QL
+type: enhancement
+issues: []

muted-tests.yml

@@ -633,6 +633,33 @@ tests:
 - class: org.elasticsearch.datastreams.DataStreamIndexSettingsProviderTests
   method: testGetAdditionalIndexSettingsMappingsMerging
   issue: https://github.com/elastic/elasticsearch/issues/135884
+- class: org.elasticsearch.upgrades.FullClusterRestartDownsampleIT
+  method: testRollupIndex {cluster=UPGRADED}
+  issue: https://github.com/elastic/elasticsearch/issues/135906
+- class: org.elasticsearch.upgrades.FullClusterRestartIT
+  method: testNewReplicasTimeSeriesMode {cluster=UPGRADED}
+  issue: https://github.com/elastic/elasticsearch/issues/135907
+- class: org.elasticsearch.upgrades.FullClusterRestartIT
+  method: testSingleDoc {cluster=UPGRADED}
+  issue: https://github.com/elastic/elasticsearch/issues/135908
+- class: org.elasticsearch.upgrades.FullClusterRestartIT
+  method: testResize {cluster=UPGRADED}
+  issue: https://github.com/elastic/elasticsearch/issues/135909
+- class: org.elasticsearch.upgrades.FullClusterRestartIT
+  method: testSystemIndexMetadataIsUpgraded {cluster=UPGRADED}
+  issue: https://github.com/elastic/elasticsearch/issues/135923
+- class: org.elasticsearch.upgrades.FullClusterRestartIT
+  method: testSearch {cluster=UPGRADED}
+  issue: https://github.com/elastic/elasticsearch/issues/135927
+- class: org.elasticsearch.xpack.inference.chunking.ChunkingSettingsBuilderTests
+  method: testBuildChunkingSettingsForElasticReranker_QueryTokenCountLessThanHalfOfTokenLimit
+  issue: https://github.com/elastic/elasticsearch/issues/135928
+- class: org.elasticsearch.upgrades.FullClusterRestartIT
+  method: testPeerRecoveryRetentionLeases {cluster=UPGRADED}
+  issue: https://github.com/elastic/elasticsearch/issues/135929
+- class: org.elasticsearch.upgrades.FullClusterRestartIT
+  method: testPersianAnalyzerBWC {cluster=UPGRADED}
+  issue: https://github.com/elastic/elasticsearch/issues/135930
 
 # Examples:
 #

rest-api-spec/src/main/resources/rest-api-spec/api/project.tags.json

@@ -0,0 +1,21 @@
+{
+  "project.tags": {
+    "documentation": {
+      "url": null,
+      "description": "Return tags defined for the project"
+    },
+    "stability": "experimental",
+    "visibility": "public",
+    "headers": {
+      "accept": ["application/json"]
+    },
+    "url": {
+      "paths": [
+        {
+          "path": "/_project/tags",
+          "methods": ["GET"]
+        }
+      ]
+    }
+  }
+}

test/test-clusters/src/main/java/org/elasticsearch/test/cluster/local/DefaultSettingsProvider.java

@@ -44,7 +44,11 @@ public Map<String, String> get(LocalNodeSpec nodeSpec) {
 
         // Limit the number of allocated processors for all nodes in the cluster by default.
         // This is to ensure that the tests run consistently across different environments.
-        settings.put("node.processors", "2");
+        if (nodeSpec.getVersion().onOrAfter("7.6.0")) {
+            settings.put("node.processors", "2");
+        } else {
+            settings.put("processors", "2");
+        }
 
         // Default the watermarks to absurdly low to prevent the tests from failing on nodes without enough disk space
         settings.put("cluster.routing.allocation.disk.watermark.low", "1b");

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNOperator.java

@@ -199,15 +199,7 @@ static final class RowFiller {
             }
         }
 
-        /**
-         * Fill a {@link Row} for {@code position}.
-         */
-        void row(int position, Row destination) {
-            writeKey(position, destination);
-            writeValues(position, destination);
-        }
-
-        private void writeKey(int position, Row row) {
+        void writeKey(int position, Row row) {
             int orderByCompositeKeyCurrentPosition = 0;
             for (int i = 0; i < keyFactories.length; i++) {
                 int valueAsBytesSize = keyFactories[i].extractor.writeKey(row.keys, position);
@@ -217,7 +209,7 @@ private void writeKey(int position, Row row) {
             }
         }
 
-        private void writeValues(int position, Row destination) {
+        void writeValues(int position, Row destination) {
             for (ValueExtractor e : valueExtractors) {
                 var refCounted = e.getRefCountedForShard(position);
                 if (refCounted != null) {
@@ -416,15 +408,28 @@ public void addInput(Page page) {
                     spare.values.clear();
                     spare.clearRefCounters();
                 }
-                rowFiller.row(i, spare);
+                rowFiller.writeKey(i, spare);
 
                 // When rows are very long, appending the values one by one can lead to lots of allocations.
                 // To avoid this, pre-allocate at least as much size as in the last seen row.
                 // Let the pre-allocation size decay in case we only have 1 huge row and smaller rows otherwise.
                 spareKeysPreAllocSize = Math.max(spare.keys.length(), spareKeysPreAllocSize / 2);
-                spareValuesPreAllocSize = Math.max(spare.values.length(), spareValuesPreAllocSize / 2);
 
-                spare = inputQueue.insertWithOverflow(spare);
+                // This is `inputQueue.insertWithOverflow` with followed by filling in the value only if we inserted.
+                if (inputQueue.size() < inputQueue.topCount) {
+                    // Heap not yet full, just add elements
+                    rowFiller.writeValues(i, spare);
+                    spareValuesPreAllocSize = Math.max(spare.values.length(), spareValuesPreAllocSize / 2);
+                    inputQueue.add(spare);
+                    spare = null;
+                } else if (inputQueue.lessThan(inputQueue.top(), spare)) {
+                    // Heap full AND this node fit in it.
+                    Row nextSpare = inputQueue.top();
+                    rowFiller.writeValues(i, spare);
+                    spareValuesPreAllocSize = Math.max(spare.values.length(), spareValuesPreAllocSize / 2);
+                    inputQueue.updateTop(spare);
+                    spare = nextSpare;
+                }
             }
         } finally {
             page.releaseBlocks();

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/topn/TopNOperatorTests.java

@@ -92,7 +92,6 @@
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.hasSize;
-import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.lessThan;
 import static org.hamcrest.Matchers.lessThanOrEqualTo;
 
@@ -470,7 +469,8 @@ private TopNOperator.Row row(
             page
         );
         TopNOperator.Row row = new TopNOperator.Row(nonBreakingBigArrays().breakerService().getBreaker("request"), sortOrders, 0, 0);
-        rf.row(position, row);
+        rf.writeKey(position, row);
+        rf.writeValues(position, row);
         return row;
     }
 
@@ -1468,14 +1468,15 @@ public void testRowResizes() {
         );
         List<ElementType> types = Collections.nCopies(columns, INT);
         List<TopNEncoder> encoders = Collections.nCopies(columns, DEFAULT_UNSORTABLE);
+        boolean asc = randomBoolean();
         try (
             TopNOperator op = new TopNOperator(
                 driverContext().blockFactory(),
                 breaker,
                 10,
                 types,
                 encoders,
-                List.of(new TopNOperator.SortOrder(0, randomBoolean(), randomBoolean())),
+                List.of(new TopNOperator.SortOrder(0, asc, randomBoolean())),
                 randomPageSize()
             )
         ) {
@@ -1491,7 +1492,8 @@ public void testRowResizes() {
 
             // 105 are from the objects
             // 1 is for the min-heap itself
-            assertThat(breaker.getMemoryRequestCount(), is(106L));
+            // -1 IF we're sorting ascending. We encode one less value.
+            assertThat(breaker.getMemoryRequestCount(), equalTo(asc ? 105L : 106L));
         }
     }