public ValidationResult validate(final Map<String, InternalTopicConfig> topicConfigs) {
        log.info("Starting to validate internal topics {}.", topicConfigs.keySet());

        final long now = time.milliseconds();
        final long deadline = now + retryTimeoutMs;

        final ValidationResult validationResult = new ValidationResult();
        final Set<String> topicDescriptionsStillToValidate = new HashSet<>(topicConfigs.keySet());
        final Set<String> topicConfigsStillToValidate = new HashSet<>(topicConfigs.keySet());
        while (!topicDescriptionsStillToValidate.isEmpty() || !topicConfigsStillToValidate.isEmpty()) {
            Map<String, KafkaFuture<TopicDescription>> descriptionsForTopic = Collections.emptyMap();
            if (!topicDescriptionsStillToValidate.isEmpty()) {
                final DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(topicDescriptionsStillToValidate);
                descriptionsForTopic = describeTopicsResult.topicNameValues();
            }
            Map<String, KafkaFuture<Config>> configsForTopic = Collections.emptyMap();
            if (!topicConfigsStillToValidate.isEmpty()) {
                final DescribeConfigsResult describeConfigsResult = adminClient.describeConfigs(
                    topicConfigsStillToValidate.stream()
                        .map(topic -> new ConfigResource(Type.TOPIC, topic))
                        .collect(Collectors.toSet())
                );
                configsForTopic = describeConfigsResult.values().entrySet().stream()
                    .collect(Collectors.toMap(entry -> entry.getKey().name(), Map.Entry::getValue));
            }

            while (!descriptionsForTopic.isEmpty() || !configsForTopic.isEmpty()) {
                if (!descriptionsForTopic.isEmpty()) {
                    doValidateTopic(
                        validationResult,
                        descriptionsForTopic,
                        topicConfigs,
                        topicDescriptionsStillToValidate,
                        (streamsSide, brokerSide) -> validatePartitionCount(validationResult, streamsSide, brokerSide)
                    );
                }
                if (!configsForTopic.isEmpty()) {
                    doValidateTopic(
                        validationResult,
                        configsForTopic,
                        topicConfigs,
                        topicConfigsStillToValidate,
                        (streamsSide, brokerSide) -> validateCleanupPolicy(validationResult, streamsSide, brokerSide)
                    );
                }

                maybeThrowTimeoutException(
                    Arrays.asList(topicDescriptionsStillToValidate, topicConfigsStillToValidate),
                    deadline,
                    String.format("Could not validate internal topics within %d milliseconds. " +
                        "This can happen if the Kafka cluster is temporarily not available.", retryTimeoutMs)
                );

                if (!descriptionsForTopic.isEmpty() || !configsForTopic.isEmpty()) {
                    Utils.sleep(100);
                }
            }

            maybeSleep(
                Arrays.asList(topicDescriptionsStillToValidate, topicConfigsStillToValidate),
                deadline,
                "validated"
            );
        }

        log.info("Completed validation of internal topics {}.", topicConfigs.keySet());
        return validationResult;
    }
@Test
    public void shouldThrowWhenConfigDescriptionsDoNotContainTopicDuringValidation() {
        final AdminClient admin = mock(AdminClient.class);
        final InternalTopicManager topicManager = new InternalTopicManager(
            time,
            admin,
            new StreamsConfig(config)
        );
        final KafkaFutureImpl<TopicDescription> topicDescriptionSuccessfulFuture = new KafkaFutureImpl<>();
        topicDescriptionSuccessfulFuture.complete(new TopicDescription(
            topic1,
            false,
            Collections.singletonList(new TopicPartitionInfo(0, broker1, cluster, Collections.emptyList()))
        ));
        when(admin.describeTopics(Collections.singleton(topic1)))
            .thenAnswer(answer -> new MockDescribeTopicsResult(mkMap(mkEntry(topic1, topicDescriptionSuccessfulFuture))));
        final KafkaFutureImpl<Config> topicConfigSuccessfulFuture = new KafkaFutureImpl<>();
        topicConfigSuccessfulFuture.complete(new Config(Collections.emptySet()));
        final ConfigResource topicResource1 = new ConfigResource(Type.TOPIC, topic1);
        final ConfigResource topicResource2 = new ConfigResource(Type.TOPIC, topic2);
        when(admin.describeConfigs(Collections.singleton(topicResource1)))
            .thenAnswer(answer -> new MockDescribeConfigsResult(mkMap(mkEntry(topicResource2, topicConfigSuccessfulFuture))));
        final InternalTopicConfig internalTopicConfig = setupRepartitionTopicConfig(topic1, 1);

        assertThrows(
            IllegalStateException.class,
            () -> topicManager.validate(Collections.singletonMap(topic1, internalTopicConfig))
        );
    }