@Test
    public void testGetOffsetsForTimesWhenSomeTopicPartitionLeadersNotKnownInitially() {
        buildFetcher();

        subscriptions.assignFromUser(mkSet(tp0, tp1));
        final String anotherTopic = "another-topic";
        final TopicPartition t2p0 = new TopicPartition(anotherTopic, 0);

        client.reset();

        // Metadata initially has one topic
        MetadataResponse initialMetadata = RequestTestUtils.metadataUpdateWithIds(3, singletonMap(topicName, 2), topicIds);
        client.updateMetadata(initialMetadata);

        // The first metadata refresh should contain one topic
        client.prepareMetadataUpdate(initialMetadata);
        client.prepareResponseFrom(listOffsetResponse(tp0, Errors.NONE, 1000L, 11L),
                metadata.fetch().leaderFor(tp0));
        client.prepareResponseFrom(listOffsetResponse(tp1, Errors.NONE, 1000L, 32L),
                metadata.fetch().leaderFor(tp1));

        // Second metadata refresh should contain two topics
        Map<String, Integer> partitionNumByTopic = new HashMap<>();
        partitionNumByTopic.put(topicName, 2);
        partitionNumByTopic.put(anotherTopic, 1);
        topicIds.put("another-topic", Uuid.randomUuid());
        MetadataResponse updatedMetadata = RequestTestUtils.metadataUpdateWithIds(3, partitionNumByTopic, topicIds);
        client.prepareMetadataUpdate(updatedMetadata);
        client.prepareResponseFrom(listOffsetResponse(t2p0, Errors.NONE, 1000L, 54L),
                metadata.fetch().leaderFor(t2p0));

        Map<TopicPartition, Long> timestampToSearch = new HashMap<>();
        timestampToSearch.put(tp0, ListOffsetsRequest.LATEST_TIMESTAMP);
        timestampToSearch.put(tp1, ListOffsetsRequest.LATEST_TIMESTAMP);
        timestampToSearch.put(t2p0, ListOffsetsRequest.LATEST_TIMESTAMP);
        Map<TopicPartition, OffsetAndTimestamp> offsetAndTimestampMap =
            offsetFetcher.offsetsForTimes(timestampToSearch, time.timer(Long.MAX_VALUE));

        assertNotNull(offsetAndTimestampMap.get(tp0), "Expect MetadataFetcher.offsetsForTimes() to return non-null result for " + tp0);
        assertNotNull(offsetAndTimestampMap.get(tp1), "Expect MetadataFetcher.offsetsForTimes() to return non-null result for " + tp1);
        assertNotNull(offsetAndTimestampMap.get(t2p0), "Expect MetadataFetcher.offsetsForTimes() to return non-null result for " + t2p0);
        assertEquals(11L, offsetAndTimestampMap.get(tp0).offset());
        assertEquals(32L, offsetAndTimestampMap.get(tp1).offset());
        assertEquals(54L, offsetAndTimestampMap.get(t2p0).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