Status

Current state: "Under Discussion"

Discussion thread: here 

JIRA: KAKFA-3741

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

There is presently no way in Kafka Streams to provide default configs for the internal topics that are created. Anyone that wants to provide topic configs can only do it via the StateStoreSupplier API, which is cumbersome and only applies to changelog topics. This will usually result in having to create the topics first (which involves resolving the generated name), and/or changing the configuration after the topics have been created.

We should provide a prefix that can be used when creating a StreamsConfig. The prefix will mark these configs as Topic Configs and should be applied at the time any internal topics are created.

Public Interfaces

We will add the following field and method to StreamsConfig

/**
 * Prefix used to provide default topic configs to be applied when creating internal topics.
 * These should be valid properties from {@link org.apache.kafka.common.config.TopicConfig TopicConfig}.
 * It is recommended to use {@link #topicPrefix(String)}.
 */
public static final String TOPIC_PREFIX = "topic.";
 
/**
 * Prefix a property with {@link #TOPIC_PREFIX}
 * used to provide default topic configs to be applied when creating internal topics.
 *
 * @param topicProp the topic property to be masked
 * @return TOPIC_PREFIX + {@code topicProp}
 */
public static String topicPrefix(final String topicProp) {
    return TOPIC_PREFIX + topicProp;
}

Proposed Changes

We will add the field and method described above to StreamsConfig. At topic creation time we will extract all properties from StreamsConfig that have the TOPIC_PREFIX and use these as the default configs when creating any internal topics. Any configs that are currently provided, i.e., via a StateStoreSupplier, will override any of the defaults that have been set in StreamsConfig. 

Example Usage

// use the topicPrefix static method
config.put(StreamsConfig.topicPrefix(TopicConfig.SEGMENT_MS_CONFIG), "100");
// or use the "topic." prefix
config.put(StreamsConfig.TOPIC_PREFIX + TopicConfig.COMPRESSION_TYPE_CONFIG, "gzip),

Compatibility, Deprecation, and Migration Plan

• None

Rejected Alternatives

Adding each individual topic config to StreamsConfig. We already have these params duplicated elsewhere.