Do not dispatch in IngestService

The ingest service dispatches to the provided executor. This work became
redundant several years ago when we moved the transport bulk action to
be dispatched from the transport threads for the action itself. With
this behavior, there is no reason to re-dispatch an ingest action.

Author: Tim-Brooks

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

@@ -347,8 +347,7 @@ public boolean isForceExecution() {
                         }
                     }
                 }
-            },
-            executor
+            }
         );
     }
 

server/src/main/java/org/elasticsearch/ingest/IngestService.java

@@ -95,7 +95,6 @@
 import java.util.Set;
 import java.util.TreeMap;
 import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.Executor;
 import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
 import java.util.function.Consumer;
@@ -826,7 +825,6 @@ private static IngestPipelinesExecutionResult failWithoutStoringIn(String index,
      * @param onCompletion A callback executed once all documents have been processed. Accepts the thread
      *                     that ingestion completed on or an exception in the event that the entire operation
      *                     has failed.
-     * @param executor Which executor the bulk request should be executed on.
      */
     public void executeBulkRequest(
         final ProjectId projectId,
@@ -836,15 +834,14 @@ public void executeBulkRequest(
         final Function<String, Boolean> resolveFailureStore,
         final TriConsumer<Integer, String, Exception> onStoreFailure,
         final TriConsumer<Integer, Exception, IndexDocFailureStoreStatus> onFailure,
-        final BiConsumer<Thread, Exception> onCompletion,
-        final Executor executor
+        final BiConsumer<Thread, Exception> onCompletion
     ) {
         assert numberOfActionRequests > 0 : "numberOfActionRequests must be greater than 0 but was [" + numberOfActionRequests + "]";
 
         // Adapt handler to ensure node features during ingest logic
         final Function<String, Boolean> adaptedResolveFailureStore = wrapResolverWithFeatureCheck(resolveFailureStore);
 
-        executor.execute(new AbstractRunnable() {
+        new AbstractRunnable() {
 
             @Override
             public void onFailure(Exception e) {
@@ -933,7 +930,7 @@ public void onFailure(Exception e) {
                     }
                 }
             }
-        });
+        }.run();
     }
 
     /**

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

@@ -85,7 +85,6 @@
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.ArgumentMatchers.same;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
@@ -425,8 +424,7 @@ public void testIngestLocal() throws Exception {
             redirectPredicate.capture(),
             redirectHandler.capture(),
             failureHandler.capture(),
-            completionHandler.capture(),
-            same(writeCoordinationExecutor)
+            completionHandler.capture()
         );
         completionHandler.getValue().accept(null, exception);
         assertTrue(failureCalled.get());
@@ -476,8 +474,7 @@ public void testSingleItemBulkActionIngestLocal() throws Exception {
             any(),
             any(),
             failureHandler.capture(),
-            completionHandler.capture(),
-            same(writeCoordinationExecutor)
+            completionHandler.capture()
         );
         completionHandler.getValue().accept(null, exception);
         assertTrue(failureCalled.get());
@@ -525,8 +522,7 @@ public void testIngestSystemLocal() throws Exception {
             any(),
             any(),
             failureHandler.capture(),
-            completionHandler.capture(),
-            same(systemWriteCoordinationExecutor)
+            completionHandler.capture()
         );
         completionHandler.getValue().accept(null, exception);
         assertTrue(failureCalled.get());
@@ -558,7 +554,7 @@ public void testIngestForward() throws Exception {
         ActionTestUtils.execute(action, null, bulkRequest, listener);
 
         // should not have executed ingest locally
-        verify(ingestService, never()).executeBulkRequest(eq(projectId), anyInt(), any(), any(), any(), any(), any(), any(), any());
+        verify(ingestService, never()).executeBulkRequest(eq(projectId), anyInt(), any(), any(), any(), any(), any(), any());
         // but instead should have sent to a remote node with the transport service
         ArgumentCaptor<DiscoveryNode> node = ArgumentCaptor.forClass(DiscoveryNode.class);
         verify(transportService).sendRequest(node.capture(), eq(TransportBulkAction.NAME), any(), remoteResponseHandler.capture());
@@ -598,7 +594,7 @@ public void testSingleItemBulkActionIngestForward() throws Exception {
         ActionTestUtils.execute(singleItemBulkWriteAction, null, indexRequest, listener);
 
         // should not have executed ingest locally
-        verify(ingestService, never()).executeBulkRequest(eq(projectId), anyInt(), any(), any(), any(), any(), any(), any(), any());
+        verify(ingestService, never()).executeBulkRequest(eq(projectId), anyInt(), any(), any(), any(), any(), any(), any());
         // but instead should have sent to a remote node with the transport service
         ArgumentCaptor<DiscoveryNode> node = ArgumentCaptor.forClass(DiscoveryNode.class);
         verify(transportService).sendRequest(node.capture(), eq(TransportBulkAction.NAME), any(), remoteResponseHandler.capture());
@@ -686,8 +682,7 @@ private void validatePipelineWithBulkUpsert(@Nullable String indexRequestIndexNa
             any(),
             any(),
             failureHandler.capture(),
-            completionHandler.capture(),
-            same(writeCoordinationExecutor)
+            completionHandler.capture()
         );
         assertEquals(indexRequest1.getPipeline(), "default_pipeline");
         assertEquals(indexRequest2.getPipeline(), "default_pipeline");
@@ -737,8 +732,7 @@ public void testDoExecuteCalledTwiceCorrectly() throws Exception {
             any(),
             any(),
             failureHandler.capture(),
-            completionHandler.capture(),
-            same(writeCoordinationExecutor)
+            completionHandler.capture()
         );
         completionHandler.getValue().accept(null, exception);
         assertFalse(action.indexCreated); // still no index yet, the ingest node failed.
@@ -831,8 +825,7 @@ public void testFindDefaultPipelineFromTemplateMatch() {
             any(),
             any(),
             failureHandler.capture(),
-            completionHandler.capture(),
-            same(writeCoordinationExecutor)
+            completionHandler.capture()
         );
     }
 
@@ -872,8 +865,7 @@ public void testFindDefaultPipelineFromV2TemplateMatch() {
             any(),
             any(),
             failureHandler.capture(),
-            completionHandler.capture(),
-            same(writeCoordinationExecutor)
+            completionHandler.capture()
         );
     }
 
@@ -902,8 +894,7 @@ public void testIngestCallbackExceptionHandled() throws Exception {
             any(),
             any(),
             failureHandler.capture(),
-            completionHandler.capture(),
-            same(writeCoordinationExecutor)
+            completionHandler.capture()
         );
         indexRequest1.autoGenerateId();
         completionHandler.getValue().accept(Thread.currentThread(), null);
@@ -942,8 +933,7 @@ private void validateDefaultPipeline(IndexRequest indexRequest) {
             any(),
             any(),
             failureHandler.capture(),
-            completionHandler.capture(),
-            same(writeCoordinationExecutor)
+            completionHandler.capture()
         );
         assertEquals(indexRequest.getPipeline(), "default_pipeline");
         completionHandler.getValue().accept(null, exception);