如何从指定offset或时间点启动Kafka MirrorMaker2

你可以根据实际需求选择以下两种方案实现精准位点启动：

方案1：基于Spark Structured Streaming checkpoint记录的位点启动 #

• 提取checkpoint中存储的源集群（ClusterA）消费位点：Spark的Kafka消费位点默认存放在checkpoint目录的offsets/子路径下，读取最新的批次快照文件，整理出所有待同步主题对应分区的源端offset值，注意这里必须取ClusterA原生的分区offset，不要混淆之前MM2同步到ClusterB的重命名主题、转换分区的offset。
• 配置MM2 source connector的起始位点：搭建ClusterA->ClusterC的MM2链路时，不要使用默认的自动重置位点策略，直接通过topic.offsets配置项为每个分区指定起始消费位点，配置格式为<ClusterA在MM2中的别名>.<主题名>:<分区号>:<offset值>，多个分区用逗号分隔。
• 启动前校验：先单实例启动MM2 source connector，查看启动日志中各分区的实际启动消费位点，和你从checkpoint提取的值完全匹配后，再全量启动MM2集群节点，避免漏数或重复消费。

方案2：基于ClusterB宕机时间点启动 #

• 确认ClusterB的精确宕机时间：取ClusterB完全停止对外服务的毫秒级Unix时间戳，不要取业务感知故障的延后时间，避免出现数据断档。
• 查询对应时间点ClusterA的分区offset：使用Kafka自带工具查询各待同步主题分区在该时间戳对应的最早消息offset，执行命令参考：

  kafka-get-offsets.sh --bootstrap-server <ClusterA的bootstrap地址> --topic <待同步主题名> --time <宕机毫秒时间戳>
  

  整理所有分区返回的offset结果。
• 同方案1的配置逻辑，将查询到的offset值填入MM2 source connector的topic.offsets配置项，校验位点无误后启动MM2链路即可。

关键注意点

• 你的Spark任务没有将消费位点提交到Kafka集群，因此MM2自带的消费者组位点同步功能（sync.group.offsets.enabled相关配置）在该场景下不生效，无需开启，避免出现位点错乱。
• 链路启动后建议先抽样校验10~20分钟的同步数据，确认ClusterC中的消息起始位置符合预期后，再将Spark任务的消费集群切换为ClusterC。

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