Change

Author: Tim-Brooks

server/src/internalClusterTest/java/org/elasticsearch/monitor/metrics/NodeIndexingMetricsIT.java

@@ -840,7 +840,7 @@ public void testIncrementalBulkHighWatermarkSplitMetrics() throws Exception {
         add512BRequests(requestsThrottle, index);
 
         CountDownLatch finishLatch = new CountDownLatch(1);
-        blockWritePool(threadPool, finishLatch);
+        blockWriteCoordinationPool(threadPool, finishLatch);
         IncrementalBulkService.Handler handlerThrottled = incrementalBulkService.newBulkRequest();
         refCounted.incRef();
         handlerThrottled.addItems(requestsThrottle, refCounted::decRef, () -> nextPage.set(true));
@@ -919,8 +919,8 @@ private static void add512BRequests(ArrayList<DocWriteRequest<?>> requests, Stri
         assertThat(total, lessThan(1024L));
     }
 
-    private static void blockWritePool(ThreadPool threadPool, CountDownLatch finishLatch) {
-        final var threadCount = threadPool.info(ThreadPool.Names.WRITE).getMax();
+    private static void blockWriteCoordinationPool(ThreadPool threadPool, CountDownLatch finishLatch) {
+        final var threadCount = threadPool.info(ThreadPool.Names.WRITE_COORDINATION).getMax();
         final var startBarrier = new CyclicBarrier(threadCount + 1);
         final var blockingTask = new AbstractRunnable() {
             @Override
@@ -940,7 +940,7 @@ public boolean isForceExecution() {
             }
         };
         for (int i = 0; i < threadCount; i++) {
-            threadPool.executor(ThreadPool.Names.WRITE).execute(blockingTask);
+            threadPool.executor(ThreadPool.Names.WRITE_COORDINATION).execute(blockingTask);
         }
         safeAwait(startBarrier);
     }

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

@@ -353,11 +353,7 @@ public boolean isForceExecution() {
                         // 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 coordination thread. Otherwise it is fine to perform
                         // coordination steps on the write thread
-                        if (originalThread == Thread.currentThread()) {
-                            runnable.run();
-                        } else {
-                            executor.execute(runnable);
-                        }
+                        runOrDispatch(executor, originalThread, runnable);
                     }
                 }
             },
@@ -366,6 +362,17 @@ public boolean isForceExecution() {
         );
     }
 
+    /**
+     * If the work is still on the original thread run the ActionRunnable synchronously. Otherwise, dispatch to the provided executor.
+     */
+    protected static void runOrDispatch(Executor executor, Thread originalThread, ActionRunnable<BulkResponse> bulkRunnable) {
+        if (originalThread == Thread.currentThread()) {
+            bulkRunnable.run();
+        } else {
+            executor.execute(bulkRunnable);
+        }
+    }
+
     /**
      * Determines if an index name is associated with either an existing data stream or a template
      * for one that has the failure store enabled.

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

@@ -385,19 +385,26 @@ protected void createMissingIndicesAndIndexData(
         Map<String, Exception> indicesExceptions = new ConcurrentHashMap<>();
         Map<String, Exception> dataStreamExceptions = new ConcurrentHashMap<>();
         Map<String, Exception> failureStoreExceptions = new ConcurrentHashMap<>();
-        Runnable executeBulkRunnable = () -> executor.execute(new ActionRunnable<>(listener) {
-            @Override
-            protected void doRun() {
-                failRequestsWhenPrerequisiteActionFailed(
-                    indicesExceptions,
-                    dataStreamExceptions,
-                    failureStoreExceptions,
-                    bulkRequest,
-                    responses
-                );
-                executeBulk(task, bulkRequest, startTimeNanos, listener, executor, responses);
-            }
-        });
+
+        Thread originalThread = Thread.currentThread();
+        Runnable executeBulkRunnable = () -> {
+            ActionRunnable<BulkResponse> bulkRunnable = new ActionRunnable<>(listener) {
+                @Override
+                protected void doRun() {
+                    failRequestsWhenPrerequisiteActionFailed(
+                        indicesExceptions,
+                        dataStreamExceptions,
+                        failureStoreExceptions,
+                        bulkRequest,
+                        responses
+                    );
+                    executeBulk(task, bulkRequest, startTimeNanos, listener, executor, responses);
+                }
+            };
+            // If we performed some async action and ended up on a different thread dispatch back to the coordination thread pool.
+            // Otherwise, just run the action since we are on the same thread.
+            runOrDispatch(executor, originalThread, bulkRunnable);
+        };
         try (RefCountingRunnable refs = new RefCountingRunnable(executeBulkRunnable)) {
             createIndices(indicesToAutoCreate, refs, indicesExceptions);
             rollOverDataStreams(bulkRequest, dataStreamsToBeRolledOver, false, refs, dataStreamExceptions);

server/src/test/java/org/elasticsearch/action/bulk/TransportBulkActionTests.java

@@ -70,6 +70,7 @@
 import org.elasticsearch.test.transport.CapturingTransport;
 import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.threadpool.ThreadPoolStats;
 import org.elasticsearch.transport.TransportService;
 import org.junit.After;
 import org.junit.Before;
@@ -422,6 +423,22 @@ private void blockWriteCoordinationThreadPool(CountDownLatch blockingLatch) {
         });
     }
 
+    public void testDispatchesToWriteCoordinationThreadPoolOnce() {
+        BulkRequest bulkRequest = new BulkRequest().add(new IndexRequest("index").id("id").source(Collections.emptyMap()));
+        PlainActionFuture<BulkResponse> future = new PlainActionFuture<>();
+        ThreadPoolStats.Stats stats = threadPool.stats()
+            .stats()
+            .stream()
+            .filter(s -> s.name().equals(ThreadPool.Names.WRITE_COORDINATION))
+            .findAny()
+            .get();
+        assertThat(stats.completed(), equalTo(0L));
+        ActionTestUtils.execute(bulkAction, null, bulkRequest, future);
+        future.actionGet();
+        stats = threadPool.stats().stats().stream().filter(s -> s.name().equals(ThreadPool.Names.WRITE_COORDINATION)).findAny().get();
+        assertThat(stats.completed(), equalTo(1L));
+    }
+
     public void testRejectCoordination() {
         BulkRequest bulkRequest = new BulkRequest().add(new IndexRequest("index").id("id").source(Collections.emptyMap()));
 

server/src/test/java/org/elasticsearch/common/util/concurrent/EsThreadPoolExecutorTests.java

@@ -32,6 +32,8 @@ public class EsThreadPoolExecutorTests extends ESSingleNodeTestCase {
     protected Settings nodeSettings() {
         return Settings.builder()
             .put("node.name", "es-thread-pool-executor-tests")
+            .put("thread_pool.write_coordination.size", 1)
+            .put("thread_pool.write_coordination.queue_size", 0)
             .put("thread_pool.write.size", 1)
             .put("thread_pool.write.queue_size", 0)
             .put("thread_pool.search.size", 1)
@@ -41,7 +43,7 @@ protected Settings nodeSettings() {
 
     public void testRejectedExecutionExceptionContainsNodeName() {
         // we test a fixed and an auto-queue executor but not scaling since it does not reject
-        runThreadPoolExecutorTest(1, ThreadPool.Names.WRITE);
+        runThreadPoolExecutorTest(1, randomFrom(ThreadPool.Names.WRITE_COORDINATION, ThreadPool.Names.WRITE));
         runThreadPoolExecutorTest(2, ThreadPool.Names.SEARCH);
 
     }

server/src/test/java/org/elasticsearch/threadpool/ThreadPoolTests.java

@@ -573,6 +573,7 @@ public void testThreadCountMetrics() throws Exception {
                 ThreadPool.Names.GENERIC,
                 ThreadPool.Names.ANALYZE,
                 ThreadPool.Names.WRITE,
+                ThreadPool.Names.WRITE_COORDINATION,
                 ThreadPool.Names.SEARCH
             );
             final ThreadPool.Info threadPoolInfo = threadPool.info(threadPoolName);