Remove dangerous default executor from HandledTransportAction (#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.ActionFilters;
 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.ActionFilters;
 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 @@ import java.util.Collections;
 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.client.internal.node.NodeClient;
 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 class GrokProcessorGetAction extends ActionType<GrokProcessorGetAction.Re
             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.ActionFilters;
 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 class TransportMultiSearchTemplateAction extends HandledTransportAction<M
         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.action.support.HandledTransportAction;
 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 class TransportSearchTemplateAction extends HandledTransportAction<Search
         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.inject.Inject;
 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 class PainlessContextAction extends ActionType<PainlessContextAction.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.client.internal.Client;
 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 class TransportRankEvalAction extends HandledTransportAction<RankEvalRequ
         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.Client;
 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 class TransportDeleteByQueryAction extends HandledTransportAction<DeleteB
         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.inject.Inject;
 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 @@ public class TransportReindexAction extends HandledTransportAction<ReindexReques
         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.client.internal.ParentTaskAssigningClient;
 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 class TransportUpdateByQueryAction extends HandledTransportAction<UpdateB
         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.ClusterState;
 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 class TransportMainAction extends HandledTransportAction<MainRequest, Mai
         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.ActionFilters;
 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 class ListTasksIT extends ESSingleNodeTestCase {
             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.cluster.node.DiscoveryNode;
 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 class MasterHistoryAction extends ActionType<MasterHistoryAction.Response
 
         @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 class TransportGetTaskAction extends HandledTransportAction<GetTaskReques
         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.cluster.node.DiscoveryNode;
 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 class RemoteClusterNodesAction extends ActionType<RemoteClusterNodesActio
 
         @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.ActionFilters;
 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 final class TransportRemoteInfoAction extends HandledTransportAction<Remo
         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.ActionListener;
 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.ActionListener;
 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.action.support.master.AcknowledgedResponse;
 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 class TransportImportDanglingIndexAction extends HandledTransportAction<I
         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.ClusterState;
 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 class TransportGetFieldMappingsAction extends HandledTransportAction<GetF
         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 class ResolveIndexAction extends ActionType<ResolveIndexAction.Response>
             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.WriteResponse;
 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 @@ public abstract class TransportSingleItemBulkWriteAction<
         Writeable.Reader<Request> requestReader,
         TransportBulkAction bulkAction
     ) {
-        super(actionName, transportService, actionFilters, requestReader);
+        super(actionName, transportService, actionFilters, requestReader, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.bulkAction = bulkAction;
     }
 

server/src/main/java/org/elasticsearch/action/get/TransportMultiGetAction.java

@@ -20,6 +20,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.AtomicArray;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.tasks.Task;
@@ -44,7 +45,7 @@ public class TransportMultiGetAction extends HandledTransportAction<MultiGetRequ
         IndexNameExpressionResolver resolver,
         IndicesService indicesService
     ) {
-        super(MultiGetAction.NAME, transportService, actionFilters, MultiGetRequest::new);
+        super(MultiGetAction.NAME, transportService, actionFilters, MultiGetRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.clusterService = clusterService;
         this.client = client;
         this.indexNameExpressionResolver = resolver;

server/src/main/java/org/elasticsearch/action/ingest/SimulatePipelineTransportAction.java

@@ -19,6 +19,7 @@ import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.common.Randomness;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.ingest.IngestService;
@@ -58,7 +59,13 @@ public class SimulatePipelineTransportAction extends HandledTransportAction<Simu
         ActionFilters actionFilters,
         IngestService ingestService
     ) {
-        super(SimulatePipelineAction.NAME, transportService, actionFilters, SimulatePipelineRequest::new);
+        super(
+            SimulatePipelineAction.NAME,
+            transportService,
+            actionFilters,
+            SimulatePipelineRequest::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.ingestService = ingestService;
         this.executionService = new SimulateExecutionService(threadPool);
         this.transportService = transportService;

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

@@ -13,6 +13,7 @@ import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
 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;
 
@@ -28,7 +29,7 @@ public class TransportClearScrollAction extends HandledTransportAction<ClearScro
         ActionFilters actionFilters,
         SearchTransportService searchTransportService
     ) {
-        super(ClearScrollAction.NAME, transportService, actionFilters, ClearScrollRequest::new);
+        super(ClearScrollAction.NAME, transportService, actionFilters, ClearScrollRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.clusterService = clusterService;
         this.searchTransportService = searchTransportService;
     }

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

@@ -14,6 +14,7 @@ import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 
@@ -33,7 +34,13 @@ public class TransportClosePointInTimeAction extends HandledTransportAction<Clos
         SearchTransportService searchTransportService,
         NamedWriteableRegistry namedWriteableRegistry
     ) {
-        super(ClosePointInTimeAction.NAME, transportService, actionFilters, ClosePointInTimeRequest::new);
+        super(
+            ClosePointInTimeAction.NAME,
+            transportService,
+            actionFilters,
+            ClosePointInTimeRequest::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.clusterService = clusterService;
         this.searchTransportService = searchTransportService;
         this.namedWriteableRegistry = namedWriteableRegistry;

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

@@ -47,7 +47,13 @@ public class TransportMultiSearchAction extends HandledTransportAction<MultiSear
         ActionFilters actionFilters,
         NodeClient client
     ) {
-        super(MultiSearchAction.NAME, transportService, actionFilters, (Writeable.Reader<MultiSearchRequest>) MultiSearchRequest::new);
+        super(
+            MultiSearchAction.NAME,
+            transportService,
+            actionFilters,
+            (Writeable.Reader<MultiSearchRequest>) MultiSearchRequest::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.threadPool = threadPool;
         this.clusterService = clusterService;
         this.allocatedProcessors = EsExecutors.allocatedProcessors(settings);
@@ -64,7 +70,13 @@ public class TransportMultiSearchAction extends HandledTransportAction<MultiSear
         LongSupplier relativeTimeProvider,
         NodeClient client
     ) {
-        super(MultiSearchAction.NAME, transportService, actionFilters, (Writeable.Reader<MultiSearchRequest>) MultiSearchRequest::new);
+        super(
+            MultiSearchAction.NAME,
+            transportService,
+            actionFilters,
+            (Writeable.Reader<MultiSearchRequest>) MultiSearchRequest::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.threadPool = threadPool;
         this.clusterService = clusterService;
         this.allocatedProcessors = allocatedProcessors;

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

@@ -62,7 +62,13 @@ public class TransportOpenPointInTimeAction extends HandledTransportAction<OpenP
         TransportSearchAction transportSearchAction,
         SearchTransportService searchTransportService
     ) {
-        super(OpenPointInTimeAction.NAME, transportService, actionFilters, OpenPointInTimeRequest::new);
+        super(
+            OpenPointInTimeAction.NAME,
+            transportService,
+            actionFilters,
+            OpenPointInTimeRequest::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.transportService = transportService;
         this.transportSearchAction = transportSearchAction;
         this.searchService = searchService;

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

@@ -47,6 +47,7 @@ import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.util.CollectionUtils;
 import org.elasticsearch.common.util.Maps;
 import org.elasticsearch.common.util.concurrent.CountDown;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.Index;
@@ -158,7 +159,13 @@ public class TransportSearchAction extends HandledTransportAction<SearchRequest,
         NamedWriteableRegistry namedWriteableRegistry,
         ExecutorSelector executorSelector
     ) {
-        super(SearchAction.NAME, transportService, actionFilters, (Writeable.Reader<SearchRequest>) SearchRequest::new);
+        super(
+            SearchAction.NAME,
+            transportService,
+            actionFilters,
+            (Writeable.Reader<SearchRequest>) SearchRequest::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.threadPool = threadPool;
         this.circuitBreaker = circuitBreakerService.getBreaker(CircuitBreaker.REQUEST);
         this.searchPhaseController = searchPhaseController;

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

@@ -13,6 +13,7 @@ import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
 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;
 
@@ -32,7 +33,7 @@ public class TransportSearchScrollAction extends HandledTransportAction<SearchSc
         ActionFilters actionFilters,
         SearchTransportService searchTransportService
     ) {
-        super(SearchScrollAction.NAME, transportService, actionFilters, SearchScrollRequest::new);
+        super(SearchScrollAction.NAME, transportService, actionFilters, SearchScrollRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.clusterService = clusterService;
         this.searchTransportService = searchTransportService;
     }

server/src/main/java/org/elasticsearch/action/support/HandledTransportAction.java

@@ -11,20 +11,21 @@ import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
-import org.elasticsearch.tasks.Task;
-import org.elasticsearch.transport.TransportChannel;
-import org.elasticsearch.transport.TransportRequestHandler;
 import org.elasticsearch.transport.TransportService;
 
 import java.util.concurrent.Executor;
 
 /**
- * A TransportAction that self registers a handler into the transport service
+ * A {@link TransportAction} which, on creation, registers a handler for its own {@link #actionName} with the transport service.
  */
 public abstract class HandledTransportAction<Request extends ActionRequest, Response extends ActionResponse> extends TransportAction<
     Request,
     Response> {
 
+    /**
+     * @deprecated always specify an executor
+     */
+    @Deprecated(forRemoval = true)
     protected HandledTransportAction(
         String actionName,
         TransportService transportService,
@@ -44,6 +45,10 @@ public abstract class HandledTransportAction<Request extends ActionRequest, Resp
         this(actionName, true, transportService, actionFilters, requestReader, executor);
     }
 
+    /**
+     * @deprecated always specify an executor
+     */
+    @Deprecated(forRemoval = true)
     protected HandledTransportAction(
         String actionName,
         boolean canTripCircuitBreaker,
@@ -63,15 +68,13 @@ public abstract class HandledTransportAction<Request extends ActionRequest, Resp
         Executor executor
     ) {
         super(actionName, actionFilters, transportService.getTaskManager());
-        transportService.registerRequestHandler(actionName, executor, false, canTripCircuitBreaker, requestReader, new TransportHandler());
+        transportService.registerRequestHandler(
+            actionName,
+            executor,
+            false,
+            canTripCircuitBreaker,
+            requestReader,
+            (request, channel, task) -> execute(task, request, new ChannelActionListener<>(channel))
+        );
     }
-
-    class TransportHandler implements TransportRequestHandler<Request> {
-        @Override
-        public final void messageReceived(final Request request, final TransportChannel channel, Task task) {
-            // We already got the task created on the network layer - no need to create it again on the transport layer
-            execute(task, request, new ChannelActionListener<>(channel));
-        }
-    }
-
 }

server/src/main/java/org/elasticsearch/action/support/ReservedStateAwareHandledTransportAction.java

@@ -13,6 +13,7 @@ import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.reservedstate.ActionWithReservedState;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
@@ -34,7 +35,7 @@ public abstract class ReservedStateAwareHandledTransportAction<Request extends A
         ActionFilters actionFilters,
         Writeable.Reader<Request> requestReader
     ) {
-        super(actionName, transportService, actionFilters, requestReader);
+        super(actionName, transportService, actionFilters, requestReader, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.clusterService = clusterService;
     }
 

server/src/main/java/org/elasticsearch/action/support/broadcast/unpromotable/TransportBroadcastUnpromotableAction.java

@@ -22,6 +22,7 @@ import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.core.Strings;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportChannel;
@@ -53,7 +54,7 @@ public abstract class TransportBroadcastUnpromotableAction<Request extends Broad
         Writeable.Reader<Request> requestReader,
         Executor executor
     ) {
-        super(actionName, transportService, actionFilters, requestReader);
+        super(actionName, transportService, actionFilters, requestReader, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.clusterService = clusterService;
         this.shardStateAction = shardStateAction;
         this.transportService = transportService;

server/src/main/java/org/elasticsearch/action/support/master/TransportMasterNodeAction.java

@@ -27,6 +27,7 @@ import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.cluster.service.MasterService;
 import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.discovery.MasterNotDiscoveredException;
 import org.elasticsearch.gateway.GatewayService;
@@ -104,7 +105,7 @@ public abstract class TransportMasterNodeAction<Request extends MasterNodeReques
         Writeable.Reader<Response> response,
         Executor executor
     ) {
-        super(actionName, canTripCircuitBreaker, transportService, actionFilters, request);
+        super(actionName, canTripCircuitBreaker, transportService, actionFilters, request, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.transportService = transportService;
         this.clusterService = clusterService;
         this.threadPool = threadPool;

server/src/main/java/org/elasticsearch/action/support/replication/TransportBroadcastReplicationAction.java

@@ -26,6 +26,7 @@ import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.routing.IndexRoutingTable;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.core.CheckedConsumer;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.tasks.Task;
@@ -65,7 +66,7 @@ public abstract class TransportBroadcastReplicationAction<
         ActionType<ShardResponse> replicatedBroadcastShardAction,
         Executor executor
     ) {
-        super(name, transportService, actionFilters, requestReader);
+        super(name, transportService, actionFilters, requestReader, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.client = client;
         this.replicatedBroadcastShardAction = replicatedBroadcastShardAction;
         this.clusterService = clusterService;

server/src/main/java/org/elasticsearch/action/support/single/instance/TransportInstanceSingleOperationAction.java

@@ -66,7 +66,7 @@ public abstract class TransportInstanceSingleOperationAction<
         IndexNameExpressionResolver indexNameExpressionResolver,
         Writeable.Reader<Request> request
     ) {
-        super(actionName, transportService, actionFilters, request);
+        super(actionName, transportService, actionFilters, request, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.threadPool = threadPool;
         this.clusterService = clusterService;
         this.transportService = transportService;

server/src/main/java/org/elasticsearch/action/synonyms/TransportDeleteSynonymRuleAction.java

@@ -13,6 +13,7 @@ import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.synonyms.SynonymsManagementAPIService;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
@@ -23,7 +24,13 @@ public class TransportDeleteSynonymRuleAction extends HandledTransportAction<Del
 
     @Inject
     public TransportDeleteSynonymRuleAction(TransportService transportService, ActionFilters actionFilters, Client client) {
-        super(DeleteSynonymRuleAction.NAME, transportService, actionFilters, DeleteSynonymRuleAction.Request::new);
+        super(
+            DeleteSynonymRuleAction.NAME,
+            transportService,
+            actionFilters,
+            DeleteSynonymRuleAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
 
         this.synonymsManagementAPIService = new SynonymsManagementAPIService(client);
     }

server/src/main/java/org/elasticsearch/action/synonyms/TransportDeleteSynonymsAction.java

@@ -14,6 +14,7 @@ import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.synonyms.SynonymsManagementAPIService;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
@@ -24,7 +25,13 @@ public class TransportDeleteSynonymsAction extends HandledTransportAction<Delete
 
     @Inject
     public TransportDeleteSynonymsAction(TransportService transportService, ActionFilters actionFilters, Client client) {
-        super(DeleteSynonymsAction.NAME, transportService, actionFilters, DeleteSynonymsAction.Request::new);
+        super(
+            DeleteSynonymsAction.NAME,
+            transportService,
+            actionFilters,
+            DeleteSynonymsAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
 
         this.synonymsManagementAPIService = new SynonymsManagementAPIService(client);
     }

server/src/main/java/org/elasticsearch/action/synonyms/TransportGetSynonymRuleAction.java

@@ -13,6 +13,7 @@ import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.synonyms.SynonymsManagementAPIService;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
@@ -23,7 +24,13 @@ public class TransportGetSynonymRuleAction extends HandledTransportAction<GetSyn
 
     @Inject
     public TransportGetSynonymRuleAction(TransportService transportService, ActionFilters actionFilters, Client client) {
-        super(GetSynonymRuleAction.NAME, transportService, actionFilters, GetSynonymRuleAction.Request::new);
+        super(
+            GetSynonymRuleAction.NAME,
+            transportService,
+            actionFilters,
+            GetSynonymRuleAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
 
         this.synonymsManagementAPIService = new SynonymsManagementAPIService(client);
     }

server/src/main/java/org/elasticsearch/action/synonyms/TransportGetSynonymsAction.java

@@ -13,6 +13,7 @@ import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.synonyms.SynonymsManagementAPIService;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
@@ -23,7 +24,7 @@ public class TransportGetSynonymsAction extends HandledTransportAction<GetSynony
 
     @Inject
     public TransportGetSynonymsAction(TransportService transportService, ActionFilters actionFilters, Client client) {
-        super(GetSynonymsAction.NAME, transportService, actionFilters, GetSynonymsAction.Request::new);
+        super(GetSynonymsAction.NAME, transportService, actionFilters, GetSynonymsAction.Request::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
 
         this.synonymsManagementAPIService = new SynonymsManagementAPIService(client);
     }

server/src/main/java/org/elasticsearch/action/synonyms/TransportGetSynonymsSetsAction.java

@@ -13,6 +13,7 @@ import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.synonyms.SynonymsManagementAPIService;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
@@ -23,7 +24,13 @@ public class TransportGetSynonymsSetsAction extends HandledTransportAction<GetSy
 
     @Inject
     public TransportGetSynonymsSetsAction(TransportService transportService, ActionFilters actionFilters, Client client) {
-        super(GetSynonymsSetsAction.NAME, transportService, actionFilters, GetSynonymsSetsAction.Request::new);
+        super(
+            GetSynonymsSetsAction.NAME,
+            transportService,
+            actionFilters,
+            GetSynonymsSetsAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
 
         this.synonymsManagementAPIService = new SynonymsManagementAPIService(client);
     }

server/src/main/java/org/elasticsearch/action/synonyms/TransportPutSynonymRuleAction.java

@@ -13,6 +13,7 @@ import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.synonyms.SynonymsManagementAPIService;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
@@ -23,7 +24,13 @@ public class TransportPutSynonymRuleAction extends HandledTransportAction<PutSyn
 
     @Inject
     public TransportPutSynonymRuleAction(TransportService transportService, ActionFilters actionFilters, Client client) {
-        super(PutSynonymRuleAction.NAME, transportService, actionFilters, PutSynonymRuleAction.Request::new);
+        super(
+            PutSynonymRuleAction.NAME,
+            transportService,
+            actionFilters,
+            PutSynonymRuleAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
 
         this.synonymsManagementAPIService = new SynonymsManagementAPIService(client);
     }

server/src/main/java/org/elasticsearch/action/synonyms/TransportPutSynonymsAction.java

@@ -13,6 +13,7 @@ import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.synonyms.SynonymsManagementAPIService;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
@@ -23,7 +24,7 @@ public class TransportPutSynonymsAction extends HandledTransportAction<PutSynony
 
     @Inject
     public TransportPutSynonymsAction(TransportService transportService, ActionFilters actionFilters, Client client) {
-        super(PutSynonymsAction.NAME, transportService, actionFilters, PutSynonymsAction.Request::new);
+        super(PutSynonymsAction.NAME, transportService, actionFilters, PutSynonymsAction.Request::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
 
         this.synonymsManagementAPIService = new SynonymsManagementAPIService(client);
     }

server/src/main/java/org/elasticsearch/action/termvectors/TransportMultiTermVectorsAction.java

@@ -19,6 +19,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.AtomicArray;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
@@ -41,7 +42,13 @@ public class TransportMultiTermVectorsAction extends HandledTransportAction<Mult
         ActionFilters actionFilters,
         IndexNameExpressionResolver indexNameExpressionResolver
     ) {
-        super(MultiTermVectorsAction.NAME, transportService, actionFilters, MultiTermVectorsRequest::new);
+        super(
+            MultiTermVectorsAction.NAME,
+            transportService,
+            actionFilters,
+            MultiTermVectorsRequest::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.clusterService = clusterService;
         this.client = client;
         this.indexNameExpressionResolver = indexNameExpressionResolver;

server/src/main/java/org/elasticsearch/health/node/action/TransportHealthNodeAction.java

@@ -20,6 +20,7 @@ import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.health.node.selection.HealthNode;
 import org.elasticsearch.tasks.CancellableTask;
@@ -73,7 +74,7 @@ public abstract class TransportHealthNodeAction<Request extends HealthNodeReques
         Writeable.Reader<Response> response,
         Executor executor
     ) {
-        super(actionName, true, transportService, actionFilters, request);
+        super(actionName, true, transportService, actionFilters, request, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.transportService = transportService;
         this.clusterService = clusterService;
         this.threadPool = threadPool;

x-pack/plugin/async-search/src/main/java/org/elasticsearch/xpack/search/TransportGetAsyncSearchAction.java

@@ -43,7 +43,7 @@ public class TransportGetAsyncSearchAction extends HandledTransportAction<GetAsy
         ThreadPool threadPool,
         BigArrays bigArrays
     ) {
-        super(GetAsyncSearchAction.NAME, transportService, actionFilters, GetAsyncResultRequest::new);
+        super(GetAsyncSearchAction.NAME, transportService, actionFilters, GetAsyncResultRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.transportService = transportService;
         this.resultsService = createResultsService(transportService, clusterService, registry, client, threadPool, bigArrays);
     }

x-pack/plugin/async-search/src/main/java/org/elasticsearch/xpack/search/TransportGetAsyncStatusAction.java

@@ -47,7 +47,7 @@ public class TransportGetAsyncStatusAction extends HandledTransportAction<GetAsy
         ThreadPool threadPool,
         BigArrays bigArrays
     ) {
-        super(GetAsyncStatusAction.NAME, transportService, actionFilters, GetAsyncStatusRequest::new);
+        super(GetAsyncStatusAction.NAME, transportService, actionFilters, GetAsyncStatusRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.transportService = transportService;
         this.clusterService = clusterService;
         this.store = new AsyncTaskIndexService<>(

x-pack/plugin/async-search/src/main/java/org/elasticsearch/xpack/search/TransportSubmitAsyncSearchAction.java

@@ -20,6 +20,7 @@ import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.search.SearchService;
@@ -62,7 +63,13 @@ public class TransportSubmitAsyncSearchAction extends HandledTransportAction<Sub
         TransportSearchAction searchAction,
         BigArrays bigArrays
     ) {
-        super(SubmitAsyncSearchAction.NAME, transportService, actionFilters, SubmitAsyncSearchRequest::new);
+        super(
+            SubmitAsyncSearchAction.NAME,
+            transportService,
+            actionFilters,
+            SubmitAsyncSearchRequest::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.clusterService = clusterService;
         this.nodeClient = nodeClient;
         this.requestToAggReduceContextBuilder = (task, request) -> searchService.aggReduceContextBuilder(

x-pack/plugin/core/src/main/java/org/elasticsearch/license/TransportGetFeatureUsageAction.java

@@ -12,6 +12,7 @@ import org.elasticsearch.action.ActionType;
 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.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 
@@ -33,7 +34,7 @@ public class TransportGetFeatureUsageAction extends HandledTransportAction<GetFe
 
     @Inject
     public TransportGetFeatureUsageAction(TransportService transportService, ActionFilters actionFilters, XPackLicenseState licenseState) {
-        super(TYPE.name(), transportService, actionFilters, GetFeatureUsageRequest::new);
+        super(TYPE.name(), transportService, actionFilters, GetFeatureUsageRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.licenseState = licenseState;
     }
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/action/AbstractTransportGetResourcesAction.java

@@ -18,6 +18,7 @@ import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.regex.Regex;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.common.xcontent.LoggingDeprecationHandler;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.index.query.BoolQueryBuilder;
@@ -69,7 +70,7 @@ public abstract class AbstractTransportGetResourcesAction<
         Client client,
         NamedXContentRegistry xContentRegistry
     ) {
-        super(actionName, transportService, actionFilters, request);
+        super(actionName, transportService, actionFilters, request, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.client = Objects.requireNonNull(client);
         this.xContentRegistry = Objects.requireNonNull(xContentRegistry);
     }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/action/TransportXPackInfoAction.java

@@ -11,6 +11,7 @@ import org.elasticsearch.action.support.ActionFilters;
 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.license.License;
 import org.elasticsearch.license.LicenseService;
 import org.elasticsearch.license.LicenseUtils;
@@ -40,7 +41,7 @@ public class TransportXPackInfoAction extends HandledTransportAction<XPackInfoRe
         LicenseService licenseService,
         NodeClient client
     ) {
-        super(XPackInfoAction.NAME, transportService, actionFilters, XPackInfoRequest::new);
+        super(XPackInfoAction.NAME, transportService, actionFilters, XPackInfoRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.licenseService = licenseService;
         this.client = client;
         this.infoActions = infoActions();

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/action/XPackInfoFeatureTransportAction.java

@@ -9,6 +9,7 @@ package org.elasticsearch.xpack.core.action;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.protocol.xpack.XPackInfoRequest;
 import org.elasticsearch.protocol.xpack.XPackInfoResponse.FeatureSetsInfo.FeatureSet;
 import org.elasticsearch.tasks.Task;
@@ -23,7 +24,7 @@ import org.elasticsearch.transport.TransportService;
 public abstract class XPackInfoFeatureTransportAction extends HandledTransportAction<XPackInfoRequest, XPackInfoFeatureResponse> {
 
     public XPackInfoFeatureTransportAction(String name, TransportService transportService, ActionFilters actionFilters) {
-        super(name, transportService, actionFilters, XPackInfoRequest::new);
+        super(name, transportService, actionFilters, XPackInfoRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
     }
 
     protected abstract String name();

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/async/TransportDeleteAsyncResultAction.java

@@ -40,7 +40,13 @@ public class TransportDeleteAsyncResultAction extends HandledTransportAction<Del
         ThreadPool threadPool,
         BigArrays bigArrays
     ) {
-        super(DeleteAsyncResultAction.NAME, transportService, actionFilters, DeleteAsyncResultRequest::new);
+        super(
+            DeleteAsyncResultAction.NAME,
+            transportService,
+            actionFilters,
+            DeleteAsyncResultRequest::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.transportService = transportService;
         this.clusterService = clusterService;
         AsyncTaskIndexService<?> store = new AsyncTaskIndexService<>(

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ssl/action/TransportGetCertificateInfoAction.java

@@ -10,6 +10,7 @@ import org.elasticsearch.action.ActionListener;
 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.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.core.ssl.SSLService;
@@ -27,7 +28,13 @@ public class TransportGetCertificateInfoAction extends HandledTransportAction<
 
     @Inject
     public TransportGetCertificateInfoAction(TransportService transportService, ActionFilters actionFilters, SSLService sslService) {
-        super(GetCertificateInfoAction.NAME, transportService, actionFilters, GetCertificateInfoAction.Request::new);
+        super(
+            GetCertificateInfoAction.NAME,
+            transportService,
+            actionFilters,
+            GetCertificateInfoAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.sslService = sslService;
     }
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/termsenum/action/TransportTermsEnumAction.java

@@ -118,7 +118,7 @@ public class TransportTermsEnumAction extends HandledTransportAction<TermsEnumRe
         Settings settings,
         IndexNameExpressionResolver indexNameExpressionResolver
     ) {
-        super(TermsEnumAction.NAME, transportService, actionFilters, TermsEnumRequest::new);
+        super(TermsEnumAction.NAME, transportService, actionFilters, TermsEnumRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
 
         this.clusterService = clusterService;
         this.searchService = searchService;

x-pack/plugin/enrich/src/main/java/org/elasticsearch/xpack/enrich/action/EnrichCoordinatorProxyAction.java

@@ -19,6 +19,7 @@ import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.client.internal.ElasticsearchClient;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.tasks.Task;
@@ -60,7 +61,7 @@ public class EnrichCoordinatorProxyAction extends ActionType<SearchResponse> {
 
         @Inject
         public TransportAction(TransportService transportService, ActionFilters actionFilters, Coordinator coordinator) {
-            super(NAME, transportService, actionFilters, SearchRequest::new);
+            super(NAME, transportService, actionFilters, SearchRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
             this.coordinator = coordinator;
         }
 

x-pack/plugin/enrich/src/main/java/org/elasticsearch/xpack/enrich/action/InternalExecutePolicyAction.java

@@ -21,6 +21,7 @@ import org.elasticsearch.common.Randomness;
 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.tasks.TaskAwareRequest;
 import org.elasticsearch.tasks.TaskCancelledException;
@@ -116,7 +117,7 @@ public class InternalExecutePolicyAction extends ActionType<Response> {
             ClusterService clusterService,
             EnrichPolicyExecutor policyExecutor
         ) {
-            super(NAME, transportService, actionFilters, Request::new);
+            super(NAME, transportService, actionFilters, Request::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
             this.clusterService = clusterService;
             this.transportService = transportService;
             this.policyExecutor = policyExecutor;

x-pack/plugin/ent-search/src/main/java/org/elasticsearch/xpack/application/analytics/action/TransportPostAnalyticsEventAction.java

@@ -11,6 +11,7 @@ import org.elasticsearch.action.ActionListener;
 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.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.application.analytics.AnalyticsEventIngestService;
@@ -32,7 +33,13 @@ public class TransportPostAnalyticsEventAction extends HandledTransportAction<
         ActionFilters actionFilters,
         AnalyticsEventIngestService eventEmitterService
     ) {
-        super(PostAnalyticsEventAction.NAME, transportService, actionFilters, PostAnalyticsEventAction.Request::new);
+        super(
+            PostAnalyticsEventAction.NAME,
+            transportService,
+            actionFilters,
+            PostAnalyticsEventAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.eventEmitterService = eventEmitterService;
     }
 

x-pack/plugin/ent-search/src/main/java/org/elasticsearch/xpack/application/rules/action/TransportDeleteQueryRulesetAction.java

@@ -14,6 +14,7 @@ import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.client.internal.Client;
 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;
 import org.elasticsearch.xpack.application.rules.QueryRulesIndexService;
@@ -28,7 +29,13 @@ public class TransportDeleteQueryRulesetAction extends HandledTransportAction<De
         ActionFilters actionFilters,
         Client client
     ) {
-        super(DeleteQueryRulesetAction.NAME, transportService, actionFilters, DeleteQueryRulesetAction.Request::new);
+        super(
+            DeleteQueryRulesetAction.NAME,
+            transportService,
+            actionFilters,
+            DeleteQueryRulesetAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.systemIndexService = new QueryRulesIndexService(client, clusterService.getClusterSettings());
     }
 

x-pack/plugin/ent-search/src/main/java/org/elasticsearch/xpack/application/rules/action/TransportGetQueryRulesetAction.java

@@ -13,6 +13,7 @@ import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.client.internal.Client;
 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;
 import org.elasticsearch.xpack.application.rules.QueryRulesIndexService;
@@ -28,7 +29,13 @@ public class TransportGetQueryRulesetAction extends HandledTransportAction<GetQu
         ActionFilters actionFilters,
         Client client
     ) {
-        super(GetQueryRulesetAction.NAME, transportService, actionFilters, GetQueryRulesetAction.Request::new);
+        super(
+            GetQueryRulesetAction.NAME,
+            transportService,
+            actionFilters,
+            GetQueryRulesetAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.systemIndexService = new QueryRulesIndexService(client, clusterService.getClusterSettings());
     }
 

x-pack/plugin/ent-search/src/main/java/org/elasticsearch/xpack/application/rules/action/TransportListQueryRulesetsAction.java

@@ -13,6 +13,7 @@ import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.client.internal.Client;
 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;
 import org.elasticsearch.xpack.application.rules.QueryRulesIndexService;
@@ -30,7 +31,13 @@ public class TransportListQueryRulesetsAction extends HandledTransportAction<
         ActionFilters actionFilters,
         Client client
     ) {
-        super(ListQueryRulesetsAction.NAME, transportService, actionFilters, ListQueryRulesetsAction.Request::new);
+        super(
+            ListQueryRulesetsAction.NAME,
+            transportService,
+            actionFilters,
+            ListQueryRulesetsAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.systemIndexService = new QueryRulesIndexService(client, clusterService.getClusterSettings());
     }
 

x-pack/plugin/ent-search/src/main/java/org/elasticsearch/xpack/application/rules/action/TransportPutQueryRulesetAction.java

@@ -13,6 +13,7 @@ import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.client.internal.Client;
 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;
 import org.elasticsearch.xpack.application.rules.QueryRulesIndexService;
@@ -28,7 +29,13 @@ public class TransportPutQueryRulesetAction extends HandledTransportAction<PutQu
         ActionFilters actionFilters,
         Client client
     ) {
-        super(PutQueryRulesetAction.NAME, transportService, actionFilters, PutQueryRulesetAction.Request::new);
+        super(
+            PutQueryRulesetAction.NAME,
+            transportService,
+            actionFilters,
+            PutQueryRulesetAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.systemIndexService = new QueryRulesIndexService(client, clusterService.getClusterSettings());
     }
 

x-pack/plugin/ent-search/src/main/java/org/elasticsearch/xpack/application/search/action/TransportDeleteSearchApplicationAction.java

@@ -16,6 +16,7 @@ import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.application.search.SearchApplicationIndexService;
@@ -34,7 +35,13 @@ public class TransportDeleteSearchApplicationAction extends HandledTransportActi
         NamedWriteableRegistry namedWriteableRegistry,
         BigArrays bigArrays
     ) {
-        super(DeleteSearchApplicationAction.NAME, transportService, actionFilters, DeleteSearchApplicationAction.Request::new);
+        super(
+            DeleteSearchApplicationAction.NAME,
+            transportService,
+            actionFilters,
+            DeleteSearchApplicationAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.systemIndexService = new SearchApplicationIndexService(client, clusterService, namedWriteableRegistry, bigArrays);
     }
 

x-pack/plugin/ent-search/src/main/java/org/elasticsearch/xpack/application/search/action/TransportGetSearchApplicationAction.java

@@ -16,6 +16,7 @@ import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.logging.HeaderWarning;
 import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.application.search.SearchApplication;
@@ -36,7 +37,13 @@ public class TransportGetSearchApplicationAction extends HandledTransportAction<
         NamedWriteableRegistry namedWriteableRegistry,
         BigArrays bigArrays
     ) {
-        super(GetSearchApplicationAction.NAME, transportService, actionFilters, GetSearchApplicationAction.Request::new);
+        super(
+            GetSearchApplicationAction.NAME,
+            transportService,
+            actionFilters,
+            GetSearchApplicationAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.systemIndexService = new SearchApplicationIndexService(client, clusterService, namedWriteableRegistry, bigArrays);
     }
 

x-pack/plugin/ent-search/src/main/java/org/elasticsearch/xpack/application/search/action/TransportListSearchApplicationAction.java

@@ -15,6 +15,7 @@ import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.application.search.SearchApplicationIndexService;
@@ -34,7 +35,13 @@ public class TransportListSearchApplicationAction extends HandledTransportAction
         NamedWriteableRegistry namedWriteableRegistry,
         BigArrays bigArrays
     ) {
-        super(ListSearchApplicationAction.NAME, transportService, actionFilters, ListSearchApplicationAction.Request::new);
+        super(
+            ListSearchApplicationAction.NAME,
+            transportService,
+            actionFilters,
+            ListSearchApplicationAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.systemIndexService = new SearchApplicationIndexService(client, clusterService, namedWriteableRegistry, bigArrays);
     }
 

x-pack/plugin/ent-search/src/main/java/org/elasticsearch/xpack/application/search/action/TransportPutSearchApplicationAction.java

@@ -16,6 +16,7 @@ import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.logging.HeaderWarning;
 import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.application.search.SearchApplication;
@@ -35,7 +36,13 @@ public class TransportPutSearchApplicationAction extends HandledTransportAction<
         NamedWriteableRegistry namedWriteableRegistry,
         BigArrays bigArrays
     ) {
-        super(PutSearchApplicationAction.NAME, transportService, actionFilters, PutSearchApplicationAction.Request::new);
+        super(
+            PutSearchApplicationAction.NAME,
+            transportService,
+            actionFilters,
+            PutSearchApplicationAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.systemIndexService = new SearchApplicationIndexService(client, clusterService, namedWriteableRegistry, bigArrays);
     }
 

x-pack/plugin/ent-search/src/main/java/org/elasticsearch/xpack/application/search/action/TransportQuerySearchApplicationAction.java

@@ -18,6 +18,7 @@ import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.logging.LogManager;
 import org.elasticsearch.logging.Logger;
 import org.elasticsearch.script.ScriptService;
@@ -47,7 +48,13 @@ public class TransportQuerySearchApplicationAction extends HandledTransportActio
         ScriptService scriptService,
         NamedXContentRegistry xContentRegistry
     ) {
-        super(QuerySearchApplicationAction.NAME, transportService, actionFilters, SearchApplicationSearchRequest::new);
+        super(
+            QuerySearchApplicationAction.NAME,
+            transportService,
+            actionFilters,
+            SearchApplicationSearchRequest::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.client = client;
         this.templateService = new SearchApplicationTemplateService(scriptService, xContentRegistry);
         this.systemIndexService = new SearchApplicationIndexService(client, clusterService, namedWriteableRegistry, bigArrays);

x-pack/plugin/ent-search/src/main/java/org/elasticsearch/xpack/application/search/action/TransportRenderSearchApplicationQueryAction.java

@@ -15,6 +15,7 @@ import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.logging.LogManager;
 import org.elasticsearch.logging.Logger;
 import org.elasticsearch.script.ScriptService;
@@ -48,7 +49,13 @@ public class TransportRenderSearchApplicationQueryAction extends HandledTranspor
         ScriptService scriptService,
         NamedXContentRegistry xContentRegistry
     ) {
-        super(RenderSearchApplicationQueryAction.NAME, transportService, actionFilters, SearchApplicationSearchRequest::new);
+        super(
+            RenderSearchApplicationQueryAction.NAME,
+            transportService,
+            actionFilters,
+            SearchApplicationSearchRequest::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.systemIndexService = new SearchApplicationIndexService(client, clusterService, namedWriteableRegistry, bigArrays);
         this.templateService = new SearchApplicationTemplateService(scriptService, xContentRegistry);
     }

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/plugin/TransportEqlSearchAction.java

@@ -86,7 +86,7 @@ public class TransportEqlSearchAction extends HandledTransportAction<EqlSearchRe
         Client client,
         BigArrays bigArrays
     ) {
-        super(EqlSearchAction.NAME, transportService, actionFilters, EqlSearchRequest::new);
+        super(EqlSearchAction.NAME, transportService, actionFilters, EqlSearchRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
 
         this.securityContext = XPackSettings.SECURITY_ENABLED.get(settings)
             ? new SecurityContext(settings, threadPool.getThreadContext())

x-pack/plugin/fleet/src/main/java/org/elasticsearch/xpack/fleet/action/TransportDeleteSecretAction.java

@@ -15,6 +15,7 @@ import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.client.internal.OriginSettingClient;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 
@@ -27,7 +28,7 @@ public class TransportDeleteSecretAction extends HandledTransportAction<DeleteSe
 
     @Inject
     public TransportDeleteSecretAction(TransportService transportService, ActionFilters actionFilters, Client client) {
-        super(DeleteSecretAction.NAME, transportService, actionFilters, DeleteSecretRequest::new);
+        super(DeleteSecretAction.NAME, transportService, actionFilters, DeleteSecretRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.client = new OriginSettingClient(client, FLEET_ORIGIN);
     }
 

x-pack/plugin/fleet/src/main/java/org/elasticsearch/xpack/fleet/action/TransportGetSecretAction.java

@@ -14,6 +14,7 @@ import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.client.internal.OriginSettingClient;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 
@@ -25,7 +26,7 @@ public class TransportGetSecretAction extends HandledTransportAction<GetSecretRe
 
     @Inject
     public TransportGetSecretAction(TransportService transportService, ActionFilters actionFilters, Client client) {
-        super(GetSecretAction.NAME, transportService, actionFilters, GetSecretRequest::new);
+        super(GetSecretAction.NAME, transportService, actionFilters, GetSecretRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.client = new OriginSettingClient(client, FLEET_ORIGIN);
     }
 

x-pack/plugin/fleet/src/main/java/org/elasticsearch/xpack/fleet/action/TransportPostSecretAction.java

@@ -13,6 +13,7 @@ import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.client.internal.OriginSettingClient;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 
@@ -26,7 +27,7 @@ public class TransportPostSecretAction extends HandledTransportAction<PostSecret
 
     @Inject
     public TransportPostSecretAction(TransportService transportService, ActionFilters actionFilters, Client client) {
-        super(PostSecretAction.NAME, transportService, actionFilters, PostSecretRequest::new);
+        super(PostSecretAction.NAME, transportService, actionFilters, PostSecretRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.client = new OriginSettingClient(client, FLEET_ORIGIN);
     }
 

x-pack/plugin/graph/src/main/java/org/elasticsearch/xpack/graph/action/TransportGraphExploreAction.java

@@ -23,6 +23,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.CollectionUtils;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.query.BoolQueryBuilder;
 import org.elasticsearch.index.query.QueryBuilders;
@@ -97,7 +98,7 @@ public class TransportGraphExploreAction extends HandledTransportAction<GraphExp
         ActionFilters actionFilters,
         XPackLicenseState licenseState
     ) {
-        super(GraphExploreAction.NAME, transportService, actionFilters, GraphExploreRequest::new);
+        super(GraphExploreAction.NAME, transportService, actionFilters, GraphExploreRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.threadPool = threadPool;
         this.client = client;
         this.licenseState = licenseState;

x-pack/plugin/identity-provider/src/main/java/org/elasticsearch/xpack/idp/action/TransportDeleteSamlServiceProviderAction.java

@@ -12,6 +12,7 @@ import org.elasticsearch.action.ActionListener;
 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.common.util.iterable.Iterables;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
@@ -36,7 +37,13 @@ public class TransportDeleteSamlServiceProviderAction extends HandledTransportAc
         ActionFilters actionFilters,
         SamlServiceProviderIndex index
     ) {
-        super(DeleteSamlServiceProviderAction.NAME, transportService, actionFilters, DeleteSamlServiceProviderRequest::new);
+        super(
+            DeleteSamlServiceProviderAction.NAME,
+            transportService,
+            actionFilters,
+            DeleteSamlServiceProviderRequest::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.index = index;
     }
 

x-pack/plugin/identity-provider/src/main/java/org/elasticsearch/xpack/idp/action/TransportPutSamlServiceProviderAction.java

@@ -17,6 +17,7 @@ import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.common.ValidationException;
 import org.elasticsearch.common.hash.MessageDigests;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.common.util.iterable.Iterables;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
@@ -56,7 +57,13 @@ public class TransportPutSamlServiceProviderAction extends HandledTransportActio
         SamlIdentityProvider identityProvider,
         Clock clock
     ) {
-        super(PutSamlServiceProviderAction.NAME, transportService, actionFilters, PutSamlServiceProviderRequest::new);
+        super(
+            PutSamlServiceProviderAction.NAME,
+            transportService,
+            actionFilters,
+            PutSamlServiceProviderRequest::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.index = index;
         this.identityProvider = identityProvider;
         this.clock = clock;

x-pack/plugin/identity-provider/src/main/java/org/elasticsearch/xpack/idp/action/TransportSamlInitiateSingleSignOnAction.java

@@ -14,6 +14,7 @@ import org.elasticsearch.action.ActionListener;
 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.rest.RestStatus;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
@@ -53,7 +54,13 @@ public class TransportSamlInitiateSingleSignOnAction extends HandledTransportAct
         SamlFactory factory,
         UserPrivilegeResolver privilegeResolver
     ) {
-        super(SamlInitiateSingleSignOnAction.NAME, transportService, actionFilters, SamlInitiateSingleSignOnRequest::new);
+        super(
+            SamlInitiateSingleSignOnAction.NAME,
+            transportService,
+            actionFilters,
+            SamlInitiateSingleSignOnRequest::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.securityContext = securityContext;
         this.identityProvider = idp;
         this.samlFactory = factory;

x-pack/plugin/identity-provider/src/main/java/org/elasticsearch/xpack/idp/action/TransportSamlMetadataAction.java

@@ -10,6 +10,7 @@ import org.elasticsearch.action.ActionListener;
 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.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.idp.saml.idp.SamlIdentityProvider;
@@ -28,7 +29,7 @@ public class TransportSamlMetadataAction extends HandledTransportAction<SamlMeta
         SamlIdentityProvider idp,
         SamlFactory factory
     ) {
-        super(SamlMetadataAction.NAME, transportService, actionFilters, SamlMetadataRequest::new);
+        super(SamlMetadataAction.NAME, transportService, actionFilters, SamlMetadataRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.identityProvider = idp;
         this.samlFactory = factory;
     }

x-pack/plugin/identity-provider/src/main/java/org/elasticsearch/xpack/idp/action/TransportSamlValidateAuthnRequestAction.java

@@ -10,6 +10,7 @@ import org.elasticsearch.action.ActionListener;
 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.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.idp.saml.authn.SamlAuthnRequestValidator;
@@ -30,7 +31,13 @@ public class TransportSamlValidateAuthnRequestAction extends HandledTransportAct
         SamlIdentityProvider idp,
         SamlFactory factory
     ) {
-        super(SamlValidateAuthnRequestAction.NAME, transportService, actionFilters, SamlValidateAuthnRequestRequest::new);
+        super(
+            SamlValidateAuthnRequestAction.NAME,
+            transportService,
+            actionFilters,
+            SamlValidateAuthnRequestRequest::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.identityProvider = idp;
         this.samlFactory = factory;
     }

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/action/TransportGetInferenceModelAction.java

@@ -12,6 +12,7 @@ import org.elasticsearch.action.ActionListener;
 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.inference.InferenceServiceRegistry;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.tasks.Task;
@@ -33,7 +34,13 @@ public class TransportGetInferenceModelAction extends HandledTransportAction<
         ModelRegistry modelRegistry,
         InferenceServiceRegistry serviceRegistry
     ) {
-        super(GetInferenceModelAction.NAME, transportService, actionFilters, GetInferenceModelAction.Request::new);
+        super(
+            GetInferenceModelAction.NAME,
+            transportService,
+            actionFilters,
+            GetInferenceModelAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.modelRegistry = modelRegistry;
         this.serviceRegistry = serviceRegistry;
     }

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/action/TransportInferenceAction.java

@@ -12,6 +12,7 @@ import org.elasticsearch.action.ActionListener;
 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.inference.InferenceService;
 import org.elasticsearch.inference.InferenceServiceRegistry;
 import org.elasticsearch.inference.Model;
@@ -33,7 +34,7 @@ public class TransportInferenceAction extends HandledTransportAction<InferenceAc
         ModelRegistry modelRegistry,
         InferenceServiceRegistry serviceRegistry
     ) {
-        super(InferenceAction.NAME, transportService, actionFilters, InferenceAction.Request::new);
+        super(InferenceAction.NAME, transportService, actionFilters, InferenceAction.Request::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.modelRegistry = modelRegistry;
         this.serviceRegistry = serviceRegistry;
     }

x-pack/plugin/logstash/src/main/java/org/elasticsearch/xpack/logstash/action/TransportDeletePipelineAction.java

@@ -16,6 +16,7 @@ import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.client.internal.OriginSettingClient;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
@@ -29,7 +30,7 @@ public class TransportDeletePipelineAction extends HandledTransportAction<Delete
 
     @Inject
     public TransportDeletePipelineAction(TransportService transportService, ActionFilters actionFilters, Client client) {
-        super(DeletePipelineAction.NAME, transportService, actionFilters, DeletePipelineRequest::new);
+        super(DeletePipelineAction.NAME, transportService, actionFilters, DeletePipelineRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.client = new OriginSettingClient(client, LOGSTASH_MANAGEMENT_ORIGIN);
     }
 

x-pack/plugin/logstash/src/main/java/org/elasticsearch/xpack/logstash/action/TransportGetPipelineAction.java

@@ -23,6 +23,7 @@ import org.elasticsearch.client.internal.OriginSettingClient;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.util.Maps;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.index.query.QueryBuilders;
@@ -57,7 +58,7 @@ public class TransportGetPipelineAction extends HandledTransportAction<GetPipeli
 
     @Inject
     public TransportGetPipelineAction(TransportService transportService, ActionFilters actionFilters, Client client) {
-        super(GetPipelineAction.NAME, transportService, actionFilters, GetPipelineRequest::new);
+        super(GetPipelineAction.NAME, transportService, actionFilters, GetPipelineRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.client = new OriginSettingClient(client, LOGSTASH_MANAGEMENT_ORIGIN);
     }
 

x-pack/plugin/logstash/src/main/java/org/elasticsearch/xpack/logstash/action/TransportPutPipelineAction.java

@@ -14,6 +14,7 @@ import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.client.internal.OriginSettingClient;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.logstash.Logstash;
@@ -26,7 +27,7 @@ public class TransportPutPipelineAction extends HandledTransportAction<PutPipeli
 
     @Inject
     public TransportPutPipelineAction(TransportService transportService, ActionFilters actionFilters, Client client) {
-        super(PutPipelineAction.NAME, transportService, actionFilters, PutPipelineRequest::new);
+        super(PutPipelineAction.NAME, transportService, actionFilters, PutPipelineRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.client = new OriginSettingClient(client, LOGSTASH_MANAGEMENT_ORIGIN);
     }
 

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportAuditMlNotificationAction.java

@@ -12,6 +12,7 @@ import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.core.common.notifications.AbstractAuditor;
@@ -37,7 +38,13 @@ public class TransportAuditMlNotificationAction extends HandledTransportAction<A
         InferenceAuditor inferenceAuditor,
         SystemAuditor systemAuditor
     ) {
-        super(AuditMlNotificationAction.NAME, transportService, actionFilters, AuditMlNotificationAction.Request::new);
+        super(
+            AuditMlNotificationAction.NAME,
+            transportService,
+            actionFilters,
+            AuditMlNotificationAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.anomalyDetectionAuditor = anomalyDetectionAuditor;
         this.dfaAuditor = dfaAuditor;
         this.inferenceAuditor = inferenceAuditor;

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportCancelJobModelSnapshotUpgradeAction.java

@@ -17,6 +17,7 @@ import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.util.concurrent.AtomicArray;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.persistent.PersistentTasksCustomMetadata;
 import org.elasticsearch.persistent.PersistentTasksService;
 import org.elasticsearch.tasks.Task;
@@ -52,7 +53,7 @@ public class TransportCancelJobModelSnapshotUpgradeAction extends HandledTranspo
         ClusterService clusterService,
         PersistentTasksService persistentTasksService
     ) {
-        super(CancelJobModelSnapshotUpgradeAction.NAME, transportService, actionFilters, Request::new);
+        super(CancelJobModelSnapshotUpgradeAction.NAME, transportService, actionFilters, Request::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.jobConfigProvider = jobConfigProvider;
         this.clusterService = clusterService;
         this.persistentTasksService = persistentTasksService;

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteCalendarAction.java

@@ -13,6 +13,7 @@ import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.index.reindex.AbstractBulkByScrollRequest;
@@ -43,7 +44,13 @@ public class TransportDeleteCalendarAction extends HandledTransportAction<Delete
         JobManager jobManager,
         JobResultsProvider jobResultsProvider
     ) {
-        super(DeleteCalendarAction.NAME, transportService, actionFilters, DeleteCalendarAction.Request::new);
+        super(
+            DeleteCalendarAction.NAME,
+            transportService,
+            actionFilters,
+            DeleteCalendarAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.client = client;
         this.jobManager = jobManager;
         this.jobResultsProvider = jobResultsProvider;

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteCalendarEventAction.java

@@ -19,6 +19,7 @@ import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
@@ -48,7 +49,13 @@ public class TransportDeleteCalendarEventAction extends HandledTransportAction<D
         JobResultsProvider jobResultsProvider,
         JobManager jobManager
     ) {
-        super(DeleteCalendarEventAction.NAME, transportService, actionFilters, DeleteCalendarEventAction.Request::new);
+        super(
+            DeleteCalendarEventAction.NAME,
+            transportService,
+            actionFilters,
+            DeleteCalendarEventAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.client = client;
         this.jobResultsProvider = jobResultsProvider;
         this.jobManager = jobManager;

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteFilterAction.java

@@ -18,6 +18,7 @@ import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
@@ -48,7 +49,13 @@ public class TransportDeleteFilterAction extends HandledTransportAction<DeleteFi
         Client client,
         JobConfigProvider jobConfigProvider
     ) {
-        super(DeleteFilterAction.NAME, transportService, actionFilters, DeleteFilterAction.Request::new);
+        super(
+            DeleteFilterAction.NAME,
+            transportService,
+            actionFilters,
+            DeleteFilterAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.client = client;
         this.jobConfigProvider = jobConfigProvider;
     }

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteForecastAction.java

@@ -24,6 +24,7 @@ import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.query.BoolQueryBuilder;
 import org.elasticsearch.index.query.QueryBuilder;
@@ -81,7 +82,13 @@ public class TransportDeleteForecastAction extends HandledTransportAction<Delete
         Client client,
         ClusterService clusterService
     ) {
-        super(DeleteForecastAction.NAME, transportService, actionFilters, DeleteForecastAction.Request::new);
+        super(
+            DeleteForecastAction.NAME,
+            transportService,
+            actionFilters,
+            DeleteForecastAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.client = client;
         this.clusterService = clusterService;
     }

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteModelSnapshotAction.java

@@ -15,6 +15,7 @@ import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.core.ml.action.DeleteModelSnapshotAction;
@@ -48,7 +49,13 @@ public class TransportDeleteModelSnapshotAction extends HandledTransportAction<D
         JobManager jobManager,
         AnomalyDetectionAuditor auditor
     ) {
-        super(DeleteModelSnapshotAction.NAME, transportService, actionFilters, DeleteModelSnapshotAction.Request::new);
+        super(
+            DeleteModelSnapshotAction.NAME,
+            transportService,
+            actionFilters,
+            DeleteModelSnapshotAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.client = client;
         this.jobManager = jobManager;
         this.jobResultsProvider = jobResultsProvider;

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportEstimateModelMemoryAction.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.unit.ByteSizeValue;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.core.ml.action.EstimateModelMemoryAction;
@@ -42,7 +43,13 @@ public class TransportEstimateModelMemoryAction extends HandledTransportAction<
 
     @Inject
     public TransportEstimateModelMemoryAction(TransportService transportService, ActionFilters actionFilters) {
-        super(EstimateModelMemoryAction.NAME, transportService, actionFilters, EstimateModelMemoryAction.Request::new);
+        super(
+            EstimateModelMemoryAction.NAME,
+            transportService,
+            actionFilters,
+            EstimateModelMemoryAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
     }
 
     @Override

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportEvaluateDataFrameAction.java

@@ -16,6 +16,7 @@ import org.elasticsearch.client.internal.ParentTaskAssigningClient;
 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.search.builder.SearchSourceBuilder;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.tasks.TaskId;
@@ -53,7 +54,13 @@ public class TransportEvaluateDataFrameAction extends HandledTransportAction<
         Client client,
         ClusterService clusterService
     ) {
-        super(EvaluateDataFrameAction.NAME, transportService, actionFilters, EvaluateDataFrameAction.Request::new);
+        super(
+            EvaluateDataFrameAction.NAME,
+            transportService,
+            actionFilters,
+            EvaluateDataFrameAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.threadPool = threadPool;
         this.client = client;
         this.securityContext = XPackSettings.SECURITY_ENABLED.get(settings)

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportExplainDataFrameAnalyticsAction.java

@@ -21,6 +21,7 @@ import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.license.LicenseUtils;
 import org.elasticsearch.license.XPackLicenseState;
@@ -81,7 +82,13 @@ public class TransportExplainDataFrameAnalyticsAction extends HandledTransportAc
         Settings settings,
         ThreadPool threadPool
     ) {
-        super(ExplainDataFrameAnalyticsAction.NAME, transportService, actionFilters, ExplainDataFrameAnalyticsAction.Request::new);
+        super(
+            ExplainDataFrameAnalyticsAction.NAME,
+            transportService,
+            actionFilters,
+            ExplainDataFrameAnalyticsAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.transportService = transportService;
         this.clusterService = Objects.requireNonNull(clusterService);
         this.client = Objects.requireNonNull(client);

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetBucketsAction.java

@@ -11,6 +11,7 @@ import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.core.ml.action.GetBucketsAction;
@@ -32,7 +33,7 @@ public class TransportGetBucketsAction extends HandledTransportAction<GetBuckets
         JobManager jobManager,
         Client client
     ) {
-        super(GetBucketsAction.NAME, transportService, actionFilters, GetBucketsAction.Request::new);
+        super(GetBucketsAction.NAME, transportService, actionFilters, GetBucketsAction.Request::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.jobResultsProvider = jobResultsProvider;
         this.jobManager = jobManager;
         this.client = client;

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetCalendarEventsAction.java

@@ -11,6 +11,7 @@ import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.core.action.util.QueryPage;
@@ -39,7 +40,13 @@ public class TransportGetCalendarEventsAction extends HandledTransportAction<
         JobResultsProvider jobResultsProvider,
         JobConfigProvider jobConfigProvider
     ) {
-        super(GetCalendarEventsAction.NAME, transportService, actionFilters, GetCalendarEventsAction.Request::new);
+        super(
+            GetCalendarEventsAction.NAME,
+            transportService,
+            actionFilters,
+            GetCalendarEventsAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.jobResultsProvider = jobResultsProvider;
         this.jobConfigProvider = jobConfigProvider;
     }

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetCalendarsAction.java

@@ -11,6 +11,7 @@ import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.core.action.util.PageParams;
@@ -28,7 +29,13 @@ public class TransportGetCalendarsAction extends HandledTransportAction<GetCalen
         ActionFilters actionFilters,
         JobResultsProvider jobResultsProvider
     ) {
-        super(GetCalendarsAction.NAME, transportService, actionFilters, GetCalendarsAction.Request::new);
+        super(
+            GetCalendarsAction.NAME,
+            transportService,
+            actionFilters,
+            GetCalendarsAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.jobResultsProvider = jobResultsProvider;
     }
 

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetCategoriesAction.java

@@ -13,6 +13,7 @@ import org.elasticsearch.client.internal.Client;
 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.tasks.CancellableTask;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.tasks.TaskId;
@@ -37,7 +38,13 @@ public class TransportGetCategoriesAction extends HandledTransportAction<GetCate
         JobManager jobManager,
         ClusterService clusterService
     ) {
-        super(GetCategoriesAction.NAME, transportService, actionFilters, GetCategoriesAction.Request::new);
+        super(
+            GetCategoriesAction.NAME,
+            transportService,
+            actionFilters,
+            GetCategoriesAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.jobResultsProvider = jobResultsProvider;
         this.client = client;
         this.jobManager = jobManager;

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetDatafeedsStatsAction.java

@@ -16,6 +16,7 @@ import org.elasticsearch.client.internal.OriginSettingClient;
 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.persistent.PersistentTasksCustomMetadata;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.tasks.TaskId;
@@ -55,7 +56,7 @@ public class TransportGetDatafeedsStatsAction extends HandledTransportAction<Req
         JobResultsProvider jobResultsProvider,
         Client client
     ) {
-        super(GetDatafeedsStatsAction.NAME, transportService, actionFilters, Request::new);
+        super(GetDatafeedsStatsAction.NAME, transportService, actionFilters, Request::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.clusterService = clusterService;
         this.datafeedConfigProvider = datafeedConfigProvider;
         this.jobResultsProvider = jobResultsProvider;

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetInfluencersAction.java

@@ -11,6 +11,7 @@ import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.core.ml.action.GetInfluencersAction;
@@ -32,7 +33,13 @@ public class TransportGetInfluencersAction extends HandledTransportAction<GetInf
         Client client,
         JobManager jobManager
     ) {
-        super(GetInfluencersAction.NAME, transportService, actionFilters, GetInfluencersAction.Request::new);
+        super(
+            GetInfluencersAction.NAME,
+            transportService,
+            actionFilters,
+            GetInfluencersAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.jobResultsProvider = jobResultsProvider;
         this.client = client;
         this.jobManager = jobManager;

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportGetModelSnapshotsAction.java

@@ -14,6 +14,7 @@ import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.transport.TransportService;
@@ -41,7 +42,13 @@ public class TransportGetModelSnapshotsAction extends HandledTransportAction<
         JobManager jobManager,
         ClusterService clusterService
     ) {
-        super(GetModelSnapshotsAction.NAME, transportService, actionFilters, GetModelSnapshotsAction.Request::new);
+        super(
+            GetModelSnapshotsAction.NAME,
+            transportService,
+            actionFilters,
+            GetModelSnapshotsAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
         this.jobResultsProvider = jobResultsProvider;
         this.jobManager = jobManager;
         this.clusterService = clusterService;