KAFKA-9726 IdentityReplicationPolicy

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

@@ -0,0 +1,113 @@
+/*
+ * 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 org.apache.kafka.common.Configurable;
+
+import java.util.Map;
+import java.util.regex.Pattern;
+
+/** LegacyReplicationPolicy attempts to mimic MirrorMaker v1, and thus by default does not
+  * rename topics. In order to support active/active replication with careful configuration,
+  * topics can be renamed with an optional suffix, e.g. "topic1.replica".
+  *
+  * LegacyReplicationPolicy cannot, in general, detect whether a topic is remote or not, nor
+  * which cluster a topic may have been replicated from. If a `remote.topic.suffix` is
+  * provided, we use that to distinguish remote topics. Otherwise we consider all topics as
+  * remote topics. If a `source.cluster.alias` is provided, we use that as the source of
+  * any remote topics. Otherwise we return null for `topicSource()`.
+  *
+  * Paradoxically, then, the default behavior of LegacyReplicationPolicy is to consider all
+  * topics to be remote but with no known source.
+  *
+  * 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 LegacyReplicationPolicy implements ReplicationPolicy, Configurable {
+
+    public static final String REMOTE_TOPIC_SUFFIX_CONFIG = "remote.topic.suffix";
+    public static final String SOURCE_CLUSTER_ALIAS_CONFIG = "source.cluster.alias";
+
+    private String remoteTopicSuffix = "";
+    private String sourceClusterAlias = null;
+
+    public LegacyReplicationPolicy() {
+    }
+
+    // Visible for testing
+    LegacyReplicationPolicy(String remoteTopicSuffix, String sourceClusterAlias) {
+      this.remoteTopicSuffix = remoteTopicSuffix;
+      this.sourceClusterAlias = sourceClusterAlias;
+    }
+
+    @Override
+    public void configure(Map<String, ?> props) {
+        if (props.containsKey(REMOTE_TOPIC_SUFFIX_CONFIG)) {
+            remoteTopicSuffix = (String) props.get(REMOTE_TOPIC_SUFFIX_CONFIG);
+        }
+        if (props.containsKey(SOURCE_CLUSTER_ALIAS_CONFIG)) {
+            sourceClusterAlias = (String) props.get(SOURCE_CLUSTER_ALIAS_CONFIG);
+        }
+    }
+
+    /** Unlike DefaultReplicationPolicy, LegacyReplicationPolicy does not include the source
+      * cluster alias in the remote topic name. By default, topic names are unchanged. If
+      * a `remote.topic.suffix` is provided, topic are renamed accordingly.
+      */
+    @Override
+    public String formatRemoteTopic(String sourceClusterAlias, String topic) {
+        return topic + remoteTopicSuffix;
+    }
+
+    /** Unlike DefaultReplicationPolicy, LegacyReplicationPolicy cannot know the source of
+      * a remote topic based on its name alone. If `source.cluster.alias` is provided,
+      * `topicSource` will return that for anything that looks like a remote topic. By default,
+      * all topics look like remote topics.
+      */
+    @Override
+    public String topicSource(String topic) {
+        if (topic.endsWith(remoteTopicSuffix)) {
+            return sourceClusterAlias;
+        } else {
+            return null;
+        }
+    }
+
+    /** Unlike DefaultReplicationPolicy, LegacyReplicationPolicy cannot distinguish remote
+      * topics from regular topics by default. If `remote.topic.suffix` is provided, this
+      * method will strip that and return the original topic. By default, `topic` is returned.
+      */
+    @Override
+    public String upstreamTopic(String topic) {
+        if (topic.endsWith(remoteTopicSuffix)) {
+            return topic.substring(0, topic.length() - remoteTopicSuffix.length());
+        } else {
+            return topic;
+        }
+    }
+
+    /** If remote.topic.suffix is provided, treat suffixed topics as internal topics.
+      * This prevents replicated topics from being further replicated when using
+      * LegacyReplicationPolicy.
+      */
+    @Override
+    public boolean isInternalTopic(String topic) {
+        return ReplicationPolicy.super.isInternalTopic(topic)
+            || (!remoteTopicSuffix.isEmpty() && topic.endsWith(remoteTopicSuffix));
+    }
+}

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 LegacyReplicationPolicy (and any other dumb 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,9 @@ 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 LegacyReplicationPolicy (and any other dumb impls) which
+            // cannot prevent cycles. In impls that do prevent cycles, we shouldn't see the same source twice.
             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-client/src/test/java/org/apache/kafka/connect/mirror/MirrorClientTest.java

@@ -37,7 +37,11 @@ private static class FakeMirrorClient extends MirrorClient {
         List<String> topics;
 
         FakeMirrorClient(List<String> topics) {
-            super(null, new DefaultReplicationPolicy(), null);
+            this(new DefaultReplicationPolicy(), topics);
+        }
+
+        FakeMirrorClient(ReplicationPolicy replicationPolicy, List<String> topics) {
+            super(null, replicationPolicy, null);
             this.topics = topics;
         }
 
@@ -131,6 +135,19 @@ public void upstreamClustersTest() throws InterruptedException {
         assertFalse(sources.contains(null));
     }
 
+    @Test
+    public void legacyUpstreamClustersTest() throws InterruptedException {
+        // When configured with a specific source, LegacyReplicationPolicy should just return the provided source.
+        MirrorClient client = new FakeMirrorClient(new LegacyReplicationPolicy("", "source"), Arrays.asList("topic1",
+            "topic2", "heartbeats", "source1.heartbeats", "source1.source2.heartbeats",
+            "source3.source4.source5.heartbeats"));
+        Set<String> sources = client.upstreamClusters();
+        assertTrue(sources.contains("source"));
+        assertFalse(sources.contains(""));
+        assertFalse(sources.contains(null));
+        assertEquals(1, sources.size());
+    }
+
     @Test
     public void remoteTopicsTest() throws InterruptedException {
         MirrorClient client = new FakeMirrorClient(Arrays.asList("topic1", "topic2", "topic3",
@@ -144,6 +161,18 @@ public void remoteTopicsTest() throws InterruptedException {
         assertTrue(remoteTopics.contains("source3.source4.source5.topic6"));
     }
 
+
+    @Test
+    public void legacyRemoteTopicsTest() throws InterruptedException {
+        // By default, LegacyReplicationPolicy should consider any topic to be remote.
+        MirrorClient client = new FakeMirrorClient(new LegacyReplicationPolicy("", "source"), Arrays.asList(
+            "topic1", "topic2", "topic3"));
+        Set<String> remoteTopics = client.remoteTopics();
+        assertTrue(remoteTopics.contains("topic1"));
+        assertTrue(remoteTopics.contains("topic2"));
+        assertTrue(remoteTopics.contains("topic3"));
+    }
+
     @Test
     public void remoteTopicsSeparatorTest() throws InterruptedException {
         MirrorClient client = new FakeMirrorClient(Arrays.asList("topic1", "topic2", "topic3",
@@ -159,4 +188,23 @@ public void remoteTopicsSeparatorTest() throws InterruptedException {
         assertTrue(remoteTopics.contains("source3__source4__source5__topic6"));
     }
 
+    @Test
+    public void testLegacyRemoteTopicWithSuffix() {
+        MirrorClient client = new FakeMirrorClient(
+            new LegacyReplicationPolicy(".replica", "primary"), Arrays.asList());
+        assertEquals("topic1.replica", client.replicationPolicy()
+            .formatRemoteTopic("primary", "topic1"));
+        assertEquals("primary", client.replicationPolicy()
+            .topicSource("topic1.replica"));
+    }
+
+    @Test
+    public void testLegacyRemoteTopicWithoutSuffix() {
+        MirrorClient client = new FakeMirrorClient(
+            new LegacyReplicationPolicy("", "primary"), Arrays.asList());
+        assertEquals("topic1", client.replicationPolicy()
+            .formatRemoteTopic("primary", "topic1"));
+        assertEquals("primary", client.replicationPolicy()
+            .topicSource("topic1"));
+    }
 }

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

@@ -492,7 +492,12 @@ boolean isCycle(String topic) {
         } else if (source.equals(sourceAndTarget.target())) {
             return true;
         } else {
-            return isCycle(replicationPolicy.upstreamTopic(topic));
+            String upstreamTopic = replicationPolicy.upstreamTopic(topic);
+            if (upstreamTopic.equals(topic)) {
+                // Extra check for LegacyReplicationPolicy
+                return false;
+            }
+            return isCycle(upstreamTopic);
         }
     }