Cleanup listener wrapping in TransportSearchAction a little (#116184)

Lets avoid parsing settings during search request processing,
remove unused field from the listener and using the delegating listener
class.

Author: original-brownbear

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

@@ -16,6 +16,7 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionListenerResponseHandler;
 import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.DelegatingActionListener;
 import org.elasticsearch.action.IndicesRequest;
 import org.elasticsearch.action.OriginalIndices;
 import org.elasticsearch.action.RemoteClusterActionType;
@@ -52,7 +53,6 @@
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Setting.Property;
-import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.ArrayUtils;
 import org.elasticsearch.common.util.CollectionUtils;
 import org.elasticsearch.common.util.Maps;
@@ -160,7 +160,7 @@ public class TransportSearchAction extends HandledTransportAction<SearchRequest,
     private final SearchResponseMetrics searchResponseMetrics;
     private final Client client;
     private final UsageService usageService;
-    private final Settings settings;
+    private final boolean collectTelemetry;
 
     @Inject
     public TransportSearchAction(
@@ -193,9 +193,10 @@ public TransportSearchAction(
         this.indexNameExpressionResolver = indexNameExpressionResolver;
         this.namedWriteableRegistry = namedWriteableRegistry;
         this.executorSelector = executorSelector;
-        this.settings = clusterService.getSettings();
+        var settings = clusterService.getSettings();
         this.defaultPreFilterShardSize = DEFAULT_PRE_FILTER_SHARD_SIZE.get(settings);
         this.ccsCheckCompatibility = SearchService.CCS_VERSION_CHECK_SETTING.get(settings);
+        this.collectTelemetry = SearchService.CCS_COLLECT_TELEMETRY.get(settings);
         this.searchResponseMetrics = searchResponseMetrics;
         this.client = client;
         this.usageService = usageService;
@@ -313,12 +314,7 @@ public long buildTookInMillis() {
 
     @Override
     protected void doExecute(Task task, SearchRequest searchRequest, ActionListener<SearchResponse> listener) {
-        executeRequest(
-            (SearchTask) task,
-            searchRequest,
-            new SearchResponseActionListener((SearchTask) task, listener),
-            AsyncSearchActionProvider::new
-        );
+        executeRequest((SearchTask) task, searchRequest, new SearchResponseActionListener(listener), AsyncSearchActionProvider::new);
     }
 
     void executeRequest(
@@ -372,7 +368,7 @@ void executeRequest(
                     searchPhaseProvider.apply(delegate)
                 );
             } else {
-                if (listener instanceof TelemetryListener tl) {
+                if (delegate instanceof TelemetryListener tl) {
                     tl.setRemotes(resolvedIndices.getRemoteClusterIndices().size());
                     if (task.isAsync()) {
                         tl.setFeature(CCSUsageTelemetry.ASYNC_FEATURE);
@@ -398,7 +394,7 @@ void executeRequest(
                 }
                 final TaskId parentTaskId = task.taskInfo(clusterService.localNode().getId(), false).taskId();
                 if (shouldMinimizeRoundtrips(rewritten)) {
-                    if (listener instanceof TelemetryListener tl) {
+                    if (delegate instanceof TelemetryListener tl) {
                         tl.setFeature(CCSUsageTelemetry.MRT_FEATURE);
                     }
                     final AggregationReduceContext.Builder aggregationReduceContextBuilder = rewritten.source() != null
@@ -508,21 +504,21 @@ void executeRequest(
                 // We set the keep alive to -1 to indicate that we don't need the pit id in the response.
                 // This is needed since we delete the pit prior to sending the response so the id doesn't exist anymore.
                 source.pointInTimeBuilder(new PointInTimeBuilder(resp.getPointInTimeId()).setKeepAlive(TimeValue.MINUS_ONE));
-                var pitListener = new SearchResponseActionListener(task, listener) {
+                var pitListener = new SearchResponseActionListener(delegate) {
                     @Override
                     public void onResponse(SearchResponse response) {
                         // we need to close the PIT first so we delay the release of the response to after the closing
                         response.incRef();
                         closePIT(
                             client,
                             original.source().pointInTimeBuilder(),
-                            () -> ActionListener.respondAndRelease(listener, response)
+                            () -> ActionListener.respondAndRelease(delegate, response)
                         );
                     }
 
                     @Override
                     public void onFailure(Exception e) {
-                        closePIT(client, original.source().pointInTimeBuilder(), () -> listener.onFailure(e));
+                        closePIT(client, original.source().pointInTimeBuilder(), () -> delegate.onFailure(e));
                     }
                 };
                 executeRequest(task, original, pitListener, searchPhaseProvider);
@@ -1874,14 +1870,13 @@ private interface TelemetryListener {
         void setClient(String client);
     }
 
-    private class SearchResponseActionListener implements ActionListener<SearchResponse>, TelemetryListener {
-        private final SearchTask task;
-        private final ActionListener<SearchResponse> listener;
+    private class SearchResponseActionListener extends DelegatingActionListener<SearchResponse, SearchResponse>
+        implements
+            TelemetryListener {
         private final CCSUsage.Builder usageBuilder;
 
-        SearchResponseActionListener(SearchTask task, ActionListener<SearchResponse> listener) {
-            this.task = task;
-            this.listener = listener;
+        SearchResponseActionListener(ActionListener<SearchResponse> listener) {
+            super(listener);
             if (listener instanceof SearchResponseActionListener srListener) {
                 usageBuilder = srListener.usageBuilder;
             } else {
@@ -1893,7 +1888,7 @@ private class SearchResponseActionListener implements ActionListener<SearchRespo
          * Should we collect telemetry for this search?
          */
         private boolean collectTelemetry() {
-            return SearchService.CCS_COLLECT_TELEMETRY.get(settings) && usageBuilder.getRemotesCount() > 0;
+            return collectTelemetry && usageBuilder.getRemotesCount() > 0;
         }
 
         public void setRemotes(int count) {
@@ -1942,7 +1937,7 @@ public void onResponse(SearchResponse searchResponse) {
                 return;
             }
             // This is last because we want to collect telemetry before returning the response.
-            listener.onResponse(searchResponse);
+            delegate.onResponse(searchResponse);
         }
 
         @Override
@@ -1952,7 +1947,7 @@ public void onFailure(Exception e) {
                 usageBuilder.setFailure(e);
                 recordTelemetry();
             }
-            listener.onFailure(e);
+            super.onFailure(e);
         }
 
         private void recordTelemetry() {