Status

Current state: Under Discussion

Discussion thread: TBA

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

This KIP proposes overriding the isInternalTopic method in the DefaultReplicationPolicy class and change its behavior based on the new configuration property:

public class DefaultReplicationPolicy implements ReplicationPolicy, Configurable {

 	public static final String ENABLE_INTERNAL_TOPIC_REPLICATION_CAPABILITY_CONFIG = MirrorClientConfig.INTERNAL_TOPIC_REPLICATION_CAPABILITY_ENABLED;
    public static final Boolean ENABLE_INTERNAL_TOPIC_REPLICATION_CAPABILITY_DEFAULT = false;
	...
	private boolean isInternalTopicsReplicationCapabilityEnabled = false;
	...

    @Override
    public void configure(Map<String, ?> props) {
        ...
		// Configure isInternalTopicsReplicationCapabilityEnabled based on props
    }

	...

    @Override
    public boolean isInternalTopic(String topic) {
        if (isInternalTopicsReplicationCapabilityEnabled) {
				// We do not filter internal topics in the replication policy in this case
				// and everything should be filtered using the TopicFilter
                return false;
        } else {
            return ReplicationPolicy.super.isInternalTopic(topic);
        }
    }
}

Compatibility, Deprecation, and Migration Plan

The default value for internal.topic.replication.capability.enabled is false, which preserves the previous behavior. It only changes when users directly change its value to true.

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

-