Merge branch 'master' into ccr

* master:
  Skip shard refreshes if shard is `search idle` (#27500)
  Remove workaround in translog rest test (#27530)
  inner_hits: Return an empty _source for nested inner hit when filtering on a field that doesn't exist.
  percolator: Avoid TooManyClauses exception if number of terms / ranges is exactly equal to 1024
  Dedup translog operations by reading in reverse (#27268)
  Ensure logging is configured for CLI commands
  Ensure `doc_stats` are changing even if refresh is disabled (#27505)
  Fix classes that can exit
  Revert "Adjust CombinedDeletionPolicy for multiple commits (#27456)"
  Transpose expected and actual, and remove duplicate info from message. (#27515)
  [DOCS] Fixed broken link in breaking changes

Author: jasontedor

core/cli/src/main/java/org/elasticsearch/cli/Command.java

@@ -38,6 +38,8 @@ public abstract class Command implements Closeable {
     /** A description of the command, used in the help output. */
     protected final String description;
 
+    private final Runnable beforeMain;
+
     /** The option parser for this command. */
     protected final OptionParser parser = new OptionParser();
 
@@ -46,8 +48,15 @@ public abstract class Command implements Closeable {
     private final OptionSpec<Void> verboseOption =
         parser.acceptsAll(Arrays.asList("v", "verbose"), "show verbose output").availableUnless(silentOption);
 
-    public Command(String description) {
+    /**
+     * Construct the command with the specified command description and runnable to execute before main is invoked.
+     *
+     * @param description the command description
+     * @param beforeMain the before-main runnable
+     */
+    public Command(final String description, final Runnable beforeMain) {
         this.description = description;
+        this.beforeMain = beforeMain;
     }
 
     private Thread shutdownHookThread;
@@ -75,7 +84,7 @@ public final int main(String[] args, Terminal terminal) throws Exception {
             Runtime.getRuntime().addShutdownHook(shutdownHookThread);
         }
 
-        beforeExecute();
+        beforeMain.run();
 
         try {
             mainWithoutErrorHandling(args, terminal);
@@ -93,12 +102,6 @@ public final int main(String[] args, Terminal terminal) throws Exception {
         return ExitCodes.OK;
     }
 
-    /**
-     * Setup method to be executed before parsing or execution of the command being run. Any exceptions thrown by the
-     * method will not be cleanly caught by the parser.
-     */
-    protected void beforeExecute() {}
-
     /**
      * Executes the command, but all errors are thrown.
      */

core/cli/src/main/java/org/elasticsearch/cli/MultiCommand.java

@@ -35,8 +35,14 @@ public class MultiCommand extends Command {
 
     private final NonOptionArgumentSpec<String> arguments = parser.nonOptions("command");
 
-    public MultiCommand(String description) {
-        super(description);
+    /**
+     * Construct the multi-command with the specified command description and runnable to execute before main is invoked.
+     *
+     * @param description the multi-command description
+     * @param beforeMain the before-main runnable
+     */
+    public MultiCommand(final String description, final Runnable beforeMain) {
+        super(description, beforeMain);
         parser.posixlyCorrect(true);
     }
 

core/src/main/java/org/elasticsearch/action/explain/TransportExplainAction.java

@@ -33,8 +33,10 @@
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.lease.Releasables;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.IndexService;
 import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.get.GetResult;
+import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.search.SearchService;
 import org.elasticsearch.search.internal.AliasFilter;
@@ -86,6 +88,19 @@ protected void resolveRequest(ClusterState state, InternalRequest request) {
         }
     }
 
+    @Override
+    protected void asyncShardOperation(ExplainRequest request, ShardId shardId, ActionListener<ExplainResponse> listener) throws IOException {
+        IndexService indexService = searchService.getIndicesService().indexServiceSafe(shardId.getIndex());
+        IndexShard indexShard = indexService.getShard(shardId.id());
+        indexShard.awaitShardSearchActive(b -> {
+            try {
+                super.asyncShardOperation(request, shardId, listener);
+            } catch (Exception ex) {
+                listener.onFailure(ex);
+            }
+        });
+    }
+
     @Override
     protected ExplainResponse shardOperation(ExplainRequest request, ShardId shardId) throws IOException {
         ShardSearchLocalRequest shardSearchLocalRequest = new ShardSearchLocalRequest(shardId,

core/src/main/java/org/elasticsearch/action/get/TransportGetAction.java

@@ -19,13 +19,13 @@
 
 package org.elasticsearch.action.get;
 
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.RoutingMissingException;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.single.shard.TransportSingleShardAction;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
-import org.elasticsearch.cluster.routing.Preference;
 import org.elasticsearch.cluster.routing.ShardIterator;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
@@ -38,6 +38,8 @@
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
+import java.io.IOException;
+
 /**
  * Performs the get operation.
  */
@@ -76,6 +78,23 @@ protected void resolveRequest(ClusterState state, InternalRequest request) {
         }
     }
 
+    @Override
+    protected void asyncShardOperation(GetRequest request, ShardId shardId, ActionListener<GetResponse> listener) throws IOException {
+        IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
+        IndexShard indexShard = indexService.getShard(shardId.id());
+        if (request.realtime()) { // we are not tied to a refresh cycle here anyway
+            listener.onResponse(shardOperation(request, shardId));
+        } else {
+            indexShard.awaitShardSearchActive(b -> {
+                try {
+                    super.asyncShardOperation(request, shardId, listener);
+                } catch (Exception ex) {
+                    listener.onFailure(ex);
+                }
+            });
+        }
+    }
+
     @Override
     protected GetResponse shardOperation(GetRequest request, ShardId shardId) {
         IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());

core/src/main/java/org/elasticsearch/action/support/single/shard/TransportSingleShardAction.java

@@ -38,6 +38,7 @@
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.logging.LoggerMessageFormat;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportChannel;
@@ -47,6 +48,8 @@
 import org.elasticsearch.transport.TransportService;
 
 import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.concurrent.Executor;
 import java.util.function.Supplier;
 
 import static org.elasticsearch.action.support.TransportActions.isShardNotAvailableException;
@@ -78,7 +81,7 @@ protected TransportSingleShardAction(Settings settings, String actionName, Threa
         if (!isSubAction()) {
             transportService.registerRequestHandler(actionName, request, ThreadPool.Names.SAME, new TransportHandler());
         }
-        transportService.registerRequestHandler(transportShardAction, request, executor, new ShardTransportHandler());
+        transportService.registerRequestHandler(transportShardAction, request, ThreadPool.Names.SAME, new ShardTransportHandler());
     }
 
     /**
@@ -97,6 +100,19 @@ protected void doExecute(Request request, ActionListener<Response> listener) {
 
     protected abstract Response shardOperation(Request request, ShardId shardId) throws IOException;
 
+    protected void asyncShardOperation(Request request, ShardId shardId, ActionListener<Response> listener) throws IOException {
+        threadPool.executor(this.executor).execute(new AbstractRunnable() {
+            @Override
+            public void onFailure(Exception e) {
+                listener.onFailure(e);
+            }
+
+            @Override
+            protected void doRun() throws Exception {
+                listener.onResponse(shardOperation(request, shardId));
+            }
+        });
+    }
     protected abstract Response newResponse();
 
     protected abstract boolean resolveIndex(Request request);
@@ -291,11 +307,27 @@ public void messageReceived(final Request request, final TransportChannel channe
             if (logger.isTraceEnabled()) {
                 logger.trace("executing [{}] on shard [{}]", request, request.internalShardId);
             }
-            Response response = shardOperation(request, request.internalShardId);
-            channel.sendResponse(response);
+            asyncShardOperation(request, request.internalShardId, new ActionListener<Response>() {
+                @Override
+                public void onResponse(Response response) {
+                    try {
+                        channel.sendResponse(response);
+                    } catch (IOException e) {
+                        onFailure(e);
+                    }
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    try {
+                        channel.sendResponse(e);
+                    } catch (IOException e1) {
+                        throw new UncheckedIOException(e1);
+                    }
+                }
+            });
         }
     }
-
     /**
      * Internal request class that gets built on each node. Holds the original request plus additional info.
      */

core/src/main/java/org/elasticsearch/action/termvectors/TransportTermVectorsAction.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.action.termvectors;
 
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.RoutingMissingException;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.single.shard.TransportSingleShardAction;
@@ -37,6 +38,8 @@
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
+import java.io.IOException;
+
 /**
  * Performs the get operation.
  */
@@ -82,6 +85,23 @@ protected void resolveRequest(ClusterState state, InternalRequest request) {
         }
     }
 
+    @Override
+    protected void asyncShardOperation(TermVectorsRequest request, ShardId shardId, ActionListener<TermVectorsResponse> listener) throws IOException {
+        IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
+        IndexShard indexShard = indexService.getShard(shardId.id());
+        if (request.realtime()) { // it's a realtime request which is not subject to refresh cycles
+            listener.onResponse(shardOperation(request, shardId));
+        } else {
+            indexShard.awaitShardSearchActive(b -> {
+                try {
+                    super.asyncShardOperation(request, shardId, listener);
+                } catch (Exception ex) {
+                    listener.onFailure(ex);
+                }
+            });
+        }
+    }
+
     @Override
     protected TermVectorsResponse shardOperation(TermVectorsRequest request, ShardId shardId) {
         IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());

core/src/main/java/org/elasticsearch/bootstrap/Elasticsearch.java

@@ -51,7 +51,7 @@ class Elasticsearch extends EnvironmentAwareCommand {
 
     // visible for testing
     Elasticsearch() {
-        super("starts elasticsearch");
+        super("starts elasticsearch", () -> {}); // we configure logging later so we override the base class from configuring logging
         versionOption = parser.acceptsAll(Arrays.asList("V", "version"),
             "Prints elasticsearch version information and exits");
         daemonizeOption = parser.acceptsAll(Arrays.asList("d", "daemonize"),
@@ -92,15 +92,6 @@ static int main(final String[] args, final Elasticsearch elasticsearch, final Te
         return elasticsearch.main(args, terminal);
     }
 
-    @Override
-    protected boolean shouldConfigureLoggingWithoutConfig() {
-        /*
-         * If we allow logging to be configured without a config before we are ready to read the log4j2.properties file, then we will fail
-         * to detect uses of logging before it is properly configured.
-         */
-        return false;
-    }
-
     @Override
     protected void execute(Terminal terminal, OptionSet options, Environment env) throws UserException {
         if (options.nonOptionArguments().isEmpty() == false) {

core/src/main/java/org/elasticsearch/bootstrap/ElasticsearchUncaughtExceptionHandler.java

@@ -65,37 +65,43 @@ public void uncaughtException(Thread t, Throwable e) {
         }
     }
 
-    // visible for testing
     static boolean isFatalUncaught(Throwable e) {
         return e instanceof Error;
     }
 
-    // visible for testing
     void onFatalUncaught(final String threadName, final Throwable t) {
         final Logger logger = Loggers.getLogger(ElasticsearchUncaughtExceptionHandler.class, loggingPrefixSupplier.get());
         logger.error(
             (org.apache.logging.log4j.util.Supplier<?>)
                 () -> new ParameterizedMessage("fatal error in thread [{}], exiting", threadName), t);
     }
 
-    // visible for testing
     void onNonFatalUncaught(final String threadName, final Throwable t) {
         final Logger logger = Loggers.getLogger(ElasticsearchUncaughtExceptionHandler.class, loggingPrefixSupplier.get());
         logger.warn((org.apache.logging.log4j.util.Supplier<?>)
             () -> new ParameterizedMessage("uncaught exception in thread [{}]", threadName), t);
     }
 
-    // visible for testing
     void halt(int status) {
-        AccessController.doPrivileged(new PrivilegedAction<Void>() {
-            @SuppressForbidden(reason = "halt")
-            @Override
-            public Void run() {
-                // we halt to prevent shutdown hooks from running
-                Runtime.getRuntime().halt(status);
-                return null;
-            }
-        });
+        AccessController.doPrivileged(new PrivilegedHaltAction(status));
+    }
+
+    static class PrivilegedHaltAction implements PrivilegedAction<Void> {
+
+        private final int status;
+
+        private PrivilegedHaltAction(final int status) {
+            this.status = status;
+        }
+
+        @SuppressForbidden(reason = "halt")
+        @Override
+        public Void run() {
+            // we halt to prevent shutdown hooks from running
+            Runtime.getRuntime().halt(status);
+            return null;
+        }
+
     }
 
 }

core/src/main/java/org/elasticsearch/bootstrap/Security.java

@@ -119,7 +119,11 @@ static void configure(Environment environment, boolean filterBadDefaults) throws
         Policy.setPolicy(new ESPolicy(createPermissions(environment), getPluginPermissions(environment), filterBadDefaults));
 
         // enable security manager
-        final String[] classesThatCanExit = new String[]{ElasticsearchUncaughtExceptionHandler.class.getName(), Command.class.getName()};
+        final String[] classesThatCanExit =
+                new String[]{
+                        // SecureSM matches class names as regular expressions so we escape the $ that arises from the nested class name
+                        ElasticsearchUncaughtExceptionHandler.PrivilegedHaltAction.class.getName().replace("$", "\\$"),
+                        Command.class.getName()};
         System.setSecurityManager(new SecureSM(classesThatCanExit));
 
         // do some basic tests