private <V> void doValidateTopic(final ValidationResult validationResult,
                                     final Map<String, KafkaFuture<V>> futuresForTopic,
                                     final Map<String, InternalTopicConfig> topicsConfigs,
                                     final Set<String> topicsStillToValidate,
                                     final BiConsumer<InternalTopicConfig, V> validator) {
        for (final String topicName : new HashSet<>(topicsStillToValidate)) {
            if (!futuresForTopic.containsKey(topicName)) {
                throw new IllegalStateException("Description results do not contain topics to validate. " + BUG_ERROR_MESSAGE);
            }
            final KafkaFuture<V> future = futuresForTopic.get(topicName);
            if (future.isDone()) {
                try {
                    final V brokerSideTopicConfig = future.get();
                    final InternalTopicConfig streamsSideTopicConfig = topicsConfigs.get(topicName);
                    validator.accept(streamsSideTopicConfig, brokerSideTopicConfig);
                    topicsStillToValidate.remove(topicName);
                } catch (final ExecutionException executionException) {
                    final Throwable cause = executionException.getCause();
                    if (cause instanceof UnknownTopicOrPartitionException) {
                        log.info("Internal topic {} is missing", topicName);
                        validationResult.addMissingTopic(topicName);
                        topicsStillToValidate.remove(topicName);
                    } else if (cause instanceof LeaderNotAvailableException) {
                        log.info("The leader of internal topic {} is not available.", topicName);
                    } else if (cause instanceof TimeoutException) {
                        log.info("Retrieving data for internal topic {} timed out.", topicName);
                    } else {
                        log.error("Unexpected error during internal topic validation: ", cause);
                        throw new StreamsException(
                            String.format("Could not validate internal topic %s for the following reason: ", topicName),
                            cause
                        );
                    }
                } catch (final InterruptedException interruptedException) {
                    throw new InterruptException(interruptedException);
                } finally {
                    futuresForTopic.remove(topicName);
                }
            }
        }
    }
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