private void resetPositionsAsync(Map<TopicPartition, Long> partitionResetTimestamps) {
        Map<Node, Map<TopicPartition, ListOffsetsPartition>> timestampsToSearchByNode =
                groupListOffsetRequests(partitionResetTimestamps, new HashSet<>());
        for (Map.Entry<Node, Map<TopicPartition, ListOffsetsPartition>> entry : timestampsToSearchByNode.entrySet()) {
            Node node = entry.getKey();
            final Map<TopicPartition, ListOffsetsPartition> resetTimestamps = entry.getValue();
            subscriptions.setNextAllowedRetry(resetTimestamps.keySet(), time.milliseconds() + requestTimeoutMs);

            RequestFuture<ListOffsetResult> future = sendListOffsetRequest(node, resetTimestamps, false);
            future.addListener(new RequestFutureListener<ListOffsetResult>() {
                @Override
                public void onSuccess(ListOffsetResult result) {
                    offsetFetcherUtils.onSuccessfulResponseForResettingPositions(resetTimestamps, result);
                }

                @Override
                public void onFailure(RuntimeException e) {
                    offsetFetcherUtils.onFailedResponseForResettingPositions(resetTimestamps, e);
                }
            });
        }
    }
Validate offsets for all assigned partitions for which a leader change has been detected.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java