OmniaGM commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r735819464



##########
File path: 
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorMakerConfig.java
##########
@@ -183,12 +187,18 @@ public MirrorClientConfig clientConfig(String cluster) {
 
         // fill in reasonable defaults
         props.putIfAbsent(GROUP_ID_CONFIG, sourceAndTarget.source() + "-mm2");
-        props.putIfAbsent(DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG, 
"mm2-offsets."
-                + sourceAndTarget.source() + ".internal");
-        props.putIfAbsent(DistributedConfig.STATUS_STORAGE_TOPIC_CONFIG, 
"mm2-status."
-                + sourceAndTarget.source() + ".internal");
-        props.putIfAbsent(DistributedConfig.CONFIG_TOPIC_CONFIG, "mm2-configs."
-                + sourceAndTarget.source() + ".internal");
+
+        String separator = 
originalsStrings().getOrDefault(REPLICATION_POLICY_SEPARATOR, 
REPLICATION_POLICY_SEPARATOR_DEFAULT);
+        if (separator.equals("-")) {

Review comment:
       Am not sure I got why we need to check that separator can't b a dash and 
throw an exception. This check seems to me like an assumption about the naming 
convention of a topic. 

##########
File path: 
connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/MirrorClientConfig.java
##########
@@ -50,8 +50,7 @@
     public static final Class<?> REPLICATION_POLICY_CLASS_DEFAULT = 
DefaultReplicationPolicy.class;
     public static final String REPLICATION_POLICY_SEPARATOR = 
"replication.policy.separator";
     private static final String REPLICATION_POLICY_SEPARATOR_DOC = "Separator 
used in remote topic naming convention.";
-    public static final String REPLICATION_POLICY_SEPARATOR_DEFAULT =
-        DefaultReplicationPolicy.SEPARATOR_DEFAULT;
+    public static final String REPLICATION_POLICY_SEPARATOR_DEFAULT = ".";

Review comment:
       I agree with @mimaison the `REPLICATION_POLICY_SEPARATOR_DEFAULT` is 
related only to `DefaultReplicationPolicy`. 

##########
File path: 
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorMakerConfig.java
##########
@@ -183,12 +187,18 @@ public MirrorClientConfig clientConfig(String cluster) {
 
         // fill in reasonable defaults
         props.putIfAbsent(GROUP_ID_CONFIG, sourceAndTarget.source() + "-mm2");
-        props.putIfAbsent(DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG, 
"mm2-offsets."
-                + sourceAndTarget.source() + ".internal");
-        props.putIfAbsent(DistributedConfig.STATUS_STORAGE_TOPIC_CONFIG, 
"mm2-status."
-                + sourceAndTarget.source() + ".internal");
-        props.putIfAbsent(DistributedConfig.CONFIG_TOPIC_CONFIG, "mm2-configs."
-                + sourceAndTarget.source() + ".internal");
+
+        String separator = 
originalsStrings().getOrDefault(REPLICATION_POLICY_SEPARATOR, 
REPLICATION_POLICY_SEPARATOR_DEFAULT);
+        if (separator.equals("-")) {
+            throw new ConfigException("You should not use a single dash as a " 
+ REPLICATION_POLICY_SEPARATOR);
+        }
+
+        props.putIfAbsent(DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG, 
"mm2-offsets" + separator

Review comment:
       Why are we generating the connect internal topic names here? if there's 
a rule for the topic naming convention it should be defined in one place which 
is `ReplicationPolicy`. 




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


Reply via email to