Three interrelated issues here:

- [ ] `TransportBroadcastByNodeAction` derivatives are typically executed via `NodeClient` which bypasses the `TransportService` and therefore does not respect (or even know about) the `executor` parameter which would fork. This means that the coordination work, including grouping all the shards by node, happens on the calling thread. When executed from the REST layer, that's a transport worker. This is #97916, but until that's fixed we need a workaround.

- [ ] The node-level responses are deserialised and processed on the receiving transport worker too.

- [ ] Several of these actions use `SAME` for their executor, bypassing the forking that _does_ exist today.

Relates #77466 
server/src/test/java/org/elasticsearch/action/support/broadcast/node/TransportBroadcastByNodeActionTests.java
}
        discoBuilder.localNodeId(newNode(0).getId());
        discoBuilder.masterNodeId(newNode(numberOfNodes - 1).getId());
        ClusterState.Builder stateBuilder = ClusterState.builder(new ClusterName(TEST_CLUSTER));
        stateBuilder.nodes(discoBuilder);
        final IndexMetadata.Builder indexMetadata = IndexMetadata.builder(TEST_INDEX)
            .settings(Settings.builder().put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT))
            .numberOfReplicas(0)
            .numberOfShards(totalIndexShards);
        stateBuilder.metadata(Metadata.builder().put(indexMetadata));
        stateBuilder.routingTable(RoutingTable.builder().add(indexRoutingTable.build()).build());
        ClusterState clusterState = stateBuilder.build();
        setState(clusterService, clusterState);
    }
    static DiscoveryNode newNode(int nodeId) {
        return DiscoveryNodeUtils.builder("node_" + nodeId).roles(emptySet()).build();
    }
    @AfterClass
    public static void destroyThreadPool() {
        ThreadPool.terminate(THREAD_POOL, 30, TimeUnit.SECONDS);
        
        THREAD_POOL = null;
    }
    public void testGlobalBlock() {
        Request request = new Request(TEST_INDEX);
        PlainActionFuture<Response> listener = new PlainActionFuture<>();
        ClusterBlocks.Builder block = ClusterBlocks.builder()
            .addGlobalBlock(new ClusterBlock(1, "test-block", false, true, false, RestStatus.SERVICE_UNAVAILABLE, ClusterBlockLevel.ALL));
        setState(clusterService, ClusterState.builder(clusterService.state()).blocks(block));
        try {