如何高效验证Kafka MirrorMaker2高吞吐量Topic消息全量复制

针对高吞吐量/大容量Topic的MM2消息复制完整性验证，推荐以下几种简便高效的方案：

1. 用MM2内置Checkpoint机制做精准对比 #

启用CheckpointConnector后，MM2会定期在目标集群的mm2-offset-checkpoints.<source-cluster> Topic里写入偏移量检查点，记录每个源Topic分区的已复制进度。

• 直接用Kafka命令行工具过滤出目标Topic的偏移量：

  kafka-console-consumer.sh --bootstrap-server target-dev:9094 --topic mm2-offset-checkpoints.dev --from-beginning --property print.key=true --property key.separator=":" | grep "你的源Topic名称"
  

• 分别查询源集群和目标集群对应Topic的最新偏移量做对比：

  # 查源集群Topic的最新偏移量
  kafka-run-class.sh kafka.tools.GetOffsetShell --bootstrap-server dev:9094 --topic 你的源Topic名称 --time -1
  # 查目标集群镜像Topic的最新偏移量
  kafka-run-class.sh kafka.tools.GetOffsetShell --bootstrap-server target-dev:9094 --topic 你的源Topic名称 --time -1
  

如果两者偏移量一致（或差值在正常同步延迟范围内），就说明消息已经完整复制。

2. 调用Connect REST API批量监控 #

MM2基于Kafka Connect实现，直接调用Connect集群的REST API就能拿到镜像任务的状态和偏移量：

curl -X GET http://<你的Connect集群地址>/connectors/<源连接器名称>/status

响应里会列出每个任务的source_offset和target_offset，逐一对比各分区的偏移量差值，要是所有分区的差值都稳定在很小范围内，同步就是正常的。还可以写个简单脚本批量遍历所有Topic的任务偏移量，自动化完成验证，特别适合高吞吐量场景。

3. 关键消息抽样验证 #

对于超大规模Topic，全量验证不现实，直接用抽样策略：

• 从源集群随机选几个时间点的消息，记录它们的唯一标识（比如业务ID、消息哈希值）。
• 在目标集群的镜像Topic里查询这些标识是否存在：

  kafka-console-consumer.sh --bootstrap-server target-dev:9094 --topic 你的源Topic名称 --from-beginning | grep "消息唯一标识"
  

要是抽样的关键消息全部命中，就能大概率确认复制的完整性。

4. 优化偏移量同步Topic的压缩配置 #

把你配置里注释掉的offset-syncs.topic.max.compaction.lag.ms参数开启，设个较小的值（比如120000），加快偏移量Topic的压缩速度，减少无效数据：

clusters:
- alias: "target-dev"
  bootstrapServers: target-dev:9094
  config:
     offset-syncs.topic.max.compaction.lag.ms: 120000

压缩后，mirrormaker2-cluster-offsets Topic里只会保留每个源Topic分区的最新偏移量记录，不管用Kafka UI还是命令行，都更容易定位目标Topic的偏移量信息。

内容的提问来源于stack exchange，提问作者Nicholas Rinaldi