Add thread pool for write coordination

This change adds a thread pool for write coordination to ensure that
bulk coordination does not get stuck on an overloaded primary node.

Author: Tim-Brooks

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

@@ -68,6 +68,7 @@ public abstract class TransportAbstractBulkAction extends HandledTransportAction
     private final IngestService ingestService;
     private final IngestActionForwarder ingestForwarder;
     protected final LongSupplier relativeTimeNanosProvider;
+    protected final Executor coordinationExecutor;
     protected final Executor writeExecutor;
     protected final Executor systemWriteExecutor;
     private final ActionType<BulkResponse> bulkAction;
@@ -92,6 +93,7 @@ public TransportAbstractBulkAction(
         this.indexingPressure = indexingPressure;
         this.systemIndices = systemIndices;
         this.projectResolver = projectResolver;
+        this.coordinationExecutor = threadPool.executor(ThreadPool.Names.WRITE_COORDINATION);
         this.writeExecutor = threadPool.executor(ThreadPool.Names.WRITE);
         this.systemWriteExecutor = threadPool.executor(ThreadPool.Names.SYSTEM_WRITE);
         this.ingestForwarder = new IngestActionForwarder(transportService);
@@ -106,8 +108,8 @@ protected void doExecute(Task task, BulkRequest bulkRequest, ActionListener<Bulk
          * This is called on the Transport thread so we can check the indexing
          * memory pressure *quickly* but we don't want to keep the transport
          * thread busy. Then, as soon as we have the indexing pressure in we fork
-         * to one of the write thread pools. We do this because juggling the
-         * bulk request can get expensive for a few reasons:
+         * to the coordinator thread pool for coordination tasks. We do this because
+         * juggling the bulk request can get expensive for a few reasons:
          * 1. Figuring out which shard should receive a bulk request might require
          *    parsing the _source.
          * 2. When dispatching the sub-requests to shards we may have to compress
@@ -131,14 +133,15 @@ protected void doExecute(Task task, BulkRequest bulkRequest, ActionListener<Bulk
             releasable = indexingPressure.markCoordinatingOperationStarted(indexingOps, indexingBytes, isOnlySystem);
         }
         final ActionListener<BulkResponse> releasingListener = ActionListener.runBefore(listener, releasable::close);
-        final Executor executor = isOnlySystem ? systemWriteExecutor : writeExecutor;
-        ensureClusterStateThenForkAndExecute(task, bulkRequest, executor, releasingListener);
+        // Use coordinationExecutor for dispatching coordination tasks
+        ensureClusterStateThenForkAndExecute(task, bulkRequest, coordinationExecutor, isOnlySystem, releasingListener);
     }
 
     private void ensureClusterStateThenForkAndExecute(
         Task task,
         BulkRequest bulkRequest,
         Executor executor,
+        boolean isOnlySystem,
         ActionListener<BulkResponse> releasingListener
     ) {
         final ClusterState initialState = clusterService.state();
@@ -159,7 +162,7 @@ private void ensureClusterStateThenForkAndExecute(
             clusterStateObserver.waitForNextChange(new ClusterStateObserver.Listener() {
                 @Override
                 public void onNewClusterState(ClusterState state) {
-                    forkAndExecute(task, bulkRequest, executor, releasingListener);
+                    forkAndExecute(task, bulkRequest, executor, isOnlySystem, releasingListener);
                 }
 
                 @Override
@@ -173,21 +176,32 @@ public void onTimeout(TimeValue timeout) {
                 }
             }, newState -> false == newState.blocks().hasGlobalBlockWithLevel(ClusterBlockLevel.WRITE));
         } else {
-            forkAndExecute(task, bulkRequest, executor, releasingListener);
+            forkAndExecute(task, bulkRequest, executor, isOnlySystem, releasingListener);
         }
     }
 
-    private void forkAndExecute(Task task, BulkRequest bulkRequest, Executor executor, ActionListener<BulkResponse> releasingListener) {
+    private void forkAndExecute(
+        Task task,
+        BulkRequest bulkRequest,
+        Executor executor,
+        boolean isOnlySystem,
+        ActionListener<BulkResponse> releasingListener
+    ) {
         executor.execute(new ActionRunnable<>(releasingListener) {
             @Override
             protected void doRun() throws IOException {
-                applyPipelinesAndDoInternalExecute(task, bulkRequest, executor, releasingListener);
+                applyPipelinesAndDoInternalExecute(task, bulkRequest, executor, isOnlySystem, releasingListener);
             }
         });
     }
 
-    private boolean applyPipelines(Task task, BulkRequest bulkRequest, Executor executor, ActionListener<BulkResponse> listener)
-        throws IOException {
+    private boolean applyPipelines(
+        Task task,
+        BulkRequest bulkRequest,
+        Executor executor,
+        boolean isOnlySystem,
+        ActionListener<BulkResponse> listener
+    ) throws IOException {
         boolean hasIndexRequestsWithPipelines = false;
         ClusterState state = clusterService.state();
         ProjectId projectId = projectResolver.getProjectId();
@@ -276,7 +290,7 @@ private boolean applyPipelines(Task task, BulkRequest bulkRequest, Executor exec
                     assert arePipelinesResolved : bulkRequest;
                 }
                 if (clusterService.localNode().isIngestNode()) {
-                    processBulkIndexIngestRequest(task, bulkRequest, executor, project, l);
+                    processBulkIndexIngestRequest(task, bulkRequest, executor, isOnlySystem, project, l);
                 } else {
                     ingestForwarder.forwardIngestRequest(bulkAction, bulkRequest, l);
                 }
@@ -290,6 +304,7 @@ private void processBulkIndexIngestRequest(
         Task task,
         BulkRequest original,
         Executor executor,
+        boolean isOnlySystem,
         ProjectMetadata metadata,
         ActionListener<BulkResponse> listener
     ) {
@@ -323,20 +338,21 @@ private void processBulkIndexIngestRequest(
                         ActionRunnable<BulkResponse> runnable = new ActionRunnable<>(actionListener) {
                             @Override
                             protected void doRun() throws IOException {
-                                applyPipelinesAndDoInternalExecute(task, bulkRequest, executor, actionListener);
+                                applyPipelinesAndDoInternalExecute(task, bulkRequest, executor, isOnlySystem, actionListener);
                             }
 
                             @Override
                             public boolean isForceExecution() {
-                                // If we fork back to a write thread we **not** should fail, because tp queue is full.
+                                // If we fork back to a coordination thread we **not** should fail, because tp queue is full.
                                 // (Otherwise the work done during ingest will be lost)
                                 // It is okay to force execution here. Throttling of write requests happens prior to
                                 // ingest when a node receives a bulk request.
                                 return true;
                             }
                         };
                         // If a processor went async and returned a response on a different thread then
-                        // before we continue the bulk request we should fork back on a write thread:
+                        // before we continue the bulk request we should fork back on a coordination thread. Otherwise it is fine to perform
+                        // coordination steps on the write thread
                         if (originalThread == Thread.currentThread()) {
                             runnable.run();
                         } else {
@@ -345,7 +361,8 @@ public boolean isForceExecution() {
                     }
                 }
             },
-            executor
+            // Use the appropriate write executor for actual ingest processing
+            isOnlySystem ? systemWriteExecutor : writeExecutor
         );
     }
 
@@ -401,10 +418,11 @@ private void applyPipelinesAndDoInternalExecute(
         Task task,
         BulkRequest bulkRequest,
         Executor executor,
+        boolean isOnlySystem,
         ActionListener<BulkResponse> listener
     ) throws IOException {
         final long relativeStartTimeNanos = relativeTimeNanos();
-        if (applyPipelines(task, bulkRequest, executor, listener) == false) {
+        if (applyPipelines(task, bulkRequest, executor, isOnlySystem, listener) == false) {
             doInternalExecute(task, bulkRequest, executor, listener, relativeStartTimeNanos);
         }
     }

server/src/main/java/org/elasticsearch/threadpool/DefaultBuiltInExecutorBuilders.java

@@ -38,6 +38,16 @@ public Map<String, ExecutorBuilder> getBuilders(Settings settings, int allocated
             ThreadPool.Names.GENERIC,
             new ScalingExecutorBuilder(ThreadPool.Names.GENERIC, 4, genericThreadPoolMax, TimeValue.timeValueSeconds(30), false)
         );
+        result.put(
+            ThreadPool.Names.WRITE_COORDINATION,
+            new FixedExecutorBuilder(
+                settings,
+                ThreadPool.Names.WRITE_COORDINATION,
+                allocatedProcessors,
+                10000,
+                EsExecutors.TaskTrackingConfig.DO_NOT_TRACK
+            )
+        );
         result.put(
             ThreadPool.Names.WRITE,
             new FixedExecutorBuilder(

server/src/main/java/org/elasticsearch/threadpool/ExecutorBuilder.java

@@ -41,6 +41,7 @@ protected static String settingsKey(final String prefix, final String key) {
 
     protected static int applyHardSizeLimit(final Settings settings, final String name) {
         if (name.equals("bulk")
+            || name.equals(ThreadPool.Names.WRITE_COORDINATION)
             || name.equals(ThreadPool.Names.WRITE)
             || name.equals(ThreadPool.Names.SYSTEM_WRITE)
             || name.equals(ThreadPool.Names.SYSTEM_CRITICAL_WRITE)) {

server/src/main/java/org/elasticsearch/threadpool/ThreadPool.java

@@ -113,6 +113,7 @@ public static class Names {
         public static final String GET = "get";
         public static final String ANALYZE = "analyze";
         public static final String WRITE = "write";
+        public static final String WRITE_COORDINATION = "write_coordination";
         public static final String SEARCH = "search";
         public static final String SEARCH_COORDINATION = "search_coordination";
         public static final String AUTO_COMPLETE = "auto_complete";
@@ -186,6 +187,7 @@ public static ThreadPoolType fromType(String type) {
         entry(Names.CLUSTER_COORDINATION, ThreadPoolType.FIXED),
         entry(Names.GET, ThreadPoolType.FIXED),
         entry(Names.ANALYZE, ThreadPoolType.FIXED),
+        entry(Names.WRITE_COORDINATION, ThreadPoolType.FIXED),
         entry(Names.WRITE, ThreadPoolType.FIXED),
         entry(Names.SEARCH, ThreadPoolType.FIXED),
         entry(Names.SEARCH_COORDINATION, ThreadPoolType.FIXED),