Fork TransportSearchAction to search coordination pool

We can run into situations where this action releases a store reference
which can block for up to seconds -> fork to search coordination to make
absolutely sure this won't cause transport threads to get blocked.

Author: original-brownbear

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

@@ -14,6 +14,7 @@
 import org.elasticsearch.TransportVersions;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionListenerResponseHandler;
+import org.elasticsearch.action.ActionRunnable;
 import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.IndicesRequest;
 import org.elasticsearch.action.OriginalIndices;
@@ -137,6 +138,7 @@ public class TransportSearchAction extends HandledTransportAction<SearchRequest,
     );
 
     private final ThreadPool threadPool;
+    private final Executor searchCoordinationExecutor;
     private final ClusterService clusterService;
     private final TransportService transportService;
     private final SearchTransportService searchTransportService;
@@ -169,6 +171,7 @@ public TransportSearchAction(
     ) {
         super(TYPE.name(), transportService, actionFilters, SearchRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.threadPool = threadPool;
+        this.searchCoordinationExecutor = threadPool.executor(ThreadPool.Names.SEARCH_COORDINATION);
         this.circuitBreaker = circuitBreakerService.getBreaker(CircuitBreaker.REQUEST);
         this.searchPhaseController = searchPhaseController;
         this.searchTransportService = searchTransportService;
@@ -288,6 +291,11 @@ public long buildTookInMillis() {
 
     @Override
     protected void doExecute(Task task, SearchRequest searchRequest, ActionListener<SearchResponse> listener) {
+        // workaround for https://github.com/elastic/elasticsearch/issues/97916 - TODO remove this when we can
+        searchCoordinationExecutor.execute(ActionRunnable.wrap(listener, l -> doExecuteForked((SearchTask) task, searchRequest, l)));
+    }
+
+    private void doExecuteForked(SearchTask task, SearchRequest searchRequest, ActionListener<SearchResponse> listener) {
         ActionListener<SearchResponse> loggingAndMetrics = listener.delegateFailureAndWrap((l, searchResponse) -> {
             searchResponseMetrics.recordTookTime(searchResponse.getTookInMillis());
             if (searchResponse.getShardFailures() != null && searchResponse.getShardFailures().length > 0) {
@@ -306,7 +314,7 @@ protected void doExecute(Task task, SearchRequest searchRequest, ActionListener<
             }
             l.onResponse(searchResponse);
         });
-        executeRequest((SearchTask) task, searchRequest, loggingAndMetrics, AsyncSearchActionProvider::new);
+        executeRequest(task, searchRequest, loggingAndMetrics, AsyncSearchActionProvider::new);
     }
 
     void executeRequest(

server/src/main/java/org/elasticsearch/env/ShardLock.java

@@ -9,6 +9,7 @@
 package org.elasticsearch.env;
 
 import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.transport.Transports;
 
 import java.io.Closeable;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -38,6 +39,7 @@ public final ShardId getShardId() {
 
     @Override
     public final void close() {
+        assert Transports.assertNotTransportThread("shard lock should not have been acquired on a transport thread");
         if (this.closed.compareAndSet(false, true)) {
             closeInternal();
         }

server/src/main/java/org/elasticsearch/index/engine/Engine.java

@@ -763,6 +763,7 @@ public Searcher acquireSearcherInternal(String source) {
 
                 @Override
                 protected void doClose() {
+                    assert Transports.assertNotTransportThread("releasing from the reference manager and/or store may block");
                     try {
                         referenceManager.release(acquire);
                     } catch (IOException e) {

server/src/main/java/org/elasticsearch/index/store/Store.java

@@ -67,6 +67,7 @@
 import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.translog.Translog;
+import org.elasticsearch.transport.Transports;
 
 import java.io.Closeable;
 import java.io.EOFException;
@@ -431,6 +432,7 @@ public boolean isClosing() {
     }
 
     private void closeInternal() {
+        assert Transports.assertNotTransportThread("store closing must not happen on the transport thread");
         // Leverage try-with-resources to close the shard lock for us
         try (Closeable c = shardLock) {
             try {