Pass Executor into request handlers (#98854)

DiannaHohensee · web-flow · commit 6caf9c1f60ba · 2023-08-29T08:17:50.000-04:00
Pass an Executor into RequestHandleRegistry#registerRequestHandler calls instead of a String. Ultimately the String executor name passed through and stashed internally is converted by the ThreadPool into an Executor on which to execute tasks. Part of the changes for #97879.
diff --git a/server/src/internalClusterTest/java/org/elasticsearch/action/IndicesRequestIT.java b/server/src/internalClusterTest/java/org/elasticsearch/action/IndicesRequestIT.java
@@ -98,6 +98,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.Executor;
 import java.util.function.Function;
 
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
@@ -754,7 +755,7 @@ public List<TransportInterceptor> getTransportInterceptors(
         @Override
         public <T extends TransportRequest> TransportRequestHandler<T> interceptHandler(
             String action,
-            String executor,
+            Executor executor,
             boolean forceExecution,
             TransportRequestHandler<T> actualHandler
         ) {
diff --git a/server/src/main/java/org/elasticsearch/common/network/NetworkModule.java b/server/src/main/java/org/elasticsearch/common/network/NetworkModule.java
@@ -49,6 +49,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.Executor;
 import java.util.function.BiConsumer;
 import java.util.function.Supplier;
 
@@ -262,7 +263,7 @@ private CompositeTransportInterceptor(List<TransportInterceptor> transportInterc
         @Override
         public <T extends TransportRequest> TransportRequestHandler<T> interceptHandler(
             String action,
-            String executor,
+            Executor executor,
             boolean forceExecution,
             TransportRequestHandler<T> actualHandler
         ) {
diff --git a/server/src/main/java/org/elasticsearch/transport/InboundHandler.java b/server/src/main/java/org/elasticsearch/transport/InboundHandler.java
@@ -295,7 +295,7 @@ private <T extends TransportRequest> void handleRequestForking(T request, Reques
         boolean success = false;
         request.incRef();
         try {
-            threadPool.executor(reg.getExecutor()).execute(threadPool.getThreadContext().preserveContextWithTracing(new AbstractRunnable() {
+            reg.getExecutor().execute(threadPool.getThreadContext().preserveContextWithTracing(new AbstractRunnable() {
                 @Override
                 protected void doRun() {
                     doHandleRequest(reg, request, channel);
diff --git a/server/src/main/java/org/elasticsearch/transport/RequestHandlerRegistry.java b/server/src/main/java/org/elasticsearch/transport/RequestHandlerRegistry.java
@@ -18,6 +18,7 @@
 import org.elasticsearch.tracing.Tracer;
 
 import java.io.IOException;
+import java.util.concurrent.Executor;
 
 import static org.elasticsearch.core.Releasables.assertOnce;
 
@@ -27,7 +28,7 @@ public class RequestHandlerRegistry<Request extends TransportRequest> implements
     private final TransportRequestHandler<Request> handler;
     private final boolean forceExecution;
     private final boolean canTripCircuitBreaker;
-    private final String executor;
+    private final Executor executor;
     private final TaskManager taskManager;
     private final Tracer tracer;
     private final Writeable.Reader<Request> requestReader;
@@ -38,7 +39,7 @@ public RequestHandlerRegistry(
         Writeable.Reader<Request> requestReader,
         TaskManager taskManager,
         TransportRequestHandler<Request> handler,
-        String executor,
+        Executor executor,
         boolean forceExecution,
         boolean canTripCircuitBreaker,
         Tracer tracer
@@ -86,7 +87,7 @@ public boolean canTripCircuitBreaker() {
         return canTripCircuitBreaker;
     }
 
-    public String getExecutor() {
+    public Executor getExecutor() {
         return executor;
     }
 
diff --git a/server/src/main/java/org/elasticsearch/transport/TransportInterceptor.java b/server/src/main/java/org/elasticsearch/transport/TransportInterceptor.java
@@ -11,19 +11,21 @@
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.io.stream.Writeable.Reader;
 
+import java.util.concurrent.Executor;
+
 /**
  * This interface allows plugins to intercept requests on both the sender and the receiver side.
  */
 public interface TransportInterceptor {
     /**
      * This is called for each handler that is registered via
-     * {@link TransportService#registerRequestHandler(String, String, boolean, boolean, Reader, TransportRequestHandler)} or
-     * {@link TransportService#registerRequestHandler(String, String, Reader, TransportRequestHandler)}. The returned handler is
+     * {@link TransportService#registerRequestHandler(String, Executor, boolean, boolean, Reader, TransportRequestHandler)} or
+     * {@link TransportService#registerRequestHandler(String, Executor, Reader, TransportRequestHandler)}. The returned handler is
      * used instead of the passed in handler. By default the provided handler is returned.
      */
     default <T extends TransportRequest> TransportRequestHandler<T> interceptHandler(
         String action,
-        String executor,
+        Executor executor,
         boolean forceExecution,
         TransportRequestHandler<T> actualHandler
     ) {
diff --git a/server/src/main/java/org/elasticsearch/transport/TransportService.java b/server/src/main/java/org/elasticsearch/transport/TransportService.java
@@ -293,7 +293,7 @@ public TransportService(
         }
         registerRequestHandler(
             HANDSHAKE_ACTION_NAME,
-            ThreadPool.Names.SAME,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE,
             false,
             false,
             HandshakeRequest::new,
@@ -1004,8 +1004,8 @@ private void sendLocalRequest(long requestId, final String action, final Transpo
                 assert false : action;
                 throw new ActionNotFoundTransportException("Action [" + action + "] not found");
             }
-            final String executor = reg.getExecutor();
-            if (ThreadPool.Names.SAME.equals(executor)) {
+            final Executor executor = reg.getExecutor();
+            if (executor == EsExecutors.DIRECT_EXECUTOR_SERVICE) {
                 try (var ignored = threadPool.getThreadContext().newTraceContext()) {
                     try {
                         reg.processMessageReceived(request, channel);
@@ -1017,7 +1017,7 @@ private void sendLocalRequest(long requestId, final String action, final Transpo
                 boolean success = false;
                 request.incRef();
                 try {
-                    threadPool.executor(executor).execute(threadPool.getThreadContext().preserveContextWithTracing(new AbstractRunnable() {
+                    executor.execute(threadPool.getThreadContext().preserveContextWithTracing(new AbstractRunnable() {
                         @Override
                         protected void doRun() throws Exception {
                             reg.processMessageReceived(request, channel);
@@ -1122,6 +1122,46 @@ public static boolean isValidActionName(String actionName) {
         return false;
     }
 
+    /**
+     * Temporary passthrough function that continues to take a String rather than Executor type.
+     *
+     * @param action
+     * @param executor
+     * @param requestReader
+     * @param handler
+     * @param <Request>
+     */
+    public <Request extends TransportRequest> void registerRequestHandler(
+        String action,
+        String executor,
+        Writeable.Reader<Request> requestReader,
+        TransportRequestHandler<Request> handler
+    ) {
+        registerRequestHandler(action, threadPool.executor(executor), requestReader, handler);
+    }
+
+    /**
+     * Temporary passthrough function that continues to take a String rather than Executor type.
+     *
+     * @param action
+     * @param executor
+     * @param forceExecution
+     * @param canTripCircuitBreaker
+     * @param requestReader
+     * @param handler
+     * @param <Request>
+     */
+    public <Request extends TransportRequest> void registerRequestHandler(
+        String action,
+        String executor,
+        boolean forceExecution,
+        boolean canTripCircuitBreaker,
+        Writeable.Reader<Request> requestReader,
+        TransportRequestHandler<Request> handler
+    ) {
+        registerRequestHandler(action, threadPool.executor(executor), forceExecution, canTripCircuitBreaker, requestReader, handler);
+    }
+
     /**
      * Registers a new request handler
      *
@@ -1132,7 +1172,7 @@ public static boolean isValidActionName(String actionName) {
      */
     public <Request extends TransportRequest> void registerRequestHandler(
         String action,
-        String executor,
+        Executor executor,
         Writeable.Reader<Request> requestReader,
         TransportRequestHandler<Request> handler
     ) {
@@ -1163,7 +1203,7 @@ public <Request extends TransportRequest> void registerRequestHandler(
      */
     public <Request extends TransportRequest> void registerRequestHandler(
         String action,
-        String executor,
+        Executor executor,
         boolean forceExecution,
         boolean canTripCircuitBreaker,
         Writeable.Reader<Request> requestReader,
diff --git a/server/src/test/java/org/elasticsearch/common/network/NetworkModuleTests.java b/server/src/test/java/org/elasticsearch/common/network/NetworkModuleTests.java
@@ -38,6 +38,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiConsumer;
@@ -237,7 +238,7 @@ public void testRegisterInterceptor() {
             @Override
             public <T extends TransportRequest> TransportRequestHandler<T> interceptHandler(
                 String action,
-                String executor,
+                Executor executor,
                 boolean forceExecution,
                 TransportRequestHandler<T> actualHandler
             ) {
diff --git a/server/src/test/java/org/elasticsearch/transport/InboundHandlerTests.java b/server/src/test/java/org/elasticsearch/transport/InboundHandlerTests.java
@@ -26,6 +26,7 @@
 import org.elasticsearch.common.network.HandlingTimeTracker;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.PageCacheRecycler;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.tasks.TaskId;
@@ -111,7 +112,7 @@ public void testPing() throws Exception {
             TestRequest::new,
             taskManager,
             (request, channel, task) -> channelCaptor.set(channel),
-            ThreadPool.Names.SAME,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE,
             false,
             true,
             Tracer.NOOP
@@ -164,7 +165,7 @@ public TestResponse read(StreamInput in) throws IOException {
                 channelCaptor.set(channel);
                 requestCaptor.set(request);
             },
-            ThreadPool.Names.SAME,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE,
             false,
             true,
             Tracer.NOOP
diff --git a/server/src/test/java/org/elasticsearch/transport/TransportServiceHandshakeTests.java b/server/src/test/java/org/elasticsearch/transport/TransportServiceHandshakeTests.java
@@ -40,6 +40,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 
 import static java.util.Collections.emptyMap;
@@ -426,7 +427,7 @@ public void setModifyBuildHash(boolean modifyBuildHash) {
         @Override
         public <T extends TransportRequest> TransportRequestHandler<T> interceptHandler(
             String action,
-            String executor,
+            Executor executor,
             boolean forceExecution,
             TransportRequestHandler<T> actualHandler
         ) {
diff --git a/test/framework/src/main/java/org/elasticsearch/test/ESIntegTestCase.java b/test/framework/src/main/java/org/elasticsearch/test/ESIntegTestCase.java
@@ -174,6 +174,7 @@
 import java.util.concurrent.Callable;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
@@ -2174,7 +2175,7 @@ public List<TransportInterceptor> getTransportInterceptors(
                 @Override
                 public <T extends TransportRequest> TransportRequestHandler<T> interceptHandler(
                     String action,
-                    String executor,
+                    Executor executor,
                     boolean forceExecution,
                     TransportRequestHandler<T> actualHandler
                 ) {
diff --git a/x-pack/plugin/security/src/internalClusterTest/java/org/elasticsearch/transport/SecurityServerTransportServiceTests.java b/x-pack/plugin/security/src/internalClusterTest/java/org/elasticsearch/transport/SecurityServerTransportServiceTests.java
@@ -22,13 +22,7 @@ public void testSecurityServerTransportServiceWrapsAllHandlers() {
                     + SecurityServerTransportInterceptor.ProfileSecuredRequestHandler.class
                     + "; do all the handler registration methods have overrides?",
                 handler.toString(),
-                "ProfileSecuredRequestHandler{action='"
-                    + handler.getAction()
-                    + "', executorName='"
-                    + handler.getExecutor()
-                    + "', forceExecution="
-                    + handler.isForceExecution()
-                    + "}"
+                "ProfileSecuredRequestHandler{action='" + handler.getAction() + "', forceExecution=" + handler.isForceExecution() + "}"
             );
         }
     }
diff --git a/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/Security.java b/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/Security.java
@@ -389,6 +389,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.concurrent.Executor;
 import java.util.function.BiConsumer;
 import java.util.function.Function;
 import java.util.function.Predicate;
@@ -1578,7 +1579,7 @@ public List<TransportInterceptor> getTransportInterceptors(NamedWriteableRegistr
             @Override
             public <T extends TransportRequest> TransportRequestHandler<T> interceptHandler(
                 String action,
-                String executor,
+                Executor executor,
                 boolean forceExecution,
                 TransportRequestHandler<T> actualHandler
             ) {
diff --git a/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/transport/SecurityServerTransportInterceptor.java b/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/transport/SecurityServerTransportInterceptor.java
@@ -16,6 +16,7 @@
 import org.elasticsearch.common.ssl.SslConfiguration;
 import org.elasticsearch.common.util.Maps;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.common.util.concurrent.RunOnce;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.license.LicenseUtils;
@@ -55,6 +56,7 @@
 import java.util.Collections;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.Executor;
 import java.util.function.Function;
 
 import static org.elasticsearch.core.Strings.format;
@@ -458,7 +460,7 @@ void assertNoAuthentication(String action) {
     @Override
     public <T extends TransportRequest> TransportRequestHandler<T> interceptHandler(
         String action,
-        String executor,
+        Executor executor,
         boolean forceExecution,
         TransportRequestHandler<T> actualHandler
     ) {
@@ -515,7 +517,7 @@ public static class ProfileSecuredRequestHandler<T extends TransportRequest> imp
         private final TransportRequestHandler<T> handler;
         private final Map<String, ServerTransportFilter> profileFilters;
         private final ThreadContext threadContext;
-        private final String executorName;
+        private final Executor executor;
         private final ThreadPool threadPool;
         private final boolean forceExecution;
         private final Logger logger;
@@ -524,14 +526,14 @@ public static class ProfileSecuredRequestHandler<T extends TransportRequest> imp
             Logger logger,
             String action,
             boolean forceExecution,
-            String executorName,
+            Executor executor,
             TransportRequestHandler<T> handler,
             Map<String, ServerTransportFilter> profileFilters,
             ThreadPool threadPool
         ) {
             this.logger = logger;
             this.action = action;
-            this.executorName = executorName;
+            this.executor = executor;
             this.handler = handler;
             this.profileFilters = profileFilters;
             this.threadContext = threadPool.getThreadContext();
@@ -572,16 +574,7 @@ public void onAfter() {
 
         @Override
         public String toString() {
-            return "ProfileSecuredRequestHandler{"
-                + "action='"
-                + action
-                + '\''
-                + ", executorName='"
-                + executorName
-                + '\''
-                + ", forceExecution="
-                + forceExecution
-                + '}';
+            return "ProfileSecuredRequestHandler{" + "action='" + action + '\'' + ", forceExecution=" + forceExecution + '}';
         }
 
         @Override
@@ -602,7 +595,7 @@ public void messageReceived(T request, TransportChannel channel, Task task) {
 
                 final AbstractRunnable receiveMessage = getReceiveRunnable(request, channel, task);
                 final ActionListener<Void> filterListener;
-                if (ThreadPool.Names.SAME.equals(executorName)) {
+                if (executor == EsExecutors.DIRECT_EXECUTOR_SERVICE) {
                     filterListener = new AbstractFilterListener(receiveMessage) {
                         @Override
                         public void onResponse(Void unused) {
@@ -625,7 +618,7 @@ public void onResponse(Void unused) {
                                 receiveMessage.run();
                             } else {
                                 try {
-                                    threadPool.executor(executorName).execute(receiveMessage);
+                                    executor.execute(receiveMessage);
                                 } catch (Exception e) {
                                     onFailure(e);
                                 }
diff --git a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/transport/SecurityServerTransportInterceptorTests.java b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/transport/SecurityServerTransportInterceptorTests.java
diff --git a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/transport/netty4/SecurityNetty4HeaderSizeLimitTests.java b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/transport/netty4/SecurityNetty4HeaderSizeLimitTests.java
diff --git a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/transport/netty4/SecurityNetty4ServerTransportAuthenticationTests.java b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/transport/netty4/SecurityNetty4ServerTransportAuthenticationTests.java
