Merge branch 'main' into 2025/04/15/ShardRoutingState

Author: DiannaHohensee

distribution/tools/server-cli/src/main/java/org/elasticsearch/server/cli/ServerCli.java

@@ -271,8 +271,7 @@ protected ServerProcess startServer(Terminal terminal, ProcessInfo processInfo,
             .withProcessInfo(processInfo)
             .withServerArgs(args)
             .withTempDir(tempDir)
-            .withJvmOptions(jvmOptions)
-            .withWorkingDir(args.logsDir());
+            .withJvmOptions(jvmOptions);
         return serverProcessBuilder.start();
     }
 

distribution/tools/server-cli/src/main/java/org/elasticsearch/server/cli/ServerProcessBuilder.java

@@ -10,6 +10,7 @@
 package org.elasticsearch.server.cli;
 
 import org.elasticsearch.bootstrap.ServerArgs;
+import org.elasticsearch.cli.ExitCodes;
 import org.elasticsearch.cli.ProcessInfo;
 import org.elasticsearch.cli.Terminal;
 import org.elasticsearch.cli.UserException;
@@ -21,6 +22,8 @@
 import java.io.IOException;
 import java.io.OutputStream;
 import java.io.UncheckedIOException;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.HashMap;
 import java.util.List;
@@ -44,7 +47,6 @@ public class ServerProcessBuilder {
     private ServerArgs serverArgs;
     private ProcessInfo processInfo;
     private List<String> jvmOptions;
-    private Path workingDir;
     private Terminal terminal;
 
     // this allows mocking the process building by tests
@@ -84,11 +86,6 @@ public ServerProcessBuilder withJvmOptions(List<String> jvmOptions) {
         return this;
     }
 
-    public ServerProcessBuilder withWorkingDir(Path workingDir) {
-        this.workingDir = workingDir;
-        return this;
-    }
-
     /**
      * Specifies the {@link Terminal} to use for reading input and writing output from/to the cli console
      */
@@ -141,6 +138,17 @@ public ServerProcess start() throws UserException {
         return start(ProcessBuilder::start);
     }
 
+    private void ensureWorkingDirExists() throws UserException {
+        Path workingDir = serverArgs.logsDir();
+        try {
+            Files.createDirectories(workingDir);
+        } catch (FileAlreadyExistsException e) {
+            throw new UserException(ExitCodes.CONFIG, "Logs dir [" + workingDir + "] exists but is not a directory", e);
+        } catch (IOException e) {
+            throw new UserException(ExitCodes.CONFIG, "Unable to create logs dir [" + workingDir + "]", e);
+        }
+    }
+
     private static void checkRequiredArgument(Object argument, String argumentName) {
         if (argument == null) {
             throw new IllegalStateException(
@@ -157,12 +165,14 @@ ServerProcess start(ProcessStarter processStarter) throws UserException {
         checkRequiredArgument(jvmOptions, "jvmOptions");
         checkRequiredArgument(terminal, "terminal");
 
+        ensureWorkingDirExists();
+
         Process jvmProcess = null;
         ErrorPumpThread errorPump;
 
         boolean success = false;
         try {
-            jvmProcess = createProcess(getCommand(), getJvmArgs(), jvmOptions, getEnvironment(), workingDir, processStarter);
+            jvmProcess = createProcess(getCommand(), getJvmArgs(), jvmOptions, getEnvironment(), serverArgs.logsDir(), processStarter);
             errorPump = new ErrorPumpThread(terminal, jvmProcess.getErrorStream());
             errorPump.start();
             sendArgs(serverArgs, jvmProcess.getOutputStream());

distribution/tools/server-cli/src/test/java/org/elasticsearch/server/cli/ServerProcessTests.java

@@ -65,7 +65,7 @@ public class ServerProcessTests extends ESTestCase {
     protected final Map<String, String> sysprops = new HashMap<>();
     protected final Map<String, String> envVars = new HashMap<>();
     Path esHomeDir;
-    Path workingDir;
+    Path logsDir;
     Settings.Builder nodeSettings;
     ProcessValidator processValidator;
     MainMethod mainCallback;
@@ -94,8 +94,8 @@ public void resetEnv() {
         sysprops.put("os.name", "Linux");
         sysprops.put("java.home", "javahome");
         sysprops.put("es.path.home", esHomeDir.toString());
+        logsDir = esHomeDir.resolve("logs");
         envVars.clear();
-        workingDir = createTempDir();
         nodeSettings = Settings.builder();
         processValidator = null;
         mainCallback = null;
@@ -207,15 +207,7 @@ ProcessInfo createProcessInfo() {
     }
 
     ServerArgs createServerArgs(boolean daemonize, boolean quiet) {
-        return new ServerArgs(
-            daemonize,
-            quiet,
-            null,
-            secrets,
-            nodeSettings.build(),
-            esHomeDir.resolve("config"),
-            esHomeDir.resolve("logs")
-        );
+        return new ServerArgs(daemonize, quiet, null, secrets, nodeSettings.build(), esHomeDir.resolve("config"), logsDir);
     }
 
     ServerProcess startProcess(boolean daemonize, boolean quiet) throws Exception {
@@ -231,8 +223,7 @@ ServerProcess startProcess(boolean daemonize, boolean quiet) throws Exception {
             .withProcessInfo(pinfo)
             .withServerArgs(createServerArgs(daemonize, quiet))
             .withJvmOptions(List.of())
-            .withTempDir(ServerProcessUtils.setupTempDir(pinfo))
-            .withWorkingDir(workingDir);
+            .withTempDir(ServerProcessUtils.setupTempDir(pinfo));
         return serverProcessBuilder.start(starter);
     }
 
@@ -241,7 +232,7 @@ public void testProcessBuilder() throws Exception {
             assertThat(pb.redirectInput(), equalTo(ProcessBuilder.Redirect.PIPE));
             assertThat(pb.redirectOutput(), equalTo(ProcessBuilder.Redirect.INHERIT));
             assertThat(pb.redirectError(), equalTo(ProcessBuilder.Redirect.PIPE));
-            assertThat(String.valueOf(pb.directory()), equalTo(workingDir.toString())); // leave default, which is working directory
+            assertThat(String.valueOf(pb.directory()), equalTo(esHomeDir.resolve("logs").toString()));
         };
         mainCallback = (args, stdin, stderr, exitCode) -> {
             try (PrintStream err = new PrintStream(stderr, true, StandardCharsets.UTF_8)) {
@@ -315,8 +306,7 @@ public void testCommandLineSysprops() throws Exception {
             .withProcessInfo(createProcessInfo())
             .withServerArgs(createServerArgs(false, false))
             .withJvmOptions(List.of("-Dfoo1=bar", "-Dfoo2=baz"))
-            .withTempDir(Path.of("."))
-            .withWorkingDir(workingDir);
+            .withTempDir(Path.of("."));
         serverProcessBuilder.start(starter).waitFor();
     }
 
@@ -433,4 +423,26 @@ public void testProcessDies() throws Exception {
         int exitCode = server.waitFor();
         assertThat(exitCode, equalTo(-9));
     }
+
+    public void testLogsDirIsFile() throws Exception {
+        Files.createFile(logsDir);
+        var e = expectThrows(UserException.class, this::runForeground);
+        assertThat(e.getMessage(), containsString("exists but is not a directory"));
+    }
+
+    public void testLogsDirCreateParents() throws Exception {
+        Path testDir = createTempDir();
+        logsDir = testDir.resolve("subdir/logs");
+        processValidator = pb -> assertThat(String.valueOf(pb.directory()), equalTo(logsDir.toString()));
+        runForeground();
+    }
+
+    public void testLogsCreateFailure() throws Exception {
+        Path testDir = createTempDir();
+        Path parentFile = testDir.resolve("exists");
+        Files.createFile(parentFile);
+        logsDir = parentFile.resolve("logs");
+        var e = expectThrows(UserException.class, this::runForeground);
+        assertThat(e.getMessage(), containsString("Unable to create logs dir"));
+    }
 }

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) {

test/framework/build.gradle

@@ -32,9 +32,12 @@ dependencies {
 
   api "org.elasticsearch:mocksocket:${versions.mocksocket}"
 
+  testImplementation project(":modules:mapper-extras")
+  testImplementation project(':x-pack:plugin:core')
   testImplementation project(':x-pack:plugin:mapper-unsigned-long')
   testImplementation project(':x-pack:plugin:mapper-counted-keyword')
-  testImplementation project(":modules:mapper-extras")
+  testImplementation project(':x-pack:plugin:mapper-constant-keyword')
+  testImplementation project(':x-pack:plugin:wildcard')
 }
 
 sourceSets {

test/framework/src/main/java/org/elasticsearch/logsdb/datageneration/FieldType.java

@@ -12,6 +12,7 @@
 import org.elasticsearch.logsdb.datageneration.datasource.DataSource;
 import org.elasticsearch.logsdb.datageneration.fields.leaf.BooleanFieldDataGenerator;
 import org.elasticsearch.logsdb.datageneration.fields.leaf.ByteFieldDataGenerator;
+import org.elasticsearch.logsdb.datageneration.fields.leaf.ConstantKeywordFieldDataGenerator;
 import org.elasticsearch.logsdb.datageneration.fields.leaf.CountedKeywordFieldDataGenerator;
 import org.elasticsearch.logsdb.datageneration.fields.leaf.DateFieldDataGenerator;
 import org.elasticsearch.logsdb.datageneration.fields.leaf.DoubleFieldDataGenerator;
@@ -26,6 +27,7 @@
 import org.elasticsearch.logsdb.datageneration.fields.leaf.ShortFieldDataGenerator;
 import org.elasticsearch.logsdb.datageneration.fields.leaf.TextFieldDataGenerator;
 import org.elasticsearch.logsdb.datageneration.fields.leaf.UnsignedLongFieldDataGenerator;
+import org.elasticsearch.logsdb.datageneration.fields.leaf.WildcardFieldDataGenerator;
 
 /**
  * Lists all leaf field types that are supported for data generation by default.
@@ -46,7 +48,9 @@ public enum FieldType {
     DATE("date"),
     GEO_POINT("geo_point"),
     TEXT("text"),
-    IP("ip");
+    IP("ip"),
+    CONSTANT_KEYWORD("constant_keyword"),
+    WILDCARD("wildcard");
 
     private final String name;
 
@@ -56,7 +60,7 @@ public enum FieldType {
 
     public FieldDataGenerator generator(String fieldName, DataSource dataSource) {
         return switch (this) {
-            case KEYWORD -> new KeywordFieldDataGenerator(fieldName, dataSource);
+            case KEYWORD -> new KeywordFieldDataGenerator(dataSource);
             case LONG -> new LongFieldDataGenerator(fieldName, dataSource);
             case UNSIGNED_LONG -> new UnsignedLongFieldDataGenerator(fieldName, dataSource);
             case INTEGER -> new IntegerFieldDataGenerator(fieldName, dataSource);
@@ -72,6 +76,8 @@ public FieldDataGenerator generator(String fieldName, DataSource dataSource) {
             case GEO_POINT -> new GeoPointFieldDataGenerator(dataSource);
             case TEXT -> new TextFieldDataGenerator(dataSource);
             case IP -> new IpFieldDataGenerator(dataSource);
+            case CONSTANT_KEYWORD -> new ConstantKeywordFieldDataGenerator();
+            case WILDCARD -> new WildcardFieldDataGenerator(dataSource);
         };
     }
 
@@ -93,6 +99,8 @@ public static FieldType tryParse(String name) {
             case "geo_point" -> FieldType.GEO_POINT;
             case "text" -> FieldType.TEXT;
             case "ip" -> FieldType.IP;
+            case "constant_keyword" -> FieldType.CONSTANT_KEYWORD;
+            case "wildcard" -> FieldType.WILDCARD;
             default -> null;
         };
     }

test/framework/src/main/java/org/elasticsearch/logsdb/datageneration/datasource/DataSourceHandler.java

@@ -78,6 +78,10 @@ default DataSourceResponse.IpGenerator handle(DataSourceRequest.IpGenerator requ
         return null;
     }
 
+    default DataSourceResponse.VersionStringGenerator handle(DataSourceRequest.VersionStringGenerator request) {
+        return null;
+    }
+
     default DataSourceResponse.NullWrapper handle(DataSourceRequest.NullWrapper request) {
         return null;
     }

test/framework/src/main/java/org/elasticsearch/logsdb/datageneration/datasource/DataSourceRequest.java

@@ -126,6 +126,12 @@ public DataSourceResponse.IpGenerator accept(DataSourceHandler handler) {
         }
     }
 
+    record VersionStringGenerator() implements DataSourceRequest<DataSourceResponse.VersionStringGenerator> {
+        public DataSourceResponse.VersionStringGenerator accept(DataSourceHandler handler) {
+            return handler.handle(this);
+        }
+    }
+
     record NullWrapper() implements DataSourceRequest<DataSourceResponse.NullWrapper> {
         public DataSourceResponse.NullWrapper accept(DataSourceHandler handler) {
             return handler.handle(this);