Status

Current state: Under Discussion

Discussion thread: here

JIRA: KAFKA-17200 

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

Motivation

In the current Mirror Maker 2 implementation, topics ending in ".internal" or "-internal" cannot be replicated as they are considered connect / mm2 internal topics. In some cases, users have business topics ending in ".internal" or "-internal" that are excluded from the replication for the same reason. This is because of two things:

(1) The ReplicationPolicy interface explicitly excludes topics from the replication that seem internal based on the following rules:

    /** Internal topics are never replicated. */
    default boolean isInternalTopic(String topic) {
        boolean isKafkaInternalTopic = topic.startsWith("__") || topic.startsWith(".");
        boolean isDefaultConnectTopic =  topic.endsWith("-internal") ||  topic.endsWith(".internal");
        return isMM2InternalTopic(topic) || isKafkaInternalTopic || isDefaultConnectTopic;
    }

(2) The topic filter excludes internal topics from the replication by default:

    public static final String TOPICS_EXCLUDE_CONFIG_ALIAS = "topics.blacklist";
    private static final String TOPICS_EXCLUDE_DOC = "List of topics and/or regexes that should not be replicated.";
    public static final String TOPICS_EXCLUDE_DEFAULT = ".*[\\-\\.]internal, .*\\.replica, __.*";

While the exclude list of the topic filter is configurable, the ReplicationPolicy interface cannot be configured in a way to enable replicating such topics. Currently, if a user already has business topics ending in for example ".internal" or "-internal", the only option is to implement a custom replication policy and override the isInternalTopic method. The goal of this proposal is to make this behavior configurable.

Public Interfaces

Proposed Changes

The default implementation of this regex based solution will keep the current behaviour and we will use the DefaultReplicationPolicy class to add the ability to override the regex using the replication.policy.internal.topics configuration:

public class DefaultReplicationPolicy implements ReplicationPolicy, Configurable {

    public static final String INTERNAL_TOPICS_CONFIG = "replication.policy.internal.topics";
    public static final String INTERNAL_TOPICS_DEFAULT = "__.*|\\..*|.*-internal|.*\\.internal";
	
    private String internalTopics = INTERNAL_TOPICS_DEFAULT;
    private Pattern internalTopicsPattern = Pattern.compile(internalTopics);
	...

    @Override
    public void configure(Map<String, ?> props) {
        if (props.containsKey(SEPARATOR_CONFIG)) {
            ...
            // When custom separator is used, to preserve the current behavior, this should also be added to the regex
            internalTopics += String.format("|.*%sinternal", Pattern.quote(separator));
        }
        // When the 'replication.policy.internal.topics' config is set, it overwrites the default value
        if (props.containsKey(INTERNAL_TOPICS_CONFIG)) {
            internalTopics = (String) props.get(INTERNAL_TOPICS_CONFIG);
        }
        internalTopicsPattern = Pattern.compile(internalTopics);     
}

	...          

    @Override
    public boolean isInternalTopic(String topic) {
        return internalTopicsPattern.matcher(topic).matches();
    } 
}

The current behavior is preserved the following way:

topic.endsWith("-internal") ||  topic.endsWith(".internal")

Note that it is not enough to override the regex using this new configuration property, but the exclude list of the TopicFilter must be modified too.

Compatibility, Deprecation, and Migration Plan

The default value for replication.policy.internal.topics considers the same topics to be internal as the current implementation.

Test Plan

Beside unit tests, this change can be tested on two Kafka cluster, with setting up a replication between them and enabling / disabling the new configuration property.

Rejected Alternatives

Already existing "workarounds" :

1. Use non-conflicting names for user topics

    This could only be a feasible option for newly created topics. In some cases it might cause too much overhead as the business / internal applications rely on this topic naming.

2. Use the replication.policy.separator to use a non-conflicting separator character

    This is only an option for new setups, as already existing setups without setting this config would use the default .internal convention. The default implementation of the ReplicationPolicy would still filter out topics ending in '.internal' or '-internal' as it is hardcoded.

3. Use a custom ReplicationPolicy that overrides this behavior

    This would be a feasible option for existing workloads too, but in my opinion this requires too much work for something that could possibly be controlled by a configuration property.

Solution proposals:

1. Add a new configuration property "internal.topic.replication.capability.enabled" that could control if the ReplicationPolicy should filter internal topics or not

   In this solution, we would override the isInternalTopic method of the DefaultReplicationPolicy based on the mentioned new configuration property. If set to true, the policy would not filter internal topics, and the isInternalTopic method would return false, leaving it to the TopicFilter to filter out any topics that should not be replicated. If set to false, the original behavior is preserved.

Risks:

• May allow replication cycles
• Higher chances of connect / mm2 internal topics being replicated because of a wrong configuration.