public synchronized boolean maybeValidatePositionForCurrentLeader(ApiVersions apiVersions,
                                                                      TopicPartition tp,
                                                                      Metadata.LeaderAndEpoch leaderAndEpoch) {
        TopicPartitionState state = assignedStateOrNull(tp);
        if (state == null) {
            log.debug("Skipping validating position for partition {} which is not currently assigned.", tp);
            return false;
        }
        if (leaderAndEpoch.leader.isPresent()) {
            NodeApiVersions nodeApiVersions = apiVersions.get(leaderAndEpoch.leader.get().idString());
            if (nodeApiVersions == null || hasUsableOffsetForLeaderEpochVersion(nodeApiVersions)) {
                return state.maybeValidatePosition(leaderAndEpoch);
            } else {
                // If the broker does not support a newer version of OffsetsForLeaderEpoch, we skip validation
                state.updatePositionLeaderNoValidation(leaderAndEpoch);
                return false;
            }
        } else {
            return state.maybeValidatePosition(leaderAndEpoch);
        }
    }
Enter the offset validation state if the leader for this partition is known to support a usable version of the
OffsetsForLeaderEpoch API. If the leader node does not support the API, simply complete the offset validation.

@param apiVersions supported API versions
@param tp topic partition to validate
@param leaderAndEpoch leader epoch of the topic partition
@return true if we enter the offset validation state
clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java