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

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

        final int epochOne = 1;

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

        Node node = metadata.fetch().nodes().get(0);
        assertFalse(client.isConnected(node.idString()));

        // Seek with a position and leader+epoch
        Metadata.LeaderAndEpoch leaderAndEpoch = new Metadata.LeaderAndEpoch(
                metadata.currentLeader(tp0).leader, Optional.of(epochOne));
        subscriptions.seekUnvalidated(tp0, new SubscriptionState.FetchPosition(20L, Optional.of(epochOne), leaderAndEpoch));
        assertFalse(client.isConnected(node.idString()));
        assertTrue(subscriptions.awaitingValidation(tp0));

        // No version information is initially available, but the node is now connected
        offsetFetcher.validatePositionsIfNeeded();
        assertTrue(subscriptions.awaitingValidation(tp0));
        assertTrue(client.isConnected(node.idString()));
        apiVersions.update(node.idString(), NodeApiVersions.create());

        // On the next call, the OffsetForLeaderEpoch request is sent and validation completes
        client.prepareResponseFrom(
            prepareOffsetsForLeaderEpochResponse(tp0, epochOne, 30L),
            node);

        offsetFetcher.validatePositionsIfNeeded();
        consumerClient.pollNoWakeup();

        assertFalse(subscriptions.awaitingValidation(tp0));
        assertEquals(20L, subscriptions.position(tp0).offset);
    }
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