private CompletableFuture<Void> sendOffsetsForLeaderEpochRequestsAndValidatePositions(
            Map<TopicPartition, SubscriptionState.FetchPosition> partitionsToValidate) {

        final Map<Node, Map<TopicPartition, SubscriptionState.FetchPosition>> regrouped =
                regroupFetchPositionsByLeader(partitionsToValidate);

        long nextResetTimeMs = time.milliseconds() + requestTimeoutMs;
        final AtomicInteger expectedResponses = new AtomicInteger(0);
        final CompletableFuture<Void> globalResult = new CompletableFuture<>();
        final List<NetworkClientDelegate.UnsentRequest> unsentRequests = new ArrayList<>();
        regrouped.forEach((node, fetchPositions) -> {

            if (node.isEmpty()) {
                metadata.requestUpdate(true);
                return;
            }

            NodeApiVersions nodeApiVersions = apiVersions.get(node.idString());
            if (nodeApiVersions == null) {
                networkClientDelegate.tryConnect(node);
                return;
            }

            if (!hasUsableOffsetForLeaderEpochVersion(nodeApiVersions)) {
                log.debug("Skipping validation of fetch offsets for partitions {} since the broker does not " +
                                "support the required protocol version (introduced in Kafka 2.3)",
                        fetchPositions.keySet());
                for (TopicPartition partition : fetchPositions.keySet()) {
                    subscriptionState.completeValidation(partition);
                }
                return;
            }

            subscriptionState.setNextAllowedRetry(fetchPositions.keySet(), nextResetTimeMs);

            CompletableFuture<OffsetsForLeaderEpochUtils.OffsetForEpochResult> partialResult =
                    buildOffsetsForLeaderEpochRequestToNode(node, fetchPositions, unsentRequests);

            partialResult.whenComplete((offsetsResult, error) -> {
                if (error == null) {
                    offsetFetcherUtils.onSuccessfulResponseForValidatingPositions(fetchPositions,
                            offsetsResult);
                } else {
                    RuntimeException e;
                    if (error instanceof RuntimeException) {
                        e = (RuntimeException) error;
                    } else {
                        e = new RuntimeException("Unexpected failure in OffsetsForLeaderEpoch " +
                                "request for validating positions", error);
                    }
                    offsetFetcherUtils.onFailedResponseForValidatingPositions(fetchPositions, e);
                }
                if (expectedResponses.decrementAndGet() == 0) {
                    globalResult.complete(null);
                }
            });
        });

        if (unsentRequests.isEmpty()) {
            globalResult.complete(null);
        } else {
            expectedResponses.set(unsentRequests.size());
            requestsToSend.addAll(unsentRequests);
        }

        return globalResult;
    }
For each partition that needs validation, make an asynchronous request to get the end-offsets
for the partition with the epoch less than or equal to the epoch the partition last saw.
<p/>
Requests are grouped by Node for efficiency.
This also adds the request to the list of unsentRequests.

@param partitionsToValidate a map of topic-partition positions to validate
@return A {@link CompletableFuture} which completes when the requests are
complete.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManager.java