@Test
    public void testOffsetValidationFencing() {
        buildFetcher();
        assignFromUser(singleton(tp0));

        Map<String, Integer> partitionCounts = new HashMap<>();
        partitionCounts.put(tp0.topic(), 4);

        final int epochOne = 1;
        final int epochTwo = 2;
        final int epochThree = 3;

        // Start with metadata, epoch=1
        metadata.updateWithCurrentRequestVersion(RequestTestUtils.metadataUpdateWithIds("dummy", 1,
                Collections.emptyMap(), partitionCounts, tp -> epochOne, topicIds), false, 0L);

        // Offset validation requires OffsetForLeaderEpoch request v3 or higher
        Node node = metadata.fetch().nodes().get(0);
        apiVersions.update(node.idString(), NodeApiVersions.create());

        // Seek with a position and leader+epoch
        Metadata.LeaderAndEpoch leaderAndEpoch = new Metadata.LeaderAndEpoch(metadata.currentLeader(tp0).leader, Optional.of(epochOne));
        subscriptions.seekValidated(tp0, new SubscriptionState.FetchPosition(0, Optional.of(epochOne), leaderAndEpoch));

        // Update metadata to epoch=2, enter validation
        metadata.updateWithCurrentRequestVersion(RequestTestUtils.metadataUpdateWithIds("dummy", 1,
                Collections.emptyMap(), partitionCounts, tp -> epochTwo, topicIds), false, 0L);
        offsetFetcher.validatePositionsIfNeeded();
        assertTrue(subscriptions.awaitingValidation(tp0));

        // Update the position to epoch=3, as we would from a fetch
        subscriptions.completeValidation(tp0);
        SubscriptionState.FetchPosition nextPosition = new SubscriptionState.FetchPosition(
                10,
                Optional.of(epochTwo),
                new Metadata.LeaderAndEpoch(leaderAndEpoch.leader, Optional.of(epochTwo)));
        subscriptions.position(tp0, nextPosition);
        subscriptions.maybeValidatePositionForCurrentLeader(apiVersions, tp0, new Metadata.LeaderAndEpoch(leaderAndEpoch.leader, Optional.of(epochThree)));

        // Prepare offset list response from async validation with epoch=2
        client.prepareResponse(prepareOffsetsForLeaderEpochResponse(tp0, epochTwo, 10L));
        consumerClient.pollNoWakeup();
        assertTrue(subscriptions.awaitingValidation(tp0), "Expected validation to fail since leader epoch changed");

        // Next round of validation, should succeed in validating the position
        offsetFetcher.validatePositionsIfNeeded();
        client.prepareResponse(prepareOffsetsForLeaderEpochResponse(tp0, epochThree, 10L));
        consumerClient.pollNoWakeup();
        assertFalse(subscriptions.awaitingValidation(tp0), "Expected validation to succeed with latest epoch");
    }
Make sure the client behaves appropriately when receiving an exception for unavailable offsets
clients/src/test/java/org/apache/kafka/clients/consumer/internals/OffsetFetcherTest.java