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 shouldThrowTimeoutExceptionInGetPartitionInfo() {
        setupTopicInMockAdminClient(topic1, Collections.emptyMap());
        final MockTime time = new MockTime(5);
        mockAdminClient.timeoutNextRequest(Integer.MAX_VALUE);

        final InternalTopicManager internalTopicManager =
            new InternalTopicManager(time, mockAdminClient, new StreamsConfig(config));

        final TimeoutException exception = assertThrows(
            TimeoutException.class,
            () -> internalTopicManager.getTopicPartitionInfo(Collections.singleton(topic1))
        );

        assertThat(
            exception.getMessage(),
            is("Could not create topics within 50 milliseconds. This can happen if the Kafka cluster is temporarily not available.")
        );
    }