Factor out a ChannelActionListener (#33819)

We use similar / same concepts in SerachTransportService and HandledTransportAction but both
duplicate the efforts with slightly different implementation details. This streamlines
sending responses / exceptions back to a channel in an ActionListener with appropriate logging.

Author: s1monw

server/src/main/java/org/elasticsearch/action/search/SearchTransportService.java

@@ -23,6 +23,7 @@
 import org.elasticsearch.action.ActionListenerResponseHandler;
 import org.elasticsearch.action.IndicesRequest;
 import org.elasticsearch.action.OriginalIndices;
+import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.component.AbstractComponent;
@@ -348,25 +349,8 @@ public void onFailure(Exception e) {
 
         transportService.registerRequestHandler(QUERY_ACTION_NAME, ThreadPool.Names.SAME, ShardSearchTransportRequest::new,
             (request, channel, task) -> {
-                searchService.executeQueryPhase(request, (SearchTask) task, new ActionListener<SearchPhaseResult>() {
-                    @Override
-                    public void onResponse(SearchPhaseResult searchPhaseResult) {
-                        try {
-                            channel.sendResponse(searchPhaseResult);
-                        } catch (IOException e) {
-                            throw new UncheckedIOException(e);
-                        }
-                    }
-
-                    @Override
-                    public void onFailure(Exception e) {
-                        try {
-                            channel.sendResponse(e);
-                        } catch (IOException e1) {
-                            throw new UncheckedIOException(e1);
-                        }
-                    }
-                });
+                searchService.executeQueryPhase(request, (SearchTask) task, new HandledTransportAction.ChannelActionListener<>(
+                    channel, QUERY_ACTION_NAME, request));
             });
         TransportActionProxy.registerProxyAction(transportService, QUERY_ACTION_NAME,
                 (request) -> ((ShardSearchRequest)request).numberOfShards() == 1 ? QueryFetchSearchResult::new : QuerySearchResult::new);

server/src/main/java/org/elasticsearch/action/support/HandledTransportAction.java

@@ -18,6 +18,8 @@
  */
 package org.elasticsearch.action.support;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequest;
@@ -27,7 +29,9 @@
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportChannel;
+import org.elasticsearch.transport.TransportRequest;
 import org.elasticsearch.transport.TransportRequestHandler;
+import org.elasticsearch.transport.TransportResponse;
 import org.elasticsearch.transport.TransportService;
 
 import java.util.function.Supplier;
@@ -63,30 +67,44 @@ protected HandledTransportAction(Settings settings, String actionName, boolean c
     }
 
     class TransportHandler implements TransportRequestHandler<Request> {
-
         @Override
         public final void messageReceived(final Request request, final TransportChannel channel, Task task) throws Exception {
             // We already got the task created on the network layer - no need to create it again on the transport layer
-            execute(task, request, new ActionListener<Response>() {
-                @Override
-                public void onResponse(Response response) {
-                    try {
-                        channel.sendResponse(response);
-                    } catch (Exception e) {
-                        onFailure(e);
-                    }
-                }
+            Logger logger = HandledTransportAction.this.logger;
+            execute(task, request, new ChannelActionListener<>(channel, actionName, request));
+        }
+    }
+
+    public static final class ChannelActionListener<Response extends TransportResponse, Request extends TransportRequest> implements
+        ActionListener<Response> {
+        private final Logger logger = LogManager.getLogger(getClass());
+        private final TransportChannel channel;
+        private final Request request;
+        private final String actionName;
+
+        public ChannelActionListener(TransportChannel channel, String actionName, Request request) {
+            this.channel = channel;
+            this.request = request;
+            this.actionName = actionName;
+        }
 
-                @Override
-                public void onFailure(Exception e) {
-                    try {
-                        channel.sendResponse(e);
-                    } catch (Exception e1) {
-                        logger.warn(() -> new ParameterizedMessage(
-                                    "Failed to send error response for action [{}] and request [{}]", actionName, request), e1);
-                    }
-                }
-            });
+        @Override
+        public void onResponse(Response response) {
+            try {
+                channel.sendResponse(response);
+            } catch (Exception e) {
+                onFailure(e);
+            }
+        }
+
+        @Override
+        public void onFailure(Exception e) {
+            try {
+                channel.sendResponse(e);
+            } catch (Exception e1) {
+                logger.warn(() -> new ParameterizedMessage(
+                    "Failed to send error response for action [{}] and request [{}]", actionName, request), e1);
+            }
         }
     }