Fork CCS search-shards handling (#98209)

Similar to #98124, this action involves doing potentially O(#shards) work on both sender and receiver so it'd be best to avoid the transport worker.
elastic · Aug 8, 2023 · cd48337 · cd48337
1 parent 847ec45
commit cd48337
Show file tree

Hide file tree

Showing 4 changed files with 12 additions and 9 deletions.
diff --git a/docs/changelog/98209.yaml b/docs/changelog/98209.yaml
@@ -0,0 +1,5 @@
+pr: 98209
+summary: Fork CCS search-shards handling
+area: Search
+type: bug
+issues: []
diff --git a/...ava/org/elasticsearch/action/admin/cluster/shards/TransportClusterSearchShardsAction.java b/...ava/org/elasticsearch/action/admin/cluster/shards/TransportClusterSearchShardsAction.java
@@ -57,7 +57,7 @@ public TransportClusterSearchShardsAction(
             ClusterSearchShardsRequest::new,
             indexNameExpressionResolver,
             ClusterSearchShardsResponse::new,
-            ThreadPool.Names.SAME
+            ThreadPool.Names.SEARCH_COORDINATION
         );
         this.indicesService = indicesService;
     }

diff --git a/server/src/main/java/org/elasticsearch/action/search/TransportSearchAction.java b/server/src/main/java/org/elasticsearch/action/search/TransportSearchAction.java
@@ -74,7 +74,6 @@
 import org.elasticsearch.transport.RemoteTransportException;
 import org.elasticsearch.transport.Transport;
 import org.elasticsearch.transport.TransportRequestOptions;
-import org.elasticsearch.transport.TransportResponseHandler;
 import org.elasticsearch.transport.TransportService;
 
 import java.util.ArrayList;
@@ -663,6 +662,7 @@ static void collectSearchShards(
                 ) {
                     @Override
                     void innerOnResponse(SearchShardsResponse searchShardsResponse) {
+                        assert ThreadPool.assertCurrentThreadPool(ThreadPool.Names.SEARCH_COORDINATION);
                         searchShardsResponses.put(clusterAlias, searchShardsResponse);
                     }
 
@@ -676,6 +676,7 @@ Map<String, SearchShardsResponse> createFinalResponse() {
                 skipUnavailable == false,
                 ActionListener.wrap(connection -> {
                     final String[] indices = entry.getValue().indices();
+                    final Executor responseExecutor = transportService.getThreadPool().executor(ThreadPool.Names.SEARCH_COORDINATION);
                     // TODO: support point-in-time
                     if (searchContext == null && connection.getTransportVersion().onOrAfter(TransportVersion.V_8_500_010)) {
                         SearchShardsRequest searchShardsRequest = new SearchShardsRequest(
@@ -692,11 +693,7 @@ Map<String, SearchShardsResponse> createFinalResponse() {
                             SearchShardsAction.NAME,
                             searchShardsRequest,
                             TransportRequestOptions.EMPTY,
-                            new ActionListenerResponseHandler<>(
-                                singleListener,
-                                SearchShardsResponse::new,
-                                TransportResponseHandler.TRANSPORT_WORKER
-                            )
+                            new ActionListenerResponseHandler<>(singleListener, SearchShardsResponse::new, responseExecutor)
                         );
                     } else {
                         ClusterSearchShardsRequest searchShardsRequest = new ClusterSearchShardsRequest(indices).indicesOptions(
@@ -710,7 +707,7 @@ Map<String, SearchShardsResponse> createFinalResponse() {
                             new ActionListenerResponseHandler<>(
                                 singleListener.map(SearchShardsResponse::fromLegacyResponse),
                                 ClusterSearchShardsResponse::new,
-                                TransportResponseHandler.TRANSPORT_WORKER
+                                responseExecutor
                             )
                         );
                     }

diff --git a/server/src/main/java/org/elasticsearch/action/search/TransportSearchShardsAction.java b/server/src/main/java/org/elasticsearch/action/search/TransportSearchShardsAction.java
@@ -60,7 +60,7 @@ public TransportSearchShardsAction(
         SearchTransportService searchTransportService,
         IndexNameExpressionResolver indexNameExpressionResolver
     ) {
-        super(SearchShardsAction.NAME, transportService, actionFilters, SearchShardsRequest::new);
+        super(SearchShardsAction.NAME, transportService, actionFilters, SearchShardsRequest::new, ThreadPool.Names.SEARCH_COORDINATION);
         this.transportService = transportService;
         this.transportSearchAction = transportSearchAction;
         this.searchService = searchService;
@@ -73,6 +73,7 @@ public TransportSearchShardsAction(
 
     @Override
     protected void doExecute(Task task, SearchShardsRequest searchShardsRequest, ActionListener<SearchShardsResponse> listener) {
+        assert ThreadPool.assertCurrentThreadPool(ThreadPool.Names.SEARCH_COORDINATION);
         final long relativeStartNanos = System.nanoTime();
         SearchRequest original = new SearchRequest(searchShardsRequest.indices()).indicesOptions(searchShardsRequest.indicesOptions())
             .routing(searchShardsRequest.routing())