public Map<String, List<TopicPartitionInfo>> getTopicPartitionInfo(final Set<String> topics) {
        log.debug("Starting to describe topics {} in partition assignor.", topics);

        long currentWallClockMs = time.milliseconds();
        final long deadlineMs = currentWallClockMs + retryTimeoutMs;

        final Set<String> topicsToDescribe = new HashSet<>(topics);
        final Map<String, List<TopicPartitionInfo>> topicPartitionInfo = new HashMap<>();

        while (!topicsToDescribe.isEmpty()) {
            final Map<String, List<TopicPartitionInfo>> existed = getTopicPartitionInfo(topicsToDescribe, null);
            topicPartitionInfo.putAll(existed);
            topicsToDescribe.removeAll(topicPartitionInfo.keySet());
            if (!topicsToDescribe.isEmpty()) {
                currentWallClockMs = time.milliseconds();

                if (currentWallClockMs >= deadlineMs) {
                    final String timeoutError = String.format(
                        "Could not create topics within %d milliseconds. " +
                            "This can happen if the Kafka cluster is temporarily not available.",
                        retryTimeoutMs);
                    log.error(timeoutError);
                    throw new TimeoutException(timeoutError);
                }
                log.info(
                    "Topics {} could not be describe fully. Will retry in {} milliseconds. Remaining time in milliseconds: {}",
                    topics,
                    retryBackOffMs,
                    deadlineMs - currentWallClockMs
                );
                Utils.sleep(retryBackOffMs);
            }
        }
        log.debug("Completed describing topics");
        return topicPartitionInfo;
    }
@Test
    public void shouldReturnCorrectPartitionInfo() {
        final TopicPartitionInfo topicPartitionInfo = new TopicPartitionInfo(0, broker1, singleReplica, Collections.emptyList());
        mockAdminClient.addTopic(
            false,
            topic1,
            Collections.singletonList(topicPartitionInfo),
            null);

        final Map<String, List<TopicPartitionInfo>> ret = internalTopicManager.getTopicPartitionInfo(Collections.singleton(topic1));
        assertEquals(Collections.singletonMap(topic1, Collections.singletonList(topicPartitionInfo)), ret);
    }