@Override
    public Map<String, String> getProperties(final Map<String, String> defaultProperties, final long additionalRetentionMs) {
        // internal topic config overridden rule: library overrides < global config overrides < per-topic config overrides
        final Map<String, String> topicConfig = new HashMap<>(WINDOWED_STORE_CHANGELOG_TOPIC_DEFAULT_OVERRIDES);

        topicConfig.putAll(defaultProperties);

        topicConfig.putAll(topicConfigs);

        if (!topicConfigs.containsKey(TopicConfig.RETENTION_MS_CONFIG)) {
            long retentionValue;
            try {
                retentionValue = Math.addExact(retentionMs, additionalRetentionMs);
            } catch (final ArithmeticException swallow) {
                retentionValue = Long.MAX_VALUE;
            }
            topicConfig.put(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(retentionValue));
        }

        return topicConfig;
    }
Get the configured properties for this topic. If no retentionMs override is provided from
the topic configs, then we add additionalRetentionMs to work out the desired retention
when cleanup.policy=compact,delete

@param additionalRetentionMs - added to retention to allow for clock drift etc
@return Properties to be used when creating the topic
streams/src/main/java/org/apache/kafka/streams/processor/internals/WindowedChangelogTopicConfig.java