private void validateCleanupPolicyForWindowedChangelogs(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)) {
            final long brokerSideRetentionMs =
                Long.parseLong(getBrokerSideConfigValue(brokerSideTopicConfig, TopicConfig.RETENTION_MS_CONFIG, topicName));
            final Map<String, String> streamsSideConfig =
                topicConfig.getProperties(defaultTopicConfigs, windowChangeLogAdditionalRetention);
            final long streamsSideRetentionMs = Long.parseLong(streamsSideConfig.get(TopicConfig.RETENTION_MS_CONFIG));
            if (brokerSideRetentionMs < streamsSideRetentionMs) {
                validationResult.addMisconfiguration(
                    topicName,
                    "Retention time (" + TopicConfig.RETENTION_MS_CONFIG + ") of existing internal topic "
                        + topicName + " is " + brokerSideRetentionMs + " but should be " + streamsSideRetentionMs + " or larger."
                );
            }
            final String brokerSideRetentionBytes =
                getBrokerSideConfigValue(brokerSideTopicConfig, TopicConfig.RETENTION_BYTES_CONFIG, topicName);
            if (brokerSideRetentionBytes != null) {
                validationResult.addMisconfiguration(
                    topicName,
                    "Retention byte (" + TopicConfig.RETENTION_BYTES_CONFIG + ") of existing internal topic "
                        + topicName + " is set but it should be unset."
                );
            }
        }
    }
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