KAFKA-9914: Fix replication cycle detection (#22079)

Fix replication cycle detection to detect recursive topic cycles if
kafka cluster naming differs between multiple MM2 instances with special
handling for IdentityReplicationPolicy.

Reviewers: Viktor Somogyi-Vass <viktorsomogyi@gmail.com>

---------

Co-authored-by: Ivan Yurchenko <ivanyu@aiven.io>
Co-authored-by: Viktor Somogyi-Vass <viktorsomogyi@gmail.com>

Author: 3 people

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

@@ -720,7 +720,16 @@ boolean isCycle(String topic) {
         String source = replicationPolicy.topicSource(topic);
         if (source == null) {
             return false;
-        } else if (source.equals(sourceAndTarget.target())) {
+        }
+
+        final boolean condition;
+        if (replicationPolicy instanceof IdentityReplicationPolicy) {
+            condition = source.equals(sourceAndTarget.target());
+        } else {
+            condition = source.equals(sourceAndTarget.source()) || source.equals(sourceAndTarget.target());
+        }
+
+        if (condition) {
             return true;
         } else {
             String upstreamTopic = replicationPolicy.upstreamTopic(topic);

connect/mirror/src/test/java/org/apache/kafka/connect/mirror/MirrorSourceConnectorTest.java

@@ -130,13 +130,31 @@ public void testReplicatesHeartbeatsWhenDisabledButFilterAllows() {
     public void testNoCycles() {
         MirrorSourceConnector connector = new MirrorSourceConnector(new SourceAndTarget("source", "target"),
             new DefaultReplicationPolicy(), x -> true, getConfigPropertyFilter());
+        assertFalse(connector.shouldReplicateTopic("source.topic1"), "should not allow cycles");
         assertFalse(connector.shouldReplicateTopic("target.topic1"), "should not allow cycles");
         assertFalse(connector.shouldReplicateTopic("target.source.topic1"), "should not allow cycles");
         assertFalse(connector.shouldReplicateTopic("source.target.topic1"), "should not allow cycles");
         assertFalse(connector.shouldReplicateTopic("target.source.target.topic1"), "should not allow cycles");
         assertFalse(connector.shouldReplicateTopic("source.target.source.topic1"), "should not allow cycles");
         assertTrue(connector.shouldReplicateTopic("topic1"), "should allow anything else");
-        assertTrue(connector.shouldReplicateTopic("source.topic1"), "should allow anything else");
+        assertTrue(connector.shouldReplicateTopic("othersource.topic1"), "should allow anything else");
+        assertTrue(connector.shouldReplicateTopic("othertarget.topic1"), "should allow anything else");
+        assertTrue(connector.shouldReplicateTopic("other.another.topic1"), "should allow anything else");
+
+        final IdentityReplicationPolicy identityReplicationPolicy = new IdentityReplicationPolicy();
+        final HashMap<String, String> props = new HashMap<>();
+        props.put("source.cluster.alias", "source");
+        identityReplicationPolicy.configure(props);
+        connector = new MirrorSourceConnector(new SourceAndTarget("source", "target"),
+            identityReplicationPolicy, x -> true, x -> true);
+        assertTrue(connector.shouldReplicateTopic("source.topic1"), "should not consider this a cycle");
+        assertTrue(connector.shouldReplicateTopic("target.topic1"), "should not consider this a cycle");
+        assertTrue(connector.shouldReplicateTopic("target.source.topic1"), "should not consider this a cycle");
+        assertTrue(connector.shouldReplicateTopic("source.target.topic1"), "should not consider this a cycle");
+        assertTrue(connector.shouldReplicateTopic("topic1"), "should not consider this a cycle");
+        assertTrue(connector.shouldReplicateTopic("othersource.topic1"), "should not consider this a cycle");
+        assertTrue(connector.shouldReplicateTopic("othertarget.topic1"), "should not consider this a cycle");
+        assertTrue(connector.shouldReplicateTopic("other.another.topic1"), "should not consider this a cycle");
     }
 
     @Test