@Test
    public void testGetOffsetsForTimesWithUnknownOffsetV0() {
        buildFetcher();
        // Empty map
        assertTrue(offsetFetcher.offsetsForTimes(new HashMap<>(), time.timer(100L)).isEmpty());
        // Unknown Offset
        client.reset();
        // Ensure metadata has both partition.
        MetadataResponse initialMetadataUpdate = RequestTestUtils.metadataUpdateWithIds(1, singletonMap(topicName, 1), topicIds);
        client.updateMetadata(initialMetadataUpdate);
        // Force LIST_OFFSETS version 0
        Node node = metadata.fetch().nodes().get(0);
        apiVersions.update(node.idString(), NodeApiVersions.create(
            ApiKeys.LIST_OFFSETS.id, (short) 0, (short) 0));

        ListOffsetsResponseData data = new ListOffsetsResponseData()
                .setThrottleTimeMs(0)
                .setTopics(Collections.singletonList(new ListOffsetsTopicResponse()
                        .setName(tp0.topic())
                        .setPartitions(Collections.singletonList(new ListOffsetsPartitionResponse()
                                .setPartitionIndex(tp0.partition())
                                .setErrorCode(Errors.NONE.code())
                                .setTimestamp(ListOffsetsResponse.UNKNOWN_TIMESTAMP)
                                .setOldStyleOffsets(Collections.emptyList())))));

        client.prepareResponseFrom(new ListOffsetsResponse(data),
                metadata.fetch().leaderFor(tp0));

        Map<TopicPartition, Long> timestampToSearch = new HashMap<>();
        timestampToSearch.put(tp0, 0L);
        Map<TopicPartition, OffsetAndTimestamp> offsetAndTimestampMap =
                offsetFetcher.offsetsForTimes(timestampToSearch, time.timer(Long.MAX_VALUE));

        assertTrue(offsetAndTimestampMap.containsKey(tp0));
        assertNull(offsetAndTimestampMap.get(tp0));
    }
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