...

Proposed Changes

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 configurationThis 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_REPLICATIONTOPICS_CAPABILITY_CONFIG = MirrorClientConfig.INTERNAL_TOPIC_REPLICATION_CAPABILITY_ENABLED"replication.policy.internal.topics";
    public static final BooleanString ENABLE_INTERNAL_TOPIC_REPLICATION_CAPABILITYTOPICS_DEFAULT = false;
	...
	private boolean isInternalTopicsReplicationCapabilityEnabled = false"__.*|\\..*|.*-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)) {
            ...
		            // Configure isInternalTopicsReplicationCapabilityEnabled based on props When custom separator is used, to preserve the current behavior, this should also be added to the regex
    }

	...

    @Override
    publicinternalTopics boolean isInternalTopic(String topic) {+= String.format("|.*%sinternal", Pattern.quote(separator));
        if}
 (isInternalTopicsReplicationCapabilityEnabled) {
				// We do not filter internal topics// inWhen the 'replication policy in this case
				// and everything should be filtered using the TopicFilter
  .policy.internal.topics' config is set, it overwrites the default value
        if (props.containsKey(INTERNAL_TOPICS_CONFIG)) {
            internalTopics = return false(String) props.get(INTERNAL_TOPICS_CONFIG);
        } else {
   
        internalTopicsPattern = Pattern.compile(internalTopics);     
}

	...          

    @Override
    return ReplicationPolicy.super.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 internalreplication.topicpolicy.replication.capability.enabled is false, which preserves the previous behavior. It only changes when users directly change its value to trueinternal.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.