private Map<String, KafkaFuture<TopicDescription>> handleDescribeTopicsByNamesWithDescribeTopicPartitionsApi(
        final Collection<String> topicNames,
        DescribeTopicsOptions options
    ) {
        final Map<String, KafkaFutureImpl<TopicDescription>> topicFutures = new HashMap<>(topicNames.size());
        final ArrayList<String> topicNamesList = new ArrayList<>();
        for (String topicName : topicNames) {
            if (topicNameIsUnrepresentable(topicName)) {
                KafkaFutureImpl<TopicDescription> future = new KafkaFutureImpl<>();
                future.completeExceptionally(new InvalidTopicException("The given topic name '" +
                    topicName + "' cannot be represented in a request."));
                topicFutures.put(topicName, future);
            } else if (!topicFutures.containsKey(topicName)) {
                topicFutures.put(topicName, new KafkaFutureImpl<>());
                topicNamesList.add(topicName);
            }
        }

        if (topicNamesList.isEmpty()) {
            return new HashMap<>(topicFutures);
        }

        // First, we need to retrieve the node info.
        DescribeClusterResult clusterResult = describeCluster();
        clusterResult.nodes().whenComplete(
            (nodes, exception) -> {
                if (exception != null) {
                    completeAllExceptionally(topicFutures.values(), exception);
                    return;
                }

                final long now = time.milliseconds();
                Map<Integer, Node> nodeIdMap = nodes.stream().collect(Collectors.toMap(Node::id, node -> node));
                runnable.call(
                    generateDescribeTopicsCallWithDescribeTopicPartitionsApi(topicNamesList, topicFutures, nodeIdMap, options, now),
                    now
                );
            });

        return new HashMap<>(topicFutures);
    }
Fail futures in the given Map which were retried due to exceeding quota. We propagate
the initial error back to the caller if the request timed out.
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java