@Override
    public ListOffsetsResult listOffsets(Map<TopicPartition, OffsetSpec> topicPartitionOffsets,
                                         ListOffsetsOptions options) {
        AdminApiFuture.SimpleAdminApiFuture<TopicPartition, ListOffsetsResultInfo> future =
            ListOffsetsHandler.newFuture(topicPartitionOffsets.keySet());
        Map<TopicPartition, Long> offsetQueriesByPartition = topicPartitionOffsets.entrySet().stream()
            .collect(Collectors.toMap(Map.Entry::getKey, e -> getOffsetFromSpec(e.getValue())));
        ListOffsetsHandler handler = new ListOffsetsHandler(offsetQueriesByPartition, options, logContext);
        invokeDriver(handler, future, options.timeoutMs);
        return new ListOffsetsResult(future.all());
    }
Returns the broker id pertaining to the given resource, or null if the resource is not associated
with a particular broker.
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java