class TransportFetchOperation implements FetchOperation {

    private static final Function<NodeFetchResponse, IntObjectMap<? extends Bucket>> GET_FETCHED = NodeFetchResponse::fetched;
    private final ActionExecutor<NodeFetchRequest, NodeFetchResponse> fetchNodeAction;
    private final Map<String, ? extends IntObjectMap<Streamer<?>[]>> nodeIdToReaderIdToStreamers;
    private final UUID jobId;
    private final int fetchPhaseId;
    private final RamAccounting ramAccounting;

    public TransportFetchOperation(ActionExecutor<NodeFetchRequest, NodeFetchResponse> fetchNodeAction,
                                   Map<String, ? extends IntObjectMap<Streamer<?>[]>> nodeIdToReaderIdToStreamers,
                                   UUID jobId,
                                   int fetchPhaseId,
                                   RamAccounting ramAccounting) {
        this.fetchNodeAction = fetchNodeAction;
        this.nodeIdToReaderIdToStreamers = nodeIdToReaderIdToStreamers;
        this.jobId = jobId;
        this.fetchPhaseId = fetchPhaseId;
        this.ramAccounting = ramAccounting;
    }

    @Override
    public CompletableFuture<IntObjectMap<? extends Bucket>> fetch(String nodeId,
                                                                   IntObjectMap<IntArrayList> toFetch,
                                                                   boolean closeContext) {
        FutureActionListener<NodeFetchResponse> listener = new FutureActionListener<>();
        return fetchNodeAction
            .execute(
                new NodeFetchRequest(nodeId,
                                     jobId,
                                     fetchPhaseId,
                                     closeContext,
                                     toFetch,
                                     nodeIdToReaderIdToStreamers.get(nodeId),
                                     ramAccountingForIncomingResponse(ramAccounting, toFetch, closeContext)))
            .whenComplete(listener)
            .thenApply(GET_FETCHED);
    }

    @VisibleForTesting
    static RamAccounting ramAccountingForIncomingResponse(RamAccounting ramAccounting,
                                                          IntObjectMap<? extends IntContainer> toFetch,
                                                          boolean closeContext) {<FILL_FUNCTION_BODY>}
}

        if (toFetch.isEmpty() && closeContext) {
            // No data will arrive, so no ram accounting needed.
            // Indeed, with valid ram accounting, incoming accounted bytes may never be released because the release
            // logic may already happened (BatchAccumulator.close() calls do not block/wait for asynchronous responses)
            return RamAccounting.NO_ACCOUNTING;
        }
        // Each response may run in a different thread and thus should use its own ram accounting instance
        return new BlockBasedRamAccounting(
            usedBytes -> {
                // Projectors usually operate single-threaded and can receive a RamAccounting instance that is not thread-safe
                // So we must ensure thread-safety here.
                synchronized (ramAccounting) {
                    ramAccounting.addBytes(usedBytes);
                }
            },
            MAX_BLOCK_SIZE_IN_BYTES
        );
    
<no_super_class>