mimaison commented on a change in pull request #11220:
URL: https://github.com/apache/kafka/pull/11220#discussion_r721467669
##
File path:
connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/ReplicationPolicy.java
##
@@ -52,9 +52,39 @@ default String originalTopic(String topic) {
}
}
+/** Returns heartbeats topic name.*/
+default String heartbeatsTopic() {
+return "heartbeats";
+}
+
+/** Returns the offset-syncs topic for given cluster alias. */
+default String offsetSyncTopic(String clusterAlias) {
+return "mm2-offset-syncs." + clusterAlias + ".internal";
+}
+
+/** Returns the name checkpoint topic for given cluster alias. */
+default String checkpointTopic(String clusterAlias) {
+return clusterAlias + ".checkpoint.internal";
+}
+
+/** check if topic is a heartbeat topic, e.g heartbeats,
us-west.heartbeats. */
+default boolean isHeartbeatTopic(String topic) {
Review comment:
Should we call this method `isHeartbeatsTopic` instead, so it lines up
with `heartbeatsTopic`?
##
File path:
connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/ReplicationPolicy.java
##
@@ -52,9 +52,39 @@ default String originalTopic(String topic) {
}
}
+/** Returns heartbeats topic name.*/
+default String heartbeatsTopic() {
+return "heartbeats";
+}
+
+/** Returns the offset-syncs topic for given cluster alias. */
+default String offsetSyncTopic(String clusterAlias) {
+return "mm2-offset-syncs." + clusterAlias + ".internal";
+}
+
+/** Returns the name checkpoint topic for given cluster alias. */
+default String checkpointTopic(String clusterAlias) {
+return clusterAlias + ".checkpoint.internal";
Review comment:
This topic is currently named `.checkpoints.internal`, see
https://github.com/apache/kafka/blob/trunk/connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/MirrorClientConfig.java#L60.
We need to keep the current name
##
File path:
connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/ReplicationPolicy.java
##
@@ -52,9 +52,39 @@ default String originalTopic(String topic) {
}
}
+/** Returns heartbeats topic name.*/
+default String heartbeatsTopic() {
+return "heartbeats";
+}
+
+/** Returns the offset-syncs topic for given cluster alias. */
+default String offsetSyncTopic(String clusterAlias) {
Review comment:
Should we called this method `offsetSyncsTopic` so it matches the actual
topic name?
##
File path:
connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/DefaultReplicationPolicy.java
##
@@ -70,4 +70,32 @@ public String upstreamTopic(String topic) {
return topic.substring(source.length() + separator.length());
}
}
+
+private String internalSuffix() {
+return separator + "internal";
+}
+
+private String checkpointsTopicSuffix() {
+return separator + "checkpoints" + internalSuffix();
+}
+
+@Override
+public String offsetSyncTopic(String targetAlias) {
Review comment:
Should we name the argument `clusterAlias` as it's not necessarily the
target cluster alias?
##
File path:
connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/ReplicationPolicy.java
##
@@ -52,9 +52,39 @@ default String originalTopic(String topic) {
}
}
+/** Returns heartbeats topic name.*/
+default String heartbeatsTopic() {
+return "heartbeats";
+}
+
+/** Returns the offset-syncs topic for given cluster alias. */
+default String offsetSyncTopic(String clusterAlias) {
+return "mm2-offset-syncs." + clusterAlias + ".internal";
+}
+
+/** Returns the name checkpoint topic for given cluster alias. */
+default String checkpointTopic(String clusterAlias) {
+return clusterAlias + ".checkpoint.internal";
Review comment:
Can we also rationalize the different names. Ig uess this method should
be `checkpointsTopic`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
