Change

Author: Tim-Brooks

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

@@ -353,7 +353,11 @@ 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
-                        runOrDispatch(executor, originalThread, runnable);
+                        if (originalThread == Thread.currentThread()) {
+                            runnable.run();
+                        } else {
+                            executor.execute(runnable);
+                        }
                     }
                 }
             },
@@ -362,17 +366,6 @@ 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,26 +385,19 @@ protected void createMissingIndicesAndIndexData(
         Map<String, Exception> indicesExceptions = new ConcurrentHashMap<>();
         Map<String, Exception> dataStreamExceptions = new ConcurrentHashMap<>();
         Map<String, Exception> failureStoreExceptions = new ConcurrentHashMap<>();
-
-        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);
-        };
+        Runnable executeBulkRunnable = () -> executor.execute(new ActionRunnable<>(listener) {
+            @Override
+            protected void doRun() {
+                failRequestsWhenPrerequisiteActionFailed(
+                    indicesExceptions,
+                    dataStreamExceptions,
+                    failureStoreExceptions,
+                    bulkRequest,
+                    responses
+                );
+                executeBulk(task, bulkRequest, startTimeNanos, listener, executor, responses);
+            }
+        });
         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

@@ -436,7 +436,9 @@ public void testDispatchesToWriteCoordinationThreadPoolOnce() {
         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));
+        // Will increment twice because it will dispatch on the first coordination attempt. And then dispatch a second time after the index
+        // is created.
+        assertThat(stats.completed(), equalTo(2L));
     }
 
     public void testRejectCoordination() {