Merge remote-tracking branch 'upstream/main' into es819-merge-optimization

Author: jordan-powers

docs/reference/elasticsearch/configuration-reference/thread-pool-settings.md

@@ -94,11 +94,7 @@ The following are the types of thread pools and their respective parameters:
 
 ### `fixed` [fixed-thread-pool]
 
-The `fixed` thread pool holds a fixed size of threads to handle the requests with a queue (optionally bounded) for pending requests that have no threads to service them.
-
-The `size` parameter controls the number of threads.
-
-The `queue_size` allows to control the size of the queue of pending requests that have no threads to execute them. By default, it is set to `-1` which means its unbounded. When a request comes in and the queue is full, it will abort the request.
+A `fixed` thread pool holds a fixed number of threads as determined by the `size` parameter. If a task is submitted to a `fixed` thread pool and there are fewer than `size` busy threads in the pool then the task will execute immediately. If all the threads are busy when a task is submitted then it will be held in a queue for later execution. The `queue_size` parameter controls the maximum size of this queue. A `queue_size` of `-1` means that the queue is unbounded, but most `fixed` thread pools specify a bound on their queue size by default. If a bounded queue is full then it will reject further work, which typically causes the corresponding requests to fail.
 
 ```yaml
 thread_pool:
@@ -114,6 +110,8 @@ The `scaling` thread pool holds a dynamic number of threads. This number is prop
 
 The `keep_alive` parameter determines how long a thread should be kept around in the thread pool without it doing any work.
 
+If a task is submitted to a `scaling` thread pool when its maximum number of threads are already busy with other tasks, the new task will be held in a queue for later execution. The queue in a `scaling` thread pool is always unbounded.
+
 ```yaml
 thread_pool:
     warmer:

muted-tests.yml

@@ -408,30 +408,12 @@ tests:
 - class: org.elasticsearch.cli.keystore.AddStringKeyStoreCommandTests
   method: testStdinWithMultipleValues
   issue: https://github.com/elastic/elasticsearch/issues/126882
-- class: org.elasticsearch.xpack.search.AsyncSearchActionIT
-  method: testDeleteCancelRunningTask
-  issue: https://github.com/elastic/elasticsearch/issues/126994
-- class: org.elasticsearch.xpack.search.AsyncSearchActionIT
-  method: testMaxResponseSize
-  issue: https://github.com/elastic/elasticsearch/issues/126995
-- class: org.elasticsearch.xpack.search.AsyncSearchActionIT
-  method: testRemoveAsyncIndex
-  issue: https://github.com/elastic/elasticsearch/issues/126975
-- class: org.elasticsearch.xpack.search.AsyncSearchActionIT
-  method: testCleanupOnFailure
-  issue: https://github.com/elastic/elasticsearch/issues/126999
-- class: org.elasticsearch.xpack.search.AsyncSearchActionIT
-  method: testUpdateStoreKeepAlive
-  issue: https://github.com/elastic/elasticsearch/issues/127001
-- class: org.elasticsearch.xpack.search.AsyncSearchActionIT
-  method: testRestartAfterCompletion
-  issue: https://github.com/elastic/elasticsearch/issues/126974
-- class: org.elasticsearch.xpack.search.AsyncSearchActionIT
-  method: testDeleteCleanupIndex
-  issue: https://github.com/elastic/elasticsearch/issues/127008
 - class: org.elasticsearch.packaging.test.DockerTests
   method: test024InstallPluginFromArchiveUsingConfigFile
   issue: https://github.com/elastic/elasticsearch/issues/126936
+- class: org.elasticsearch.repositories.blobstore.testkit.analyze.RepositoryAnalysisFailureIT
+  method: testFailsOnReadError
+  issue: https://github.com/elastic/elasticsearch/issues/127029
 
 # Examples:
 #

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

@@ -26,6 +26,7 @@
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Releasable;
+import org.elasticsearch.core.Releasables;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.search.SearchPhaseResult;
 import org.elasticsearch.search.SearchService;
@@ -162,7 +163,7 @@ public void consumeResult(SearchPhaseResult result, Runnable next) {
         consume(querySearchResult, next);
     }
 
-    private final List<Tuple<TopDocsStats, MergeResult>> batchedResults = new ArrayList<>();
+    private final ArrayDeque<Tuple<TopDocsStats, MergeResult>> batchedResults = new ArrayDeque<>();
 
     /**
      * Unlinks partial merge results from this instance and returns them as a partial merge result to be sent to the coordinating node.
@@ -214,7 +215,7 @@ public SearchPhaseController.ReducedQueryPhase reduce() throws Exception {
         buffer.sort(RESULT_COMPARATOR);
         final TopDocsStats topDocsStats = this.topDocsStats;
         var mergeResult = this.mergeResult;
-        final List<Tuple<TopDocsStats, MergeResult>> batchedResults;
+        final ArrayDeque<Tuple<TopDocsStats, MergeResult>> batchedResults;
         synchronized (this.batchedResults) {
             batchedResults = this.batchedResults;
         }
@@ -226,8 +227,8 @@ public SearchPhaseController.ReducedQueryPhase reduce() throws Exception {
         if (mergeResult != null) {
             consumePartialMergeResult(mergeResult, topDocsList, aggsList);
         }
-        for (int i = 0; i < batchedResults.size(); i++) {
-            Tuple<TopDocsStats, MergeResult> batchedResult = batchedResults.set(i, null);
+        Tuple<TopDocsStats, MergeResult> batchedResult;
+        while ((batchedResult = batchedResults.poll()) != null) {
             topDocsStats.add(batchedResult.v1());
             consumePartialMergeResult(batchedResult.v2(), topDocsList, aggsList);
         }
@@ -528,6 +529,12 @@ private void releaseBuffer() {
                 querySearchResult.releaseAggs();
             }
         }
+        synchronized (this.batchedResults) {
+            Tuple<TopDocsStats, MergeResult> batchedResult;
+            while ((batchedResult = batchedResults.poll()) != null) {
+                Releasables.close(batchedResult.v2().reducedAggs());
+            }
+        }
     }
 
     private synchronized void onMergeFailure(Exception exc) {