MirrorMaker2: Check if bootstrap servers are the same for different c…

…luster aliases
strimzi · Oct 7, 2024 · 4d90a71 · 4d90a71
1 parent e2566e6
commit 4d90a71
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 17 deletions.
diff --git a/...operator/src/main/java/io/strimzi/operator/cluster/model/KafkaMirrorMaker2Connectors.java b/...operator/src/main/java/io/strimzi/operator/cluster/model/KafkaMirrorMaker2Connectors.java
@@ -119,8 +119,9 @@ public static KafkaMirrorMaker2Connectors fromCrd(Reconciliation reconciliation,
                         errorMessages.add("Target cluster alias " + mirror.getTargetCluster() + " is used in a mirror definition, but cluster with this alias does not exist in cluster definitions");
                     }
 
-                    if (!mirror.getTargetCluster().equals(connectCluster) && !hasMatchingBootstrapServers(kafkaMirrorMaker2, mirror, connectCluster)) {
-                        errorMessages.add("Connect cluster alias (currently set to " + connectCluster + ") has to be the same as the target cluster alias " + mirror.getTargetCluster());
+                    if (!mirror.getTargetCluster().equals(connectCluster) && !hasMatchingBootstrapServers(kafkaMirrorMaker2, connectCluster, mirror.getTargetCluster())) {
+                        errorMessages.add("Connect cluster alias (currently set to " + connectCluster + ") must match the target cluster alias " + mirror.getTargetCluster() + " or both clusters must have the same bootstrap servers.");
+
                     }
                 }
 
@@ -330,22 +331,19 @@ private static String addTLSConfigToMirrorMaker2ConnectorConfig(Map<String, Obje
         }
     }
 
-    private static boolean hasMatchingBootstrapServers(KafkaMirrorMaker2 kafkaMirrorMaker2, KafkaMirrorMaker2MirrorSpec mirror, String connectCluster) {
-        Map<String, String> configs = new HashMap<>();
+    private static boolean hasMatchingBootstrapServers(KafkaMirrorMaker2 kafkaMirrorMaker2, String connectClusterAlias, String targetClusterAlias) {
+        List<String> configs = new ArrayList<>();
 
-        Optional.ofNullable(kafkaMirrorMaker2)
+        configs =  Optional.ofNullable(kafkaMirrorMaker2)
                 .map(KafkaMirrorMaker2::getSpec)
                 .map(KafkaMirrorMaker2Spec::getClusters)
                 .orElse(Collections.emptyList())
-                .forEach(clusterLists -> {
-                    if (connectCluster != null && connectCluster.equals(clusterLists.getAlias())) {
-                        configs.put("connectClusterBootstrapServer", clusterLists.getBootstrapServers());
-                    }
-                    if (mirror.getTargetCluster() != null && mirror.getTargetCluster().equals(clusterLists.getAlias())) {
-                        configs.put("targetClusterBootstrapServers", clusterLists.getBootstrapServers());
-                    }
-                });
+                .stream()
+                .filter(cluster -> connectClusterAlias.equals(cluster.getAlias()) || targetClusterAlias.equals(cluster.getAlias()))
+                .map(KafkaMirrorMaker2ClusterSpec::getBootstrapServers)
+                .toList();
+
+        return configs.size() == 2 && configs.get(0).equals(configs.get(1)) ;
 
-        return configs.get("connectClusterBootstrapServer").equals(configs.get("targetClusterBootstrapServers"));
     }
 }
diff --git a/...ator/src/test/java/io/strimzi/operator/cluster/model/KafkaMirrorMaker2ConnectorsTest.java b/...ator/src/test/java/io/strimzi/operator/cluster/model/KafkaMirrorMaker2ConnectorsTest.java
@@ -118,7 +118,7 @@ public void testFailingValidation()    {
         assertThat(ex.getMessage(), is("KafkaMirrorMaker2 resource validation failed: " +
                 "[Each MirrorMaker 2 mirror definition has to specify the source cluster alias, " +
                 "Target cluster alias wrong-target is used in a mirror definition, but cluster with this alias does not exist in cluster definitions, " +
-                "Connect cluster alias (currently set to target) has to be the same as the target cluster alias wrong-target]"));
+                "Connect cluster alias (currently set to target) must match the target cluster alias wrong-target or both clusters must have the same bootstrap servers.]"));
     }
 
     @Test
@@ -131,7 +131,7 @@ public void testMirrorTargetClusterNotSameAsConnectCluster() {
                 .build();
         InvalidResourceException ex = assertThrows(InvalidResourceException.class, () -> KafkaMirrorMaker2Connectors.validateConnectors(kmm2));
         assertThat(ex.getMessage(), is("KafkaMirrorMaker2 resource validation failed: " +
-                "[Connect cluster alias (currently set to source) has to be the same as the target cluster alias target]"));
+                "[Connect cluster alias (currently set to source) must match the target cluster alias target or both clusters must have the same bootstrap servers.]"));
 
         // A case where one mirror has the correct target cluster, but the other does not
         KafkaMirrorMaker2 kmm2CorrectAndIncorrectMirror = new KafkaMirrorMaker2Builder(KMM2)
@@ -147,7 +147,7 @@ public void testMirrorTargetClusterNotSameAsConnectCluster() {
                 .build();
         ex = assertThrows(InvalidResourceException.class, () -> KafkaMirrorMaker2Connectors.validateConnectors(kmm2CorrectAndIncorrectMirror));
         assertThat(ex.getMessage(), is("KafkaMirrorMaker2 resource validation failed: " +
-                "[Connect cluster alias (currently set to target) has to be the same as the target cluster alias third]"));
+                "[Connect cluster alias (currently set to target) must match the target cluster alias third or both clusters must have the same bootstrap servers.]"));
     }