Map<TopicPartition, SubscriptionState.FetchPosition> getPartitionsToValidate() {
        RuntimeException exception = cachedValidatePositionsException.getAndSet(null);
        if (exception != null)
            throw exception;

        // Validate each partition against the current leader and epoch
        // If we see a new metadata version, check all partitions
        validatePositionsOnMetadataChange();

        // Collect positions needing validation, with backoff
        return subscriptionState
                .partitionsNeedingValidation(time.milliseconds())
                .stream()
                .filter(tp -> subscriptionState.position(tp) != null)
                .collect(Collectors.toMap(Function.identity(), subscriptionState::position));
    }
Callback for the response of the list offset call.

@param listOffsetsResponse The response from the server.
@return {@link OffsetFetcherUtils.ListOffsetResult} extracted from the response, containing the fetched offsets
and partitions to retry.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcherUtils.java