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

        ListOffsetsResponseData data = new ListOffsetsResponseData()
                .setThrottleTimeMs(0)
                .setTopics(Collections.singletonList(new ListOffsetsTopicResponse()
                        .setName(tp0.topic())
                        .setPartitions(Arrays.asList(
                                new ListOffsetsPartitionResponse()
                                    .setPartitionIndex(tp0.partition())
                                    .setErrorCode(Errors.NOT_LEADER_OR_FOLLOWER.code())
                                    .setTimestamp(ListOffsetsResponse.UNKNOWN_TIMESTAMP)
                                    .setOffset(ListOffsetsResponse.UNKNOWN_OFFSET),
                                new ListOffsetsPartitionResponse()
                                    .setPartitionIndex(tp1.partition())
                                    .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code())
                                    .setTimestamp(ListOffsetsResponse.UNKNOWN_TIMESTAMP)
                                    .setOffset(ListOffsetsResponse.UNKNOWN_OFFSET)))));
        client.prepareResponse(new ListOffsetsResponse(data));

        Map<TopicPartition, Long> offsetsToSearch = new HashMap<>();
        offsetsToSearch.put(tp0, ListOffsetsRequest.EARLIEST_TIMESTAMP);
        offsetsToSearch.put(tp1, ListOffsetsRequest.EARLIEST_TIMESTAMP);

        assertThrows(TimeoutException.class, () -> offsetFetcher.offsetsForTimes(offsetsToSearch, time.timer(1)));
    }
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