class TransportFetchNodeAction extends TransportAction<NodeFetchRequest, NodeFetchResponse> {

    private final Transports transports;
    private final NodeFetchOperation nodeFetchOperation;

    @Inject
    public TransportFetchNodeAction(Settings settings,
                                    TransportService transportService,
                                    Transports transports,
                                    ThreadPool threadPool,
                                    JobsLogs jobsLogs,
                                    TasksService tasksService,
                                    CircuitBreakerService circuitBreakerService) {
        super(FetchNodeAction.NAME);
        this.transports = transports;
        this.nodeFetchOperation = new NodeFetchOperation(
            (ThreadPoolExecutor) threadPool.executor(ThreadPool.Names.SEARCH),
            EsExecutors.numberOfProcessors(settings),
            jobsLogs,
            tasksService,
            circuitBreakerService.getBreaker(HierarchyCircuitBreakerService.QUERY)
        );

        transportService.registerRequestHandler(
            FetchNodeAction.NAME,
            ThreadPool.Names.SEARCH,
            // force execution because this handler might receive empty close requests which
            // need to be processed to not leak the FetchTask.
            // This shouldn't cause too much of an issue because fetch requests always happen after a query phase.
            // If the threadPool is overloaded the query phase would fail first.
            true,
            false,
            NodeFetchRequest.FetchRequest::new,
            new NodeActionRequestHandler<>(this::nodeOperation)
        );
    }

    @Override
    public void doExecute(NodeFetchRequest nodeFetchRequest, ActionListener<NodeFetchResponse> listener) {
        transports.sendRequest(
            FetchNodeAction.NAME,
            nodeFetchRequest.nodeId(),
            nodeFetchRequest.innerRequest(),
            listener,
            new ActionListenerResponseHandler<>(listener, nodeFetchRequest.createResponseReader())
        );
    }

    private CompletableFuture<NodeFetchResponse> nodeOperation(final NodeFetchRequest.FetchRequest request) {<FILL_FUNCTION_BODY>}
}

        CompletableFuture<? extends IntObjectMap<StreamBucket>> resultFuture = nodeFetchOperation.fetch(
            request.jobId(),
            request.fetchPhaseId(),
            request.toFetch(),
            request.isCloseContext()
        );
        return resultFuture.thenApply(NodeFetchResponse::new);
    
<methods>public final CompletableFuture<io.crate.execution.engine.fetch.NodeFetchResponse> execute(io.crate.execution.engine.fetch.NodeFetchRequest) ,public final CompletableFuture<T> execute(io.crate.execution.engine.fetch.NodeFetchRequest, Function<? super io.crate.execution.engine.fetch.NodeFetchResponse,? extends T>) <variables>protected final non-sealed java.lang.String actionName