private ListOffsetsResponse listOffsetResponse(Map<TopicPartition, Long> offsets, Errors error, long timestamp, int leaderEpoch) {
        Map<String, List<ListOffsetsPartitionResponse>> responses = new HashMap<>();
        for (Map.Entry<TopicPartition, Long> entry : offsets.entrySet()) {
            TopicPartition tp = entry.getKey();
            responses.putIfAbsent(tp.topic(), new ArrayList<>());
            responses.get(tp.topic()).add(new ListOffsetsPartitionResponse()
                                .setPartitionIndex(tp.partition())
                                .setErrorCode(error.code())
                                .setOffset(entry.getValue())
                                .setTimestamp(timestamp)
                                .setLeaderEpoch(leaderEpoch));
        }
        List<ListOffsetsTopicResponse> topics = new ArrayList<>();
        for (Map.Entry<String, List<ListOffsetsPartitionResponse>> response : responses.entrySet()) {
            topics.add(new ListOffsetsTopicResponse()
                    .setName(response.getKey())
                    .setPartitions(response.getValue()));
        }
        ListOffsetsResponseData data = new ListOffsetsResponseData().setTopics(topics);
        return new ListOffsetsResponse(data);
    }
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