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

...

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

public interface ReplicationPolicy {
    ...

    default boolean isMM2InternalTopic(String topic) {
        // 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 checkpoint topic (as by default, it ends with ".checkpoints.internal", but can be overwritten by implementing classes)
        return  topic.startsWith("mm2") && topic.endsWith(".internal")  || isCheckpointsTopic(topic);
    }

    default boolean isInternalTopic(String topic) {
        boolean isKafkaInternalTopic = topic.startsWith("__") || topic.startsWith(".");
        return isMM2InternalTopic(topic) || isKafkaInternalTopic;
    }

}

And remove the following override from modify the implementation of the isMM2InternalTopic method in 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 usedto include the configurable internal suffix:

public class DefaultReplicationPolicy implements ReplicationPolicy, Configurable { 
    ...

      

 // Remove
    @Override@Override
    public boolean isMM2InternalTopic(String topic) {
        return topic.startsWith("mm2") && topic.endsWith(internalSuffix()) || isCheckpointsTopic(topic);
    }    
}

The 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, __.*";

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.

Rejected Alternatives

Already existing "workarounds" :

...