apache-kafka - 从 Confluent Replicator 迁移到 Apache Mirror Maker 2.0 时保持相同的偏移量

Question

我们目前正在尝试将 Confluent 复制器迁移到 Apache Open Source Mirror Maker v2.0。我们面临一个问题，当镜像制造商在同一主题上启动时，已经被复制器复制的消息再次被复制。这不应该发生，因为消息在目标集群中被复制。以下是更多详细信息：

1. RCA：replicator 为复制消息分配一个消费者组。这个消费者组维护了源主题的偏移量。但是我们无法将相同的消费者组分配给镜像制造商 2 中的消费者配置。
2. Mirror Maker 1.0：可以在 consumer.properties 文件中分配作为同一个消费者组工作的消息，并且在复制器停止后立即选择消息。
3. 尝试source.cluster.consumer.group.id在 mirror maker 2.0 中以所有可用选项（在集群模式、连接独立和连接分布式模式下）运行和配置，但 mirror maker 2.0 在复制消息时将使用者组 ID 分配为 null。

如果有人做过同样的事情并试图与 mirror maker 2.0 保持相同的偏移量，那么任何指针。

Answer 1

我们有一个粗略的方法来解决这个问题。以下是高级步骤：

• 从 Replicator 的内部主题中读取消息以存储偏移量。[连接偏移]
• 该主题存储所有主题的偏移量，这些主题在 key:value 对中被复制。例如

键：["replicator-group",{"topic":"TEST","partition":0}]
值：{"offset":24}

• 对于每个主题和分区，每当复制一条新消息时，都会产生一条具有相同键但增加了偏移量的新消息到连接偏移量主题。
• 将此消息的密钥转换为Mirror Maker 2 格式，并在MirrorMaker2 的内部主题中生成。【可以在mirrormaker2-connect-distributed.properties文件中更改内部主题】镜像制作者内部主题的格式为：

键：["mirrormaker-group",{"cluster":"","partition":0,"topic":"TEST"}]
值：{"offset":24}

• 发布消息后，一旦镜像制造商重新启动，它将读取内部主题以获取该消息必须复制的主题的最新偏移量，这样我们可以确保没有重复的消息被复制。