Remove dangerous default executor from HandledTransportAction (elasti…

…c#100162)

Today subclasses of `HandledTransportAction` can specify the executor on
which they run, but the executor is optional and if omitted will use
`DIRECT_EXECUTOR_SERVICE`, which means the action runs on a transport
thread. This is a dangerous default behaviour because it makes it easy
to add new transport actions which implicitly run on a network thread,
which is very hard to pick up in reviews.

This commit makes the executor explicit in all callers, and marks the
dangerous methods for removal.

client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/bulk/TransportNoopBulkAction.java

@@ -17,6 +17,7 @@
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.action.update.UpdateResponse;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
@@ -30,7 +31,7 @@ public class TransportNoopBulkAction extends HandledTransportAction<BulkRequest,
 
     @Inject
     public TransportNoopBulkAction(TransportService transportService, ActionFilters actionFilters) {
-        super(NoopBulkAction.NAME, transportService, actionFilters, BulkRequest::new);
+        super(NoopBulkAction.NAME, transportService, actionFilters, BulkRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
     }
 
     @Override

client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/search/TransportNoopSearchAction.java

@@ -16,6 +16,7 @@
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.search.SearchHits;
 import org.elasticsearch.search.aggregations.InternalAggregations;
@@ -30,7 +31,13 @@
 public class TransportNoopSearchAction extends HandledTransportAction<SearchRequest, SearchResponse> {
     @Inject
     public TransportNoopSearchAction(TransportService transportService, ActionFilters actionFilters) {
-        super(NoopSearchAction.NAME, transportService, actionFilters, (Writeable.Reader<SearchRequest>) SearchRequest::new);
+        super(
+            NoopSearchAction.NAME,
+            transportService,
+            actionFilters,
+            (Writeable.Reader<SearchRequest>) SearchRequest::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
     }
 
     @Override

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/GrokProcessorGetAction.java

@@ -19,6 +19,7 @@
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.grok.GrokBuiltinPatterns;
 import org.elasticsearch.grok.PatternBank;
 import org.elasticsearch.rest.BaseRestHandler;
@@ -139,7 +140,7 @@ public TransportAction(TransportService transportService, ActionFilters actionFi
             PatternBank legacyGrokPatterns,
             PatternBank ecsV1GrokPatterns
         ) {
-            super(NAME, transportService, actionFilters, Request::new);
+            super(NAME, transportService, actionFilters, Request::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
             this.legacyGrokPatterns = legacyGrokPatterns.bank();
             this.sortedLegacyGrokPatterns = new TreeMap<>(this.legacyGrokPatterns);
             this.ecsV1GrokPatterns = ecsV1GrokPatterns.bank();

modules/lang-mustache/src/main/java/org/elasticsearch/script/mustache/TransportMultiSearchTemplateAction.java

@@ -16,6 +16,7 @@
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.client.internal.node.NodeClient;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
@@ -44,7 +45,13 @@ public TransportMultiSearchTemplateAction(
         NodeClient client,
         UsageService usageService
     ) {
-        super(MultiSearchTemplateAction.NAME, transportService, actionFilters, MultiSearchTemplateRequest::new);
+        super(
+            MultiSearchTemplateAction.NAME,
+            transportService,
+            actionFilters,
+            MultiSearchTemplateRequest::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.scriptService = scriptService;
         this.xContentRegistry = xContentRegistry;
         this.client = client;

modules/lang-mustache/src/main/java/org/elasticsearch/script/mustache/TransportSearchTemplateAction.java

@@ -15,6 +15,7 @@
 import org.elasticsearch.client.internal.node.NodeClient;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.common.xcontent.LoggingDeprecationHandler;
 import org.elasticsearch.rest.action.search.RestSearchAction;
 import org.elasticsearch.script.Script;
@@ -54,7 +55,7 @@ public TransportSearchTemplateAction(
         NodeClient client,
         UsageService usageService
     ) {
-        super(SearchTemplateAction.NAME, transportService, actionFilters, SearchTemplateRequest::new);
+        super(SearchTemplateAction.NAME, transportService, actionFilters, SearchTemplateRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.scriptService = scriptService;
         this.xContentRegistry = xContentRegistry;
         this.client = client;

modules/lang-painless/src/main/java/org/elasticsearch/painless/action/PainlessContextAction.java

@@ -20,6 +20,7 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.painless.PainlessScriptEngine;
 import org.elasticsearch.painless.lookup.PainlessLookup;
 import org.elasticsearch.rest.BaseRestHandler;
@@ -142,7 +143,7 @@ public static class TransportAction extends HandledTransportAction<Request, Resp
 
         @Inject
         public TransportAction(TransportService transportService, ActionFilters actionFilters, PainlessScriptEngine painlessScriptEngine) {
-            super(NAME, transportService, actionFilters, (Writeable.Reader<Request>) Request::new);
+            super(NAME, transportService, actionFilters, (Writeable.Reader<Request>) Request::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
             this.painlessScriptEngine = painlessScriptEngine;
         }
 

modules/rank-eval/src/main/java/org/elasticsearch/index/rankeval/TransportRankEvalAction.java

@@ -20,6 +20,7 @@
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.util.Maps;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.common.xcontent.LoggingDeprecationHandler;
 import org.elasticsearch.script.Script;
 import org.elasticsearch.script.ScriptService;
@@ -69,7 +70,7 @@ public TransportRankEvalAction(
         ScriptService scriptService,
         NamedXContentRegistry namedXContentRegistry
     ) {
-        super(RankEvalAction.NAME, transportService, actionFilters, RankEvalRequest::new);
+        super(RankEvalAction.NAME, transportService, actionFilters, RankEvalRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.scriptService = scriptService;
         this.namedXContentRegistry = namedXContentRegistry;
         this.client = client;

modules/reindex/src/main/java/org/elasticsearch/reindex/TransportDeleteByQueryAction.java

@@ -15,6 +15,7 @@
 import org.elasticsearch.client.internal.ParentTaskAssigningClient;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.index.reindex.BulkByScrollResponse;
 import org.elasticsearch.index.reindex.BulkByScrollTask;
 import org.elasticsearch.index.reindex.DeleteByQueryAction;
@@ -40,7 +41,7 @@ public TransportDeleteByQueryAction(
         ScriptService scriptService,
         ClusterService clusterService
     ) {
-        super(DeleteByQueryAction.NAME, transportService, actionFilters, DeleteByQueryRequest::new);
+        super(DeleteByQueryAction.NAME, transportService, actionFilters, DeleteByQueryRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.threadPool = threadPool;
         this.client = client;
         this.scriptService = scriptService;

modules/reindex/src/main/java/org/elasticsearch/reindex/TransportReindexAction.java

@@ -19,6 +19,7 @@
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.index.reindex.BulkByScrollResponse;
 import org.elasticsearch.index.reindex.BulkByScrollTask;
 import org.elasticsearch.index.reindex.ReindexAction;
@@ -82,7 +83,7 @@ protected TransportReindexAction(
         TransportService transportService,
         ReindexSslConfig sslConfig
     ) {
-        super(name, transportService, actionFilters, ReindexRequest::new);
+        super(name, transportService, actionFilters, ReindexRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.client = client;
         this.reindexValidator = new ReindexValidator(settings, clusterService, indexNameExpressionResolver, autoCreateIndex);
         this.reindexer = new Reindexer(clusterService, client, threadPool, scriptService, sslConfig);

modules/reindex/src/main/java/org/elasticsearch/reindex/TransportUpdateByQueryAction.java

@@ -18,6 +18,7 @@
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.index.reindex.BulkByScrollResponse;
 import org.elasticsearch.index.reindex.BulkByScrollTask;
 import org.elasticsearch.index.reindex.ScrollableHitSource;
@@ -53,7 +54,7 @@ public TransportUpdateByQueryAction(
         ScriptService scriptService,
         ClusterService clusterService
     ) {
-        super(UpdateByQueryAction.NAME, transportService, actionFilters, UpdateByQueryRequest::new);
+        super(UpdateByQueryAction.NAME, transportService, actionFilters, UpdateByQueryRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.threadPool = threadPool;
         this.client = client;
         this.scriptService = scriptService;

modules/rest-root/src/main/java/org/elasticsearch/rest/root/TransportMainAction.java

@@ -16,6 +16,7 @@
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.node.Node;
 import org.elasticsearch.tasks.Task;
@@ -33,7 +34,7 @@ public TransportMainAction(
         ActionFilters actionFilters,
         ClusterService clusterService
     ) {
-        super(MainAction.NAME, transportService, actionFilters, MainRequest::new);
+        super(MainAction.NAME, transportService, actionFilters, MainRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.nodeName = Node.NODE_NAME_SETTING.get(settings);
         this.clusterService = clusterService;
     }

server/src/internalClusterTest/java/org/elasticsearch/action/admin/cluster/tasks/ListTasksIT.java

@@ -18,6 +18,7 @@
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.plugins.ActionPlugin;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.plugins.PluginsService;
@@ -185,7 +186,7 @@ public TestTransportAction(
             PluginsService pluginsService,
             ThreadPool threadPool
         ) {
-            super(NAME, transportService, actionFilters, in -> new TestRequest());
+            super(NAME, transportService, actionFilters, in -> new TestRequest(), EsExecutors.DIRECT_EXECUTOR_SERVICE);
             testPlugin = pluginsService.filterPlugins(TestPlugin.class).get(0);
             this.threadPool = threadPool;
         }

server/src/main/java/org/elasticsearch/action/admin/cluster/coordination/MasterHistoryAction.java

@@ -20,6 +20,7 @@
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 
@@ -129,7 +130,7 @@ public static class TransportAction extends HandledTransportAction<Request, Resp
 
         @Inject
         public TransportAction(TransportService transportService, ActionFilters actionFilters, MasterHistoryService masterHistoryService) {
-            super(MasterHistoryAction.NAME, transportService, actionFilters, MasterHistoryAction.Request::new);
+            super(MasterHistoryAction.NAME, transportService, actionFilters, Request::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
             this.masterHistoryService = masterHistoryService;
         }
 

server/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/get/TransportGetTaskAction.java

@@ -76,7 +76,7 @@ public TransportGetTaskAction(
         Client client,
         NamedXContentRegistry xContentRegistry
     ) {
-        super(GetTaskAction.NAME, transportService, actionFilters, GetTaskRequest::new);
+        super(GetTaskAction.NAME, transportService, actionFilters, GetTaskRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.threadPool = threadPool;
         this.clusterService = clusterService;
         this.transportService = transportService;

server/src/main/java/org/elasticsearch/action/admin/cluster/remote/RemoteClusterNodesAction.java

@@ -25,6 +25,7 @@
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.RemoteClusterServerInfo;
@@ -98,7 +99,7 @@ public static class TransportAction extends HandledTransportAction<Request, Resp
 
         @Inject
         public TransportAction(TransportService transportService, ActionFilters actionFilters) {
-            super(RemoteClusterNodesAction.NAME, transportService, actionFilters, Request::new);
+            super(RemoteClusterNodesAction.NAME, transportService, actionFilters, Request::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
             this.transportService = transportService;
         }
 

server/src/main/java/org/elasticsearch/action/admin/cluster/remote/TransportRemoteInfoAction.java

@@ -14,6 +14,7 @@
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.cluster.node.DiscoveryNodeRole;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.RemoteClusterService;
 import org.elasticsearch.transport.TransportService;
@@ -30,7 +31,7 @@ public TransportRemoteInfoAction(
         ActionFilters actionFilters,
         SearchTransportService searchTransportService
     ) {
-        super(RemoteInfoAction.NAME, transportService, actionFilters, RemoteInfoRequest::new);
+        super(RemoteInfoAction.NAME, transportService, actionFilters, RemoteInfoRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.remoteClusterService = searchTransportService.getRemoteClusterService();
     }
 

server/src/main/java/org/elasticsearch/action/admin/cluster/storedscripts/TransportGetScriptContextAction.java

@@ -11,6 +11,7 @@
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.script.ScriptContextInfo;
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.tasks.Task;
@@ -24,7 +25,13 @@ public class TransportGetScriptContextAction extends HandledTransportAction<GetS
 
     @Inject
     public TransportGetScriptContextAction(TransportService transportService, ActionFilters actionFilters, ScriptService scriptService) {
-        super(GetScriptContextAction.NAME, transportService, actionFilters, GetScriptContextRequest::new);
+        super(
+            GetScriptContextAction.NAME,
+            transportService,
+            actionFilters,
+            GetScriptContextRequest::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.scriptService = scriptService;
     }
 

server/src/main/java/org/elasticsearch/action/admin/cluster/storedscripts/TransportGetScriptLanguageAction.java

@@ -12,6 +12,7 @@
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
@@ -21,7 +22,13 @@ public class TransportGetScriptLanguageAction extends HandledTransportAction<Get
 
     @Inject
     public TransportGetScriptLanguageAction(TransportService transportService, ActionFilters actionFilters, ScriptService scriptService) {
-        super(GetScriptLanguageAction.NAME, transportService, actionFilters, GetScriptLanguageRequest::new);
+        super(
+            GetScriptLanguageAction.NAME,
+            transportService,
+            actionFilters,
+            GetScriptLanguageRequest::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.scriptService = scriptService;
     }
 

server/src/main/java/org/elasticsearch/action/admin/indices/dangling/import_index/TransportImportDanglingIndexAction.java

@@ -22,6 +22,7 @@
 import org.elasticsearch.client.internal.node.NodeClient;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.gateway.LocalAllocateDangledIndices;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
@@ -49,7 +50,13 @@ public TransportImportDanglingIndexAction(
         LocalAllocateDangledIndices danglingIndexAllocator,
         NodeClient nodeClient
     ) {
-        super(ImportDanglingIndexAction.NAME, transportService, actionFilters, ImportDanglingIndexRequest::new);
+        super(
+            ImportDanglingIndexAction.NAME,
+            transportService,
+            actionFilters,
+            ImportDanglingIndexRequest::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.danglingIndexAllocator = danglingIndexAllocator;
         this.nodeClient = nodeClient;
     }

server/src/main/java/org/elasticsearch/action/admin/indices/mapping/get/TransportGetFieldMappingsAction.java

@@ -16,6 +16,7 @@
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 
@@ -41,7 +42,13 @@ public TransportGetFieldMappingsAction(
         IndexNameExpressionResolver indexNameExpressionResolver,
         NodeClient client
     ) {
-        super(GetFieldMappingsAction.NAME, transportService, actionFilters, GetFieldMappingsRequest::new);
+        super(
+            GetFieldMappingsAction.NAME,
+            transportService,
+            actionFilters,
+            GetFieldMappingsRequest::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.clusterService = clusterService;
         this.indexNameExpressionResolver = indexNameExpressionResolver;
         this.client = client;

server/src/main/java/org/elasticsearch/action/admin/indices/resolve/ResolveIndexAction.java

@@ -454,7 +454,7 @@ public TransportAction(
             ActionFilters actionFilters,
             IndexNameExpressionResolver indexNameExpressionResolver
         ) {
-            super(NAME, transportService, actionFilters, Request::new);
+            super(NAME, transportService, actionFilters, Request::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
             this.threadPool = threadPool;
             this.clusterService = clusterService;
             this.remoteClusterService = transportService.getRemoteClusterService();

server/src/main/java/org/elasticsearch/action/bulk/TransportSingleItemBulkWriteAction.java

@@ -17,6 +17,7 @@
 import org.elasticsearch.action.support.replication.ReplicatedWriteRequest;
 import org.elasticsearch.action.support.replication.ReplicationResponse;
 import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 
@@ -38,7 +39,7 @@ protected TransportSingleItemBulkWriteAction(
         Writeable.Reader<Request> requestReader,
         TransportBulkAction bulkAction
     ) {
-        super(actionName, transportService, actionFilters, requestReader);
+        super(actionName, transportService, actionFilters, requestReader, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.bulkAction = bulkAction;
     }