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 shouldReportMisconfigurationsOfPartitionCount() {
        setupTopicInMockAdminClient(topic1, repartitionTopicConfig());
        setupTopicInMockAdminClient(topic2, repartitionTopicConfig());
        setupTopicInMockAdminClient(topic3, repartitionTopicConfig());
        final InternalTopicConfig internalTopicConfig1 = setupRepartitionTopicConfig(topic1, 2);
        final InternalTopicConfig internalTopicConfig2 = setupRepartitionTopicConfig(topic2, 3);
        final InternalTopicConfig internalTopicConfig3 = setupRepartitionTopicConfig(topic3, 1);

        final ValidationResult validationResult = internalTopicManager.validate(mkMap(
            mkEntry(topic1, internalTopicConfig1),
            mkEntry(topic2, internalTopicConfig2),
            mkEntry(topic3, internalTopicConfig3)
        ));

        final Map<String, List<String>> misconfigurationsForTopics = validationResult.misconfigurationsForTopics();
        assertThat(validationResult.missingTopics(), empty());
        assertThat(misconfigurationsForTopics.size(), is(2));
        assertThat(misconfigurationsForTopics, hasKey(topic1));
        assertThat(misconfigurationsForTopics.get(topic1).size(), is(1));
        assertThat(
            misconfigurationsForTopics.get(topic1).get(0),
            is("Internal topic " + topic1 + " requires 2 partitions, but the existing topic on the broker has 1 partitions.")
        );
        assertThat(misconfigurationsForTopics, hasKey(topic2));
        assertThat(misconfigurationsForTopics.get(topic2).size(), is(1));
        assertThat(
            misconfigurationsForTopics.get(topic2).get(0),
            is("Internal topic " + topic2 + " requires 3 partitions, but the existing topic on the broker has 1 partitions.")
        );
        assertThat(misconfigurationsForTopics, not(hasKey(topic3)));
    }