Fork response reading in TransportNodesAction (#97899)

All `TransportNodesAction` implementations fork before computing the final
response, but in some cases (notably `TransportNodesStatsAction`) even reading
the node-level responses is too expensive to run on a `transport_worker`
thread. With this commit we fork the deserialization of the node-level
responses to the final executor too.

docs/changelog/97899.yaml

@@ -0,0 +1,5 @@
+pr: 97899
+summary: Fork response reading in `TransportNodesAction`
+area: Distributed
+type: bug
+issues: []

server/src/main/java/org/elasticsearch/action/admin/cluster/node/stats/TransportNodesStatsAction.java

@@ -23,6 +23,7 @@ import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportRequest;
 import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.transport.Transports;
 
 import java.io.IOException;
 import java.util.List;
@@ -70,6 +71,7 @@ public class TransportNodesStatsAction extends TransportNodesAction<
 
     @Override
     protected NodeStats newNodeResponse(StreamInput in, DiscoveryNode node) throws IOException {
+        assert Transports.assertNotTransportThread("deserializing node stats is too expensive for a transport thread");
         return new NodeStats(in);
     }
 

server/src/main/java/org/elasticsearch/action/support/nodes/TransportNodesAction.java

@@ -110,7 +110,7 @@ public abstract class TransportNodesAction<
                     transportNodeAction,
                     nodeRequest,
                     transportRequestOptions,
-                    new ActionListenerResponseHandler<>(listener, nodeResponseReader(discoveryNode))
+                    new ActionListenerResponseHandler<>(listener, nodeResponseReader(discoveryNode), finalExecutor)
                 );
             }