[GitHub] [kafka] dongjinleekr commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
dongjinleekr commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r751078367
##
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:
Okay. Thanks for your patience in convincing me.
Then, 1) Making the default internal topic names (e.g.,
`DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG`) in the standalone mode
consistent 2) Provide a way to migrate internal topics are beyond the scope of
this issue, and it should be filed as a separate KIP. Right?
--
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
[GitHub] [kafka] dongjinleekr commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
dongjinleekr commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r745085224
##
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:
> One issue with c. is that it works for new environments. If users
already have MM2 running, it's using topics with the current names.
Yes, I already thought of this backward compatibility issue - how about
this?: **"If the user is using a custom separator, no MM2 internal topics are
created yet, and there are already default MM2 internal topics, migrate the
default MM2 internal topic contents into the new one."**
With this strategy, the users already running MM2 can change the
configuration without any issues and manual work.
Plus, it is also good for the `MirrorClient` to export/import the MM2
internal topics. This strategy is not automatic but, it gives the users more
flexibility - it handles the separator change from a custom one to the other
custom one.
How about these approaches?
--
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
[GitHub] [kafka] dongjinleekr commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
dongjinleekr commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r743692427
##
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:
I have also thought over this issue again.
The core of the problem seems to be, there is some gap in our configuration
policy, and it causes the Kafka Connect internal topics may be mirrored into
the other cluster. At first, I thought it could only happen in standalone mode,
but it is not true - my apologies for the confusion.
There are two cases this problem may happen:
1. Using a custom separator in standalone mode after KIP-690 (i.e., AK
3.1.0), what I and @OmniaGM discussed until now.
2. Running MM2 in Kafka Connect with the default `[config, offset,
status].storage.topic` configuration. (what @mimaison just pointed out)
2 may be prevented by `topics` configuration override, and it seems like the
users are already doing so. (Yes, it has been like this since day 1.) So, let's
focus on 1 only.
To prevent those topics from being mirrored with a custom separator, there
are three approaches:
a. Fix `ReplicationPolicy#isInternalTopic` only.
b. Leave the users to override `topics` configuration like Kafka Connect
mode.
c. Make `mm2-[offsets, status, configs].{source}.internal` topics to use
given separator.
Approach a seems unavailable; to determine whether given `mm2-[offsets,
status, configs].{source}.internal` topic is an internal one,
`ReplicationPolicy#isInternalTopic` method should know the source cluster's
alias, since those topics include `{source}` in their name. But actually, it
takes a topic name as a parameter only.
Approach b is good, but it may be not very clear for the users with little
experience.
Removing approach a and b, only c survives - and it is why I thought
applying the separator to those topic names is the only way to make it work
identically with the default separator case (which does not mirror those topics
by default).
How do you think?
##
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:
I have also thought over this issue again.
The core of the problem seems to be, there is some gap in our configuration
policy, and it causes the Kafka Connect internal topics may be mirrored into
the other cluster. At first, I thought it could only happen in standalone mode,
but it is not true - my apologies for the confusion.
There are two cases this problem may happen:
1. Using a custom separator in standalone mode after KIP-690 (i.e., AK
3.1.0), what I and @OmniaGM discussed until now.
2. Running MM2 in Kafka Connect with the default `[config, offset,
status].storage.topic
[GitHub] [kafka] dongjinleekr commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
dongjinleekr commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r743692427
##
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:
I have also thought over this issue again.
The core of the problem seems to be, there is some gap in our configuration
policy, and it causes the Kafka Connect internal topics may be mirrored into
the other cluster. At first, I thought it could only happen in standalone mode,
but it is not true - my apologies for the confusion.
There are two cases this problem may happen:
1. Using a custom separator in standalone mode after KIP-690 (i.e., AK
3.1.0), what I and @OmniaGM discussed until now.
2. Running MM2 in Kafka Connect with the default `[config, offset,
status].storage.topic` configuration. (what @mimaison just pointed out)
2 may be prevented by `topics` configuration override, and it seems like the
users are already doing so. (Yes, it has been like this since day 1.) So, let's
focus on 1 only.
To prevent those topics from being mirrored with a custom separator, there
are three approaches:
a. Fix `ReplicationPolicy#isInternalTopic` only.
b. Leave the users to override `topics` configuration like Kafka Connect
mode.
c. Make `mm2-[offsets, status, configs].{source}.internal` topics to use
given separator.
Approach a seems unavailable; to determine whether given `mm2-[offsets,
status, configs].{source}.internal` topic is an internal one,
`ReplicationPolicy#isInternalTopic` method should know the source cluster's
alias, since those topics include `{source}` in their name. But actually, it
takes a topic name as a parameter only.
Approach b is good, but it may be not very clear for the users with little
experience.
Removing approach a and b, only c survives - and it is why I thought
applying the separator to those topic names is the only way to make it work
identically with the default separator case (which does not mirror those topics
by default).
How do you think?
--
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
[GitHub] [kafka] dongjinleekr commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
dongjinleekr commented on a change in pull request #11431: URL: https://github.com/apache/kafka/pull/11431#discussion_r742541439 ## 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: Got it. -- 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
[GitHub] [kafka] dongjinleekr commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
dongjinleekr commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r740270998
##
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:
> They are created in any mode if there is no value for
`DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG`
Yes, I agree. I mean, it is created as `mm2-offsets.{source}.internal` if
run in standalone mode, and `DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG` is
not explicitly set. (You know, there is a default value of it in Kafka Connect.)
> The PR's approach is trying to control the Connect topics that MM2 needs
to set up using the separator.
Totally agree.
> This is where I am not sure it's a minor fix or something that requires a
KIP that follows KIP-690.
Yes, I am also not sure whether this issue is fixing an unintended, omitted
function or proposing a new interface. @mimaison @ryannedolan How do you think?
--
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
[GitHub] [kafka] dongjinleekr commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
dongjinleekr commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r73941
##
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:
> DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG is one of the connect's
internal topics.
Oh, sorry for my misunderstanding; Yes, you are right;
`DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG` is one of Kafka Connect's
internal topics but, it is created as `mm2-offsets.{source}.internal` **only
when MM2 is running in standalone mode.** (see:
`MirrorMakerConfig.workerConfig`) My apologies.
If the user is running MM2 in connect mode, the user is responsible for
configuring `DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG`,
`DistributedConfig.CONFIG_TOPIC_CONFIG`, etc. It is what you are meaning. Right?
--
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
[GitHub] [kafka] dongjinleekr commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
dongjinleekr commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r738940752
##
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:
> This issue can easily be fixed by handling `.internal` in
`isInternalTopic` default implementation.
I agree. But in that case, the mm2 internal topics (offsets, configs,
status) will not be consistent with the other topics and
'replication.policy.separator' configuration. Isn't it?
--
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
[GitHub] [kafka] dongjinleekr commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
dongjinleekr commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r738939808
##
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:
Oh no, this lines replace the original
`props.putIfAbsent(DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG,
"mm2-offsets." + sourceAndTarget.source() + ".internal");`, etc. not Connect
API's internal topics.
--
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
[GitHub] [kafka] dongjinleekr commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
dongjinleekr commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r736291869
##
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:
As I described in Jira, the MM2 offset/config/status topics like
`mm2-offsets.{source}.internal` includes `-` so allowing dash as a separator
may be problematic. How do you think?
##
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:
So, You mean moving topic name generating logic into `ReplicationPolicy`
would be better. Do I understand correctly?
--
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
[GitHub] [kafka] dongjinleekr commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
dongjinleekr commented on a change in pull request #11431: URL: https://github.com/apache/kafka/pull/11431#discussion_r736258242 ## 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 thought at first but, `REPLICATION_POLICY_CLASS_DEFAULT`, `REPLICATION_POLICY_SEPARATOR`, etc are defined here, and `DefaultReplicationPolicy` has the copies of them. I thought this way is better for consistency. How do you think? -- 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
