...

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

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 class DefaultReplicationPolicy implementsinterface 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 should alsocan be addedoverwritten toby theimplementing regexclasses)
        return  topic.startsWith("mm2")   internalTopics += String.format("|.*%sinternal", Pattern.quote(separator));
  && topic.endsWith("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;
    }

}

And remove the following override from the DefaultReplicationPolicy, as the default implementation in the interface checks if the topic ends with "internal" and ignore the separator, even if a custom one is used:

public class DefaultReplicationPolicy implements ReplicationPolicy, Configurable   internalTopicsPattern = Pattern.compile(internalTopics);{ 
     
}

	...

       //   Remove

     @Override @Override
    public boolean isInternalTopicisMM2InternalTopic(String topic) {
        return  internalTopicsPatterntopic.matcherendsWith(internalSuffix(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

...

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.