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);
        }
    }
@Test
    public void testMaybeValidatePositionForCurrentLeader() {
        NodeApiVersions oldApis = NodeApiVersions.create(ApiKeys.OFFSET_FOR_LEADER_EPOCH.id, (short) 0, (short) 2);
        ApiVersions apiVersions = new ApiVersions();
        apiVersions.update("1", oldApis);

        Node broker1 = new Node(1, "localhost", 9092);
        state.assignFromUser(Collections.singleton(tp0));

        state.seekUnvalidated(tp0, new SubscriptionState.FetchPosition(10L, Optional.of(5),
                new Metadata.LeaderAndEpoch(Optional.of(broker1), Optional.of(10))));

        // if API is too old to be usable, we just skip validation
        assertFalse(state.maybeValidatePositionForCurrentLeader(apiVersions, tp0, new Metadata.LeaderAndEpoch(
                Optional.of(broker1), Optional.of(10))));
        assertTrue(state.hasValidPosition(tp0));

        // New API
        apiVersions.update("1", NodeApiVersions.create());
        state.seekUnvalidated(tp0, new SubscriptionState.FetchPosition(10L, Optional.of(5),
                new Metadata.LeaderAndEpoch(Optional.of(broker1), Optional.of(10))));

        // API is too old to be usable, we just skip validation
        assertTrue(state.maybeValidatePositionForCurrentLeader(apiVersions, tp0, new Metadata.LeaderAndEpoch(
                Optional.of(broker1), Optional.of(10))));
        assertFalse(state.hasValidPosition(tp0));

        // tp1 is not part of the subscription, so validation should be skipped.
        assertFalse(state.maybeValidatePositionForCurrentLeader(apiVersions, tp1, new Metadata.LeaderAndEpoch(
            Optional.of(broker1), Optional.of(10))));
        assertFalse(state.assignedPartitions().contains(tp1));
    }