public Map<String, Object> topicCreationOtherConfigs(String group) {
        if (enrichedSourceConfig == null) {
            return Collections.emptyMap();
        }
        return enrichedSourceConfig.originalsWithPrefix(TOPIC_CREATION_PREFIX + group + '.').entrySet().stream()
                .filter(e -> {
                    String key = e.getKey();
                    return !(INCLUDE_REGEX_CONFIG.equals(key) || EXCLUDE_REGEX_CONFIG.equals(key)
                            || REPLICATION_FACTOR_CONFIG.equals(key) || PARTITIONS_CONFIG.equals(key));
                })
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
    }
@Test
    public void shouldAllowSettingTopicProperties() {
        Map<String, String> topicProps = new HashMap<>();
        topicProps.put(CLEANUP_POLICY_CONFIG, CLEANUP_POLICY_COMPACT);
        topicProps.put(COMPRESSION_TYPE_CONFIG, "lz4");
        topicProps.put(RETENTION_MS_CONFIG, String.valueOf(TimeUnit.DAYS.toMillis(30)));

        Map<String, String> props = defaultConnectorPropsWithTopicCreation();
        topicProps.forEach((k, v) -> props.put(DEFAULT_TOPIC_CREATION_PREFIX + k, v));

        SourceConnectorConfig config = new SourceConnectorConfig(MOCK_PLUGINS, props, true);
        assertEquals(topicProps,
                convertToStringValues(config.topicCreationOtherConfigs(DEFAULT_TOPIC_CREATION_GROUP)));
    }