private void validateCleanupPolicyForVersionedChangelogs(final ValidationResult validationResult,
                                                             final InternalTopicConfig topicConfig,
                                                             final Config brokerSideTopicConfig) {
        final String topicName = topicConfig.name();
        final String cleanupPolicy = getBrokerSideConfigValue(brokerSideTopicConfig, TopicConfig.CLEANUP_POLICY_CONFIG, topicName);

        if (cleanupPolicy.contains(TopicConfig.CLEANUP_POLICY_DELETE)) {
            validationResult.addMisconfiguration(
                topicName,
                "Cleanup policy (" + TopicConfig.CLEANUP_POLICY_CONFIG + ") of existing internal topic "
                    + topicName + " should not contain \""
                    + TopicConfig.CLEANUP_POLICY_DELETE + "\"."
            );
        }

        final long brokerSideCompactionLagMs =
            Long.parseLong(getBrokerSideConfigValue(brokerSideTopicConfig, TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG, topicName));
        final Map<String, String> streamsSideConfig =
            topicConfig.getProperties(defaultTopicConfigs, windowChangeLogAdditionalRetention);
        final long streamsSideCompactionLagMs = Long.parseLong(streamsSideConfig.get(TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG));
        if (brokerSideCompactionLagMs < streamsSideCompactionLagMs) {
            validationResult.addMisconfiguration(
                topicName,
                "Min compaction lag (" + TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG + ") of existing internal topic "
                    + topicName + " is " + brokerSideCompactionLagMs + " but should be " + streamsSideCompactionLagMs + " or larger."
            );
        }
    }
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