private PollResult pollInternal(Map<Node, FetchSessionHandler.FetchRequestData> fetchRequests,
                                    ResponseHandler<ClientResponse> successHandler,
                                    ResponseHandler<Throwable> errorHandler) {
        List<UnsentRequest> requests = fetchRequests.entrySet().stream().map(entry -> {
            final Node fetchTarget = entry.getKey();
            final FetchSessionHandler.FetchRequestData data = entry.getValue();
            final FetchRequest.Builder request = createFetchRequest(fetchTarget, data);
            final BiConsumer<ClientResponse, Throwable> responseHandler = (clientResponse, error) -> {
                if (error != null)
                    errorHandler.handle(fetchTarget, data, error);
                else
                    successHandler.handle(fetchTarget, data, clientResponse);
            };

            return new UnsentRequest(request, Optional.of(fetchTarget)).whenComplete(responseHandler);
        }).collect(Collectors.toList());

        return new PollResult(requests);
    }
Creates the {@link PollResult poll result} that contains a list of zero or more
{@link FetchRequest.Builder fetch requests}.

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