Status

Current state: Under Discussion Accepted

Discussion thread: here

Vote: here

JIRA: KAFKA-17200 

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

Note: The original name of the KIP was changed from "Make the replication of internal topics configurable" to "Allow the replication of user internal topics" as it better describes the latest proposal and the goal in general.

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:

...

    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 behaviour configurable.

Public Interfaces

...

This KIP proposes to change the ReplicationPolicy interface, and make the filtering of internal topics more precise as described in the "Proposed Changes" section.

With changing the implementation of the interface, this would also affect implementing classes:

• DefaultReplicationPolicy
• IdentityReplicationPolicy

The default exclude list of the DefaultTopicFilter will use the same, more specific pattern.

Proposed Changes

This KIP proposes to make the condition for internal topics more specific in the ReplicationPolicy

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 classinterface DefaultReplicationPolicy implements ReplicationPolicy, Configurable {

    public static final String INTERNAL_TOPICS_CONFIG = "replication.policy.internal.topics";

    publicdefault staticboolean final isMM2InternalTopic(String INTERNAL_TOPICS_DEFAULT = "__.*|\\..*|.*-internal|.*\\.internal";
	
topic) {
    private String internalTopics = INTERNAL_TOPICS_DEFAULT;
    private Pattern internalTopicsPattern = Pattern.compile(internalTopics);
	...

    @Override
    public void configure(Map<String, ?> props) {
        if (props.containsKey(SEPARATOR_CONFIG)) {
            ... // With this change, we only consider a topic to be mm2 internal if it
        // 1. starts with "mm2" and ends wit "internal", or
        // 2. it is //a Whencheckpoint customtopic separator(as isby useddefault, toit preserveends the current behaviorwith ".checkpoints.internal", thisbut shouldcan also be addedoverwritten toby theimplementing regexclasses)
        return  topic.startsWith("mm2")   internalTopics += String.format&& topic.endsWith("|.*%sinternal", Pattern.quote(separator)internal")  || isCheckpointsTopic(topic);
        }

    default boolean isInternalTopic(String topic) //{
 When the 'replication.policy.internal.topics' config is set, it overwritesboolean theisKafkaInternalTopic default value
        if (props.containsKey(INTERNAL_TOPICS_CONFIG)) {= topic.startsWith("__") || topic.startsWith(".");
            internalTopics = (String) props.get(INTERNAL_TOPICS_CONFIG);
  return isMM2InternalTopic(topic) || isKafkaInternalTopic;
      }
        internalTopicsPattern = Pattern.compile(internalTopics);     
}

	}

}

And modify the implementation of the isMM2InternalTopic method in the DefaultReplicationPolicy, to include the configurable internal suffix:

public class DefaultReplicationPolicy implements ReplicationPolicy, Configurable { 
    ...          

    @Override
    public boolean isInternalTopicisMM2InternalTopic(String topic) {
        return internalTopicsPatterntopic.matcher(topic).matches(startsWith("mm2") && topic.endsWith(internalSuffix()) || isCheckpointsTopic(topic);
    }    
}

The current behavior is preserved the following way:

...

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

...

exclude list of the DefaultTopicFilter would also require some modification to reflect the new pattern:

public static final String TOPICS_EXCLUDE_DEFAULT = "mm2.*internal, .*\\.replica, __.*";

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

...

Backward Compatibility Considerations:

1. Anyone who relies on the current behaviour to block the replication of already existing user topics ending in ".internal" or "-internal", might need to update the TopicFilter, as with this change these topics will not be explicitly excluded.
2. Anyone who uses a custom ReplicationPolicy implementation might need to update their source code to get the same behaviour.

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" :

...

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

2. Make the internal topics configurable with a regex.

    In this solution, the ReplicationPolicy would consider a topic to be internal, if it matches a specific regex. The default regex would result in the same behavior as before, something like "__.*|\\..*|.*-internal|.*\\.internal". We would make this regex configurable by the user using a new configuration property called replication.policy.internal.topics. When users specify a value for this property, it overwrites the default behavior.

Risks:

• Easier to make mistakes resulting in inconsistent behavior
• It becomes possible for users to configure the DefaultReplicationPolicy class in a way that for some topic T, isInternalTopic(T) returns false, but isCheckpointsTopic(T), isHeartbeatsTopic(T), or isMM2InternalTopic(T) return true.