private List<RequestFuture<ClientResponse>> sendFetchesInternal(Map<Node, FetchSessionHandler.FetchRequestData> fetchRequests,
                                                                    ResponseHandler<ClientResponse> successHandler,
                                                                    ResponseHandler<Throwable> errorHandler) {
        final List<RequestFuture<ClientResponse>> requestFutures = new ArrayList<>();

        for (Map.Entry<Node, FetchSessionHandler.FetchRequestData> entry : fetchRequests.entrySet()) {
            final Node fetchTarget = entry.getKey();
            final FetchSessionHandler.FetchRequestData data = entry.getValue();
            final FetchRequest.Builder request = createFetchRequest(fetchTarget, data);
            final RequestFuture<ClientResponse> responseFuture = client.send(fetchTarget, request);

            responseFuture.addListener(new RequestFutureListener<ClientResponse>() {
                @Override
                public void onSuccess(ClientResponse resp) {
                    successHandler.handle(fetchTarget, data, resp);
                }

                @Override
                public void onFailure(RuntimeException e) {
                    errorHandler.handle(fetchTarget, data, e);
                }
            });

            requestFutures.add(responseFuture);
        }

        return requestFutures;
    }
Creates the {@link FetchRequest.Builder fetch request},
{@link NetworkClient#send(ClientRequest, long) enqueues/sends it, and adds the {@link RequestFuture callback}
for the response.

@param fetchRequests  {@link Map} of {@link Node nodes} to their
                      {@link FetchSessionHandler.FetchRequestData request data}
@param successHandler {@link ResponseHandler Handler for successful responses}
@param errorHandler   {@link ResponseHandler Handler for failure responses}
@return List of {@link RequestFuture callbacks}
clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java