KAFKA-9726: Add IdentityReplicationPolicy to MirrorMaker2 (#10652)

connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/IdentityReplicationPolicy.java

@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.connect.mirror;
+
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** IdentityReplicationPolicy does not rename remote topics. This is useful for migrating
+  * from legacy MM1, or for any use-case involving one-way replication.
+  *
+  * N.B. MirrorMaker is not able to prevent cycles when using this class, so take care that
+  * your replication topology is acyclic. If migrating from MirrorMaker v1, this will likely
+  * already be the case.
+  */
+public class IdentityReplicationPolicy extends DefaultReplicationPolicy {
+    private static final Logger log = LoggerFactory.getLogger(IdentityReplicationPolicy.class);
+
+    public static final String SOURCE_CLUSTER_ALIAS_CONFIG = "source.cluster.alias";
+
+    private String sourceClusterAlias = null;
+
+    @Override
+    public void configure(Map<String, ?> props) {
+        super.configure(props);
+        if (props.containsKey(SOURCE_CLUSTER_ALIAS_CONFIG)) {
+            sourceClusterAlias = (String) props.get(SOURCE_CLUSTER_ALIAS_CONFIG);
+            log.info("Using source cluster alias `{}`.", sourceClusterAlias);
+        }
+    }
+
+    /** Unlike DefaultReplicationPolicy, IdentityReplicationPolicy does not include the source
+      * cluster alias in the remote topic name. Instead, topic names are unchanged.
+      *
+      * In the special case of heartbeats, we defer to DefaultReplicationPolicy.
+      */
+    @Override
+    public String formatRemoteTopic(String sourceClusterAlias, String topic) {
+        if (looksLikeHeartbeat(topic)) {
+            return super.formatRemoteTopic(sourceClusterAlias, topic);
+        } else {
+            return topic;
+        }
+    }
+
+    /** Unlike DefaultReplicationPolicy, IdendityReplicationPolicy cannot know the source of
+      * a remote topic based on its name alone. If `source.cluster.alias` is provided,
+      * `topicSource` will return that.
+      *
+      * In the special case of heartbeats, we defer to DefaultReplicationPolicy.
+      */
+    @Override
+    public String topicSource(String topic) {
+        if (looksLikeHeartbeat(topic)) {
+            return super.topicSource(topic);
+        } else {
+            return sourceClusterAlias;
+        }
+    }
+
+    /** Since any topic may be a "remote topic", this just returns `topic`.
+      *
+      * In the special case of heartbeats, we defer to DefaultReplicationPolicy.
+      */
+    @Override
+    public String upstreamTopic(String topic) {
+        if (looksLikeHeartbeat(topic)) {
+            return super.upstreamTopic(topic);
+        } else {
+            return topic;
+        }
+    }
+
+    private boolean looksLikeHeartbeat(String topic) {
+        return topic != null && topic.endsWith(MirrorClientConfig.HEARTBEATS_TOPIC);
+    }
+}

connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/MirrorClient.java

@@ -192,6 +192,7 @@ Set<String> listTopics() throws InterruptedException {
 
     int countHopsForTopic(String topic, String sourceClusterAlias) {
         int hops = 0;
+        Set<String> visited = new HashSet<>();
         while (true) {
             hops++;
             String source = replicationPolicy.topicSource(topic);
@@ -201,6 +202,12 @@ int countHopsForTopic(String topic, String sourceClusterAlias) {
             if (source.equals(sourceClusterAlias)) {
                 return hops;
             }
+            if (visited.contains(source)) {
+                // Extra check for IdentityReplicationPolicy and similar impls that cannot prevent cycles.
+                // We assume we're stuck in a cycle and will never find sourceClusterAlias.
+                return -1;
+            }
+            visited.add(source);
             topic = replicationPolicy.upstreamTopic(topic);
         } 
     }
@@ -223,7 +230,8 @@ boolean isRemoteTopic(String topic) {
     Set<String> allSources(String topic) {
         Set<String> sources = new HashSet<>();
         String source = replicationPolicy.topicSource(topic);
-        while (source != null) {
+        while (source != null && !sources.contains(source)) {
+            // The extra Set.contains above is for ReplicationPolicies that cannot prevent cycles.
             sources.add(source);
             topic = replicationPolicy.upstreamTopic(topic);
             source = replicationPolicy.topicSource(topic);

connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/ReplicationPolicy.java

@@ -45,7 +45,7 @@ public interface ReplicationPolicy {
      */
     default String originalTopic(String topic) {
         String upstream = upstreamTopic(topic);
-        if (upstream == null) {
+        if (upstream == null || upstream.equals(topic)) {
             return topic;
         } else {
             return originalTopic(upstream);

connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorMakerConfig.java

@@ -175,6 +175,7 @@ public Map<String, String> workerConfig(SourceAndTarget sourceAndTarget) {
         props.putAll(stringsWithPrefix("header.converter"));
         props.putAll(stringsWithPrefix("task"));
         props.putAll(stringsWithPrefix("worker"));
+        props.putAll(stringsWithPrefix("replication.policy"));
 
         // transform any expression like ${provider:path:key}, since the worker doesn't do so
         props = transform(props);
@@ -203,6 +204,7 @@ public Map<String, String> connectorBaseConfig(SourceAndTarget sourceAndTarget,
         props.keySet().retainAll(MirrorConnectorConfig.CONNECTOR_CONFIG_DEF.names());
 
         props.putAll(stringsWithPrefix(CONFIG_PROVIDERS_CONFIG));
+        props.putAll(stringsWithPrefix("replication.policy"));
 
         Map<String, String> sourceClusterProps = clusterProps(sourceAndTarget.source());
         // attrs non prefixed with producer|consumer|admin

connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceConnector.java

@@ -494,7 +494,12 @@ boolean isCycle(String topic) {
         } else if (source.equals(sourceAndTarget.source()) || source.equals(sourceAndTarget.target())) {
             return true;
         } else {
-            return isCycle(replicationPolicy.upstreamTopic(topic));
+            String upstreamTopic = replicationPolicy.upstreamTopic(topic);
+            if (upstreamTopic.equals(topic)) {
+                // Extra check for IdentityReplicationPolicy and similar impls that don't prevent cycles.
+                return false;
+            }
+            return isCycle(upstreamTopic);
         }
     }