MirrorMaker2 - 使用独立连接器自定义主题重命名答案

【问题标题】：MirrorMaker2 - Custom rename of topics with standalone connectorMirrorMaker2 - 使用独立连接器自定义主题重命名
【发布时间】：2020-05-15 06:54:59
【问题描述】：

我想将 MirrorMaker 作为独立连接器运行。到目前为止，我还没有找到任何有关配置的文档。据我想象，以下内容将复制myTopic。

现在在目标集群中，我需要主题有另一个名称foo（不是自动重命名）。这是由MirrorSourceConnector 直接支持还是我需要其他方法来实现这一点？

connector.class = org.apache.kafka.connect.mirror.MirrorSourceConnector
tasksMax = 2
topics = myTopic
source.cluster.bootstrap.servers = sourceHost:9092
target.cluster.bootstrap.servers = sinkHost:9092

【问题讨论】：

不确定。 MirrorMaker1 可以用作连接器吗？
截至目前，MirrorMaker2 上的文档非常糟糕。来自 Kafka Sumit 2020 的这段视频表明这是可能的（大约 14:35）：confluent.io/resources/kafka-summit-2020/… 但我不知道该怎么做......

标签： apache-kafka apache-kafka-mirrormaker

【解决方案1】：

所以Kafka Mirror Maker 源代码有一个不错的readme.md。

您的配置方式会有所不同，具体取决于您是直接运行 MM2 还是在 Kafka Connect 中运行。你说的直接，在链接的readme.md。

基本上：

默认情况下，复制的主题基于“源集群”重命名别名”：

topic-1 --> source.topic-1

这可以通过覆盖 replication.policy.separator 来定制属性（默认为句点）。如果您需要更多地控制如何定义了远程主题，您可以实现自定义 ReplicationPolicy 并覆盖 replication.policy.class（默认为 DefaultReplicationPolicy)。

不幸的是，这意味着您不能仅通过配置代码重命名主题。（DefaultReplicationPolicy 类只允许您指定分隔符，不能指定其他内容）。这可能是因为当您指定要镜像的主题时，您使用的是正则表达式，而不是单个主题名称（即使您的源集群主题配置属性只是主题的名称 - 它仍然被视为正则表达式）。

所以，回到文档：ReplicationPolicy 是 Kafka 连接源代码中的 Java 接口，因此您需要实现一个自定义 Java 类来实现 ReplicationPolicy，然后确保它在运行时位于类路径中MM2。

假设您确实编写了这样一个类，并将其命名为com.moffatt.kafka.connect.mirror.FooReplicationPolicy。 Kafka Connect 附带的默认（显然是唯一的）复制策略类是您班级的一个好模板：DefaultReplicationPolicy。您可以看到构建自己的并不会太困难。您可以轻松添加一个 Map - 无论是硬编码的还是配置的 - 查找特定配置的主题名称并将其映射到目标主题名称。

您可以通过在配置中将新类指定为： replication.policy.class = com.moffatt.kafka.connect.mirror.FooReplicationPolicy

【讨论】：