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;
    }
Validates the internal topics passed.

The validation of the internal topics verifies if the topics:
- are missing on the brokers
- have the expected number of partitions
- have configured a clean-up policy that avoids data loss

@param topicConfigs internal topics to validate

@return validation results that contains
        - the set of missing internal topics on the brokers
        - descriptions of misconfigurations per topic
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java