You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/10/24 12:34:52 UTC
[GitHub] [kafka] dongjinleekr opened a new pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
dongjinleekr opened a new pull request #11431:
URL: https://github.com/apache/kafka/pull/11431
I found this issue while working on [KAFKA-13365](https://issues.apache.org/jira/browse/KAFKA-13365). Since [KAFKA-10777](https://issues.apache.org/jira/browse/KAFKA-10777) is scheduled to be shipped with 3.1.0, this issue would be better to be fixed.
cc/ @ryannedolan @mimaison @OmniaGM
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
OmniaGM commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r737592730
##########
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:
On second thought, I think Connect API's internal topics should not be impacted by the separator at all (check my comment above for more details).
Customers have already a way to override Connect API's topics. However, I recommend keeping the fix simple and updating `isInternalTopic` to the following.
```java
/** Internal topics are never replicated.
Internal topics are Kafka internal topic, Connect API's internal topics, MM2 offset-syncs and checkpoints topics.
*/
default boolean isInternalTopic(String topic) {
boolean isKafkaInternalTopic = topic.startsWith("__") || topic.startsWith(".") || topic.endsWith("-internal") || topic.endsWith(".internal");
return isMM2InternalTopic(topic) || isKafkaInternalTopic;
}
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mimaison commented on pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
mimaison commented on pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#issuecomment-977746470
Backported to 3.1: https://github.com/apache/kafka/commit/417630d9be91bc62b908d8337fe7fc937eca823c
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mimaison commented on pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
mimaison commented on pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#issuecomment-971789372
@dajac Can we apply with on 3.1? Without this fix, when using MirrorMaker in dedicated mode and overriding the replication policy separator, Connect internal topics get mirrored.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@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
Posted by GitBox <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
OmniaGM commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r737592730
##########
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:
On second thought, I don't think connect's internal topics should be controlled by the separator at all. Please check my comment above.
Customers have already a way to override connect's topics, for keeping the backward compatibility I just recommend keeping the fix simple and updating `isInternalTopic` to the following
```java
/** Internal topics are never replicated. */
default boolean isInternalTopic(String topic) {
boolean isKafkaInternalTopic = topic.startsWith("__") || topic.startsWith(".") || topic.endsWith("-internal") || topic.endsWith(".internal");
return isMM2InternalTopic(topic) || isKafkaInternalTopic;
}
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
OmniaGM commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r737584380
##########
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:
> mm2-offsets.{source}.internal includes - so allowing dash as a separator may be problematic
I am still not sure why this is problematic?
In your Jira, you mentioned the following
> MirrorMaker2 creates internal topics to track the offsets, configs, and status of the MM2 tasks. But, these topics are not affected by a custom 'replication.policy.separator' settings - that is, these topics may be replicated against the user`s intention.
This issue can easily be fixed by handling `.internal` in `isInternalTopic` default implementation. I would suggest just doing this small fix instead.
I don't think we need to use the separator or replication policy to define any of connect's topics as users have already a way to override them using `<CLUSTER_ALIAS>.config.storage.topic`, `<CLUSTER_ALIAS>.offset.storage.topic` and `<CLUSTER_ALIAS>.status.storage.topic` configs. No point of controlling the same config in two different ways.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
OmniaGM commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r739870251
##########
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:
> only when MM2 is running in standalone mode.
They are created in any mode if there is no value for DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG
> 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?
Small clarification, users can use `DistributedConfig` with any mode (even standalone) to override the name of these topics.
And they always had the power to do so, and if the topic name didn't match the `isInternalTopic` policy, it would replicate anyway. The main problem I am seeing is preventing the unintentional replication of these topics for users who use default names for `DistributedConfig`'s topics, for example `mm2-offsets.{source}.internal` with other configurations that would make these topics not match the default implementation of `isInternalTopic` any more after upgrading to 3.1 which includes KIP-690.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dongjinleekr commented on pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
dongjinleekr commented on pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#issuecomment-955660566
So, We can now organize the issues like the following:
1. Preventing the malfunctioning (i.e., replication of MM2 internal topics)
2. Keeping consistency (as far as possible.)
3. Maintaining backward compatibility (we MUST.)
My approach focuses on standalone mode with 1 and 2, and @OmniaGM 's approach (i.e., fixing `isInternalTopic` method) focuses on connect mode with 1 and 3. If we agree on this, we are now on the same page. 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@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
Posted by GitBox <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
OmniaGM commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r739870251
##########
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:
> only when MM2 is running in standalone mode.
They are created in any mode if there is no value for DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG
> 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?
Yes, in general, users always had the power to override the name of these topics, and if the topic name didn't match the `isInternalTopic` policy, it would replicate anyway. The main problem I am seeing is preventing the unintentional replication of these topics for users who use default names for these topics, for example'mm2-offsets.{source}.internal` and setup diff value for the separator after upgrading to 3.1 which includes KIP-690.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@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
Posted by GitBox <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mimaison commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
mimaison commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r742038737
##########
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:
My bad, I missed the fact that this line is about the Kafka Connect's internal topics and not MM2's. If I understand correctly this behaviour is not new and it's not related to changes made in KIP-690.
I agree it's unfortunate that if users change the default separator, these Connect internal topics get mirrored in dedicated mode.
In Connect mode, this could also happen. For example, if `offset.storage.topic` is not set to a value that is caught by `isInternalTopic()`.
If we wanted to use the separator for Connect's topic, I'd prefer doing it via a small KIP. As it's likely been like this since day 1, this seems this may not be a very common use case and I'm not sure we necessarily need to do something. If a user in dedicated mode wants a different separator, they will need to also override the Connect topic names to ensure they match the separator.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
OmniaGM commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r739869555
##########
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:
Yes, most production use-cases will be using MM2 with Connect mode. However, we need to prevent the unintentional replication of these topics for users who either didn't set up their connect's topics or those who run standalone mode.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
OmniaGM commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r737584380
##########
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:
> mm2-offsets.{source}.internal includes - so allowing dash as a separator may be problematic
I am still not sure why this is problematic?
In your Jira, you mentioned the following
> MirrorMaker2 creates internal topics to track the offsets, configs, and status of the MM2 tasks. But, these topics are not affected by a custom 'replication.policy.separator' settings - that is, these topics may be replicated against the user`s intention.
This issue can easily be fixed by handling `.internal` in `isInternalTopic` default implementation.
I don't think we need to use the separator or replication policy to define any of connect's topics as users have already a way to override them using `<CLUSTER_ALIAS>.config.storage.topic`, `<CLUSTER_ALIAS>.offset.storage.topic` and `<CLUSTER_ALIAS>.status.storage.topic` configs.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
OmniaGM commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r739870251
##########
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:
Yes. The main problem am seeing is preventing the unintentional replication of these topics for users who either didn't set up their connect's topics or those who run standalone/dedicated cluster mode.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mimaison merged pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
mimaison merged pull request #11431:
URL: https://github.com/apache/kafka/pull/11431
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
OmniaGM commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r739869555
##########
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:
Yes, most production use-cases will be using MM2 with connect mode. We just need to protect users who either didn't bother by setting up their connect topics, or those who run standalone mode.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
OmniaGM commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r739870251
##########
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:
> only when MM2 is running in standalone mode.
They are created in any mode if there is no value for DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG
> 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?
Small clarification, users can use `DistributedConfig` with any mode (even standalone) to override the name of these topics.
And they always had the power to do so, even before KIP-690, and if this new topic name didn't match the `isInternalTopic` policy, it would replicate.
The PR's approach is trying to control the Connect topics that MM2 needs to set up using the separator; this is where I am not sure it's a minor fix or something that requires a KIP that follows KIP-690.
My suggestion, is to introduce the minor fix first and propose another KIP if you believe that Connect internal topics created by MM2 Workers should to be controlled by the separator as well.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@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
Posted by GitBox <gi...@apache.org>.
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` 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mimaison commented on pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
mimaison commented on pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#issuecomment-971432869
@dongjinleekr I think it's probably something we want in 3.1 otherwise KIP-690 may break some users, WDYT?
cc @dajac
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
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 be a dash and throw an exception. This check seems to me like an assumption about the naming convention of a topic which is why we moved internal topics to `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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dongjinleekr commented on pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
dongjinleekr commented on pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#issuecomment-971418712
@mimaison Here is the fix.
I added `ReplicationPolicyTest` to validate the implementation. It seems like we need to test the other methods but, let's focus on `DefaultReplicationPolicy#isInternalTopic` only in this PR. (I will open a separate PR for it later.)
@OmniaGM Thanks for suggesting the right alternative. :bow:
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
OmniaGM commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r739499675
##########
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's internal topics.
`DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG` is one of the connect's internal topics.
```
private static final String OFFSET_STORAGE_TOPIC_CONFIG_DOC = "The name of the Kafka topic where connector offsets are stored";
```
My point is users already can control these types of topics using the `DistributedConfig` so there's no point in controlling them again using the separator. The main issue I think we need to fix first is preventing is the replication of these 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@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
Posted by GitBox <gi...@apache.org>.
dongjinleekr commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r739777741
##########
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
OmniaGM commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r739499675
##########
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's internal topics.
`DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG` is one of the connect's internal topics.
```
private static final String OFFSET_STORAGE_TOPIC_CONFIG_DOC = "The name of the Kafka topic where connector offsets are stored";
```
My point is customers already can control these types of topics using the `DistributedConfig` so there's no point in controlling them again using the seprator. The main issue I think we need to fix first is preventing is the replication of these 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@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
Posted by GitBox <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mimaison commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
mimaison commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r742038737
##########
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:
My bad, I missed the fact that this line is about the Kafka Connect's internal topics and not MM2's. I think it's fine to keep the code as is.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mimaison commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
mimaison commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r742038737
##########
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:
My bad, I missed the fact that this line is about the Kafka Connect's internal topics and not MM2's. I think it's fine to keep the code as is.
##########
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:
Yeah but I expect the value for this config to come from a class implementing the interface so in this case `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:
My bad, I missed the fact that this line is about the Kafka Connect's internal topics and not MM2's. If I understand correctly this behaviour is not new and it's not related to changes made in KIP-690.
I agree it's unfortunate that if users change the default separator, these Connect internal topics get mirrored in dedicated mode.
In Connect mode, this could also happen. For example, if `offset.storage.topic` is not set to a value that is caught by `isInternalTopic()`.
If we wanted to use the separator for Connect's topic, I'd prefer doing it via a small KIP. As it's likely been like this since day 1, this seems this may not be a very common use case and I'm not sure we necessarily need to do something. If a user in dedicated mode wants a different separator, they will need to also override the Connect topic names to ensure they match the separator.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mimaison commented on pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
mimaison commented on pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#issuecomment-971790860
@dongjinleekr I'll rename KAFKA-13397 to match the fix. Then if you want to propose a further KIP/PR to actually honor the separator we can use a new JIRA.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mimaison commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
mimaison commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r742038737
##########
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:
My bad, I missed the fact that this line is about the Kafka Connect's internal topics and not MM2's. I think it's fine to keep the code as is.
##########
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:
Yeah but I expect the value for this config to come from a class implementing the interface so in this case `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:
My bad, I missed the fact that this line is about the Kafka Connect's internal topics and not MM2's. If I understand correctly this behaviour is not new and it's not related to changes made in KIP-690.
I agree it's unfortunate that if users change the default separator, these Connect internal topics get mirrored in dedicated mode.
In Connect mode, this could also happen. For example, if `offset.storage.topic` is not set to a value that is caught by `isInternalTopic()`.
If we wanted to use the separator for Connect's topic, I'd prefer doing it via a small KIP. As it's likely been like this since day 1, this seems this may not be a very common use case and I'm not sure we necessarily need to do something. If a user in dedicated mode wants a different separator, they will need to also override the Connect topic names to ensure they match the separator.
##########
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:
My bad, I missed the fact that this line is about the Kafka Connect's internal topics and not MM2's. I think it's fine to keep the code as is.
##########
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:
Yeah but I expect the value for this config to come from a class implementing the interface so in this case `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:
My bad, I missed the fact that this line is about the Kafka Connect's internal topics and not MM2's. If I understand correctly this behaviour is not new and it's not related to changes made in KIP-690.
I agree it's unfortunate that if users change the default separator, these Connect internal topics get mirrored in dedicated mode.
In Connect mode, this could also happen. For example, if `offset.storage.topic` is not set to a value that is caught by `isInternalTopic()`.
If we wanted to use the separator for Connect's topic, I'd prefer doing it via a small KIP. As it's likely been like this since day 1, this seems this may not be a very common use case and I'm not sure we necessarily need to do something. If a user in dedicated mode wants a different separator, they will need to also override the Connect topic names to ensure they match the separator.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@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
Posted by GitBox <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
OmniaGM commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r739870251
##########
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:
> only when MM2 is running in standalone mode.
They are created in any mode if there is no value for DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG
> 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?
Yes, in general, users always had the power to override the name of these topics, and if the topic name didn't match the `isInternalTopic` policy, it would replicate anyway. The main problem I am seeing is preventing the unintentional replication of these topics for users who use default names for these topics, for example `mm2-offsets.{source}.internal` and setup diff value for the separator after upgrading to 3.1 which includes KIP-690.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
OmniaGM commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r739870251
##########
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:
> only when MM2 is running in standalone mode.
They are created in any mode if there is no value for DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG
> 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?
Yes, in general, users always had the power to override the name of these topics and if the topic name didn't match the `isInternalTopic` policy it will replicate anyway. The main problem am seeing is preventing the unintentional replication of these topics for users who use default names for these topics for example`mm2-offsets.{source}.internal` and setup diff value for the separator.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
OmniaGM commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r739494732
##########
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:
Users can make it consistent using connect API's configuration. For now, to keep the backward compatibility for the most use-cases, let's fix `isInternalTopic`
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
OmniaGM commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r737592730
##########
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:
On second thought, I think Connect API's internal topics should not be impacted by the separator at all (check my comment above for more details).
Customers have already a way to override Connect API's topics. However, I recommend keeping the fix simple and updating `isInternalTopic` to the following.
```java
/** Internal topics are never replicated.
Internal topics are Kafka internal topic, Connect API's internal topics, MM2 offset-syncs and checkpoints topics.
*/
default boolean isInternalTopic(String topic) {
boolean isKafkaInternalTopic = topic.startsWith("__") || topic.startsWith(".")
boolean isDefaultConnectTopic = topic.endsWith("-internal") || topic.endsWith(".internal");
return isMM2InternalTopic(topic) || isKafkaInternalTopic || isDefaultConnectTopic;
}
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
OmniaGM commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r739869555
##########
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:
Yes. The main problem am seeing is preventing the unintentional replication of these topics for users who either didn't set up their connect's topics or those who run standalone/dedicated cluster mode.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mimaison commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
mimaison commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r749214261
##########
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 think the fix @OmniaGM proposed above (https://github.com/apache/kafka/pull/11431#discussion_r737592730) makes sense and I think it's fully compatible. WDYT?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
OmniaGM commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r737592730
##########
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:
On second thought, I think Connect API's internal topics should not be impacted by the separator at all (check my comment above for more details).
Customers have already a way to override Connect API's topics. However, I recommend keeping the fix simple and updating `isInternalTopic` to the following.
```java
/** Internal topics are never replicated.
Internal topics are Kafka internal topic, Connect API's internal topics, MM2 offset-syncs and checkpoints topics.
*/
default boolean isInternalTopic(String topic) {
boolean isKafkaInternalTopic = topic.startsWith("__") || topic.startsWith(".")
boolean isConnectTopic = topic.endsWith("-internal") || topic.endsWith(".internal");
return isMM2InternalTopic(topic) || isKafkaInternalTopic || isConnectTopic;
}
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
OmniaGM commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r737592730
##########
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:
On second thought, I think Connect API's internal topics should not be impacted by the separator at all (check my comment above for more details).
Customers have already a way to override Connect API's topics. However, I recommend keeping the fix simple and updating `isInternalTopic` to the following.
```java
/** Internal topics are never replicated.
Internal topics are Kafka internal topic, Connect API's internal topics, MM2 offset-syncs and checkpoints topics.
*/
default boolean isInternalTopic(String topic) {
boolean isKafkaInternalTopic = topic.startsWith("__") || topic.startsWith(".")
boolean isConnectTopic = topic.endsWith("-internal") || topic.endsWith(".internal");
return isMM2InternalTopic(topic) || isKafkaInternalTopic || isConnectTopic;
}
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@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
Posted by GitBox <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mimaison commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
mimaison commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r735543104
##########
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:
Isn't it better to keep the separator in `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("-")) {
Review comment:
We did not have this check before, why is it needed? Also checks here are only applied when running in "driver" mode.
##########
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:
Should we actually try to use the configured `ReplicationPolicy` here instead of generating the topic names? What happens if you provide a custom policy in driver mode?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
OmniaGM commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r739870251
##########
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:
> only when MM2 is running in standalone mode.
They are created in any mode if there is no value for DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG
> 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?
Small clarification, users can use `DistributedConfig` with any mode (even standalone) to override the name of these topics.
And they always had the power to do so, and even without KIP-690 if this new topic name didn't match the `isInternalTopic` policy, it would replicate.
The PR's approach is trying to control the Connect topics that MM2 needs to set up using the separator; this is where I am not sure it's a minor fix or something that requires a KIP that follows KIP-690.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
OmniaGM commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r737584380
##########
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:
> mm2-offsets.{source}.internal includes - so allowing dash as a separator may be problematic
I am still not sure why this is problematic?
In your Jira, you mentioned the following
> MirrorMaker2 creates internal topics to track the offsets, configs, and status of the MM2 tasks. But, these topics are not affected by a custom 'replication.policy.separator' settings - that is, these topics may be replicated against the user`s intention.
This issue can easily be fixed by handling `.internal` in `isInternalTopic` default implementation. I would suggest just doing this small fix instead.
I don't think we need to use the separator or replication policy to define any of connect's topics as users have already a way to override them using `<CLUSTER_ALIAS>.config.storage.topic`, `<CLUSTER_ALIAS>.offset.storage.topic` and `<CLUSTER_ALIAS>.status.storage.topic` configs.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@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
Posted by GitBox <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
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 be a dash and throw an exception. This check seems to me like an assumption about the naming convention of a topic.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
OmniaGM commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r739870251
##########
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:
> only when MM2 is running in standalone mode.
They are created in any mode if there is no value for DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG
> 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?
Small clarification, users can use `DistributedConfig` with any mode (even standalone) to override the name of these topics.
And they always had the power to do so, and if the topic name didn't match the `isInternalTopic` policy, it would replicate anyway. The main problem I am seeing is preventing the unintentional replication of these topics for users who use default names for `DistributedConfig`'s topics, for example `mm2-offsets.{source}.internal` with other configurations that would make these topics not matching `isInternalTopic` any more after upgrading to 3.1 which includes KIP-690.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] OmniaGM commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
OmniaGM commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r739870251
##########
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:
> only when MM2 is running in standalone mode.
They are created in any mode if there is no value for DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG
> 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?
Small clarification, users can use `DistributedConfig` with any mode (even standalone) to override the name of these topics.
And they always had the power to do so, even before KIP-690 and if this new topic name didn't match the `isInternalTopic` policy, it would replicate.
The PR's approach is trying to control the Connect topics that MM2 needs to set up using the separator; this is where I am not sure it's a minor fix or something that requires a KIP that follows KIP-690.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mimaison commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
mimaison commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r742040313
##########
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:
Yeah but I expect the value for this config to come from a class implementing the interface so in this case `DefaultReplicationPolicy`.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@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
Posted by GitBox <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@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
Posted by GitBox <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dongjinleekr commented on pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
dongjinleekr commented on pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#issuecomment-971474599
@mimaison @dajac Whatever it is, I will follow your (plural) decision; If you think we need a hotfix KIP, I can do it by this Friday.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mimaison commented on a change in pull request #11431: KAFKA-13397: Honor 'replication.policy.separator' configuration when creating MirrorMaker2 internal topics
Posted by GitBox <gi...@apache.org>.
mimaison commented on a change in pull request #11431:
URL: https://github.com/apache/kafka/pull/11431#discussion_r744751390
##########
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.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org