Kafka跨集群同步：MirrorMaker2原理、配置与高可用实践

1. 为什么今天还在认真聊 MirrorMaker？它真没被完全淘汰

Kafka 集群数据备份——这个词一出来，很多人第一反应是“不就是跨集群同步吗？用 MirrorMaker2 不就完了？”但我在某金融级实时风控系统迁移项目里连续踩了三周坑之后才真正明白：MirrorMaker 不是一个开箱即用的按钮，而是一套需要亲手调教的精密传动机构。它背后牵扯的不是简单的消息复制，而是分区水位对齐、消费者偏移映射、主题配置继承、ACL 权限穿透、甚至时钟漂移引发的重复消费判定失效。你看到的是mirror-maker.sh一行命令，我实际调试时盯着的是__consumer_offsets主题里每个分区的offset和timestamp如何在源集群和目标集群间做非线性映射。

关键词“Kafka 集群数据备份”听着像运维兜底动作，实则直指容灾体系最脆弱的一环：当主集群因网络分区或磁盘故障不可写时，备份集群能否在秒级内接管生产流量，且不丢一条风控指令、不重放一笔清算请求？这就决定了 MirrorMaker 的配置不能只看吞吐量，更要抠每一个时间戳语义、每一段元数据同步粒度、每一次失败重试的退避策略。我见过太多团队把 MirrorMaker 当成“高级 rsync”，结果在真实故障切换时发现：消费者组 offset 差了 12 万条，主题 retention.ms 被强制覆盖为 7 天（而源集群是 30 天），连__transaction_state这个内部事务状态主题都没同步——导致下游 Flink 作业重启后直接卡死在事务恢复阶段。

它适合谁？不是刚学完 Kafka 基础概念的新手，而是已经部署过至少两个以上 Kafka 集群、处理过跨机房延迟敏感型业务（比如实时推荐、支付对账、IoT 设备心跳）、且正在设计异地多活或灰度发布链路的中高级工程师。如果你的场景只是“每天凌晨把测试环境数据 dump 到开发环境”，那真没必要碰 MirrorMaker；但如果你的 SLA 要求 RPO=0、RTO<30 秒，那你现在读的每一行配置说明，都可能决定下一次线上事故的持续时间。

2. MirrorMaker 的本质不是复制工具，而是元数据翻译器

2.1 它解决的从来不是“怎么搬数据”，而是“怎么让数据在新地方活下来”

很多人误以为 MirrorMaker 的核心工作是消费源集群再生产到目标集群，这没错，但只说对了 30%。真正让它区别于普通 consumer-producer 管道的关键，在于它对Kafka 元数据的深度感知与主动翻译能力。举个最典型的例子：源集群有个主题user_clicks_v2，配置了cleanup.policy=compact,delete，retention.ms=2592000000（30 天），segment.bytes=1073741824（1GB）。如果用纯脚本拉取再写入，目标集群会按默认配置创建主题——意味着retention.ms=604800000（7 天）、segment.bytes=1073741824（这个倒可能一致）、cleanup.policy=delete。结果就是：源集群保留 30 天的用户点击明细，在备份集群里 7 天后自动清空，下游做用户行为回溯时直接报错OffsetOutOfRangeException。

MirrorMaker1（MM1）通过--enable-configs参数开启配置同步，但它只同步topic-level configs，且无法处理 ACL、配额、动态 broker 配置等。而 MirrorMaker2（MM2）引入了ConfigPropertyFilter和TopicFilter插件机制，允许你编写 Java 类，在主题创建前拦截并重写任意配置项。比如我们写的RetentionOverrideFilter，逻辑很简单：检测到主题名匹配^user_.*正则时，强制将retention.ms改为2592000000，同时把min.insync.replicas从源集群的2动态提升为3（因为目标集群副本数更多，要利用冗余提升可靠性）。这种“翻译”不是简单拷贝，而是结合目标集群物理拓扑做的主动适配。

再看更隐蔽的元数据：消费者组 offset。MM1 采用ConsumerOffsetChecker+Producer.send()方式粗暴同步，问题在于它把 offset 当作普通消息发到__consumer_offsets，但目标集群的__consumer_offsets是由其自身 GroupCoordinator 管理的，格式和序列化方式与源集群不兼容。这就导致 offset 同步后，消费者启动时读到的是乱码，或者干脆触发UnknownMemberIdException。MM2 彻底重构了 offset 同步路径：它不直接写__consumer_offsets，而是通过 AdminClient 的alterConsumerGroupOffsets()API，以原子方式更新目标集群的 group state。这个 API 要求你提供GroupId、TopicPartition和OffsetAndMetadata，而 MM2 在同步过程中会维护一个内存映射表，记录源集群group-A的topic-X-0offset=123456789，对应目标集群group-A-mirror的topic-X-0offset=123456789（注意：group id 默认加-mirror后缀，这是可配置的）。这个映射过程不是静态的，而是随着每个批次消费完成实时更新，并持久化到 MM2 自己的 checkpoint topic 中（默认mm2-checkpoints.internal），确保进程重启后能从断点续传。

提示：MM2 的 checkpoint topic 必须在目标集群手动创建，且 replication.factor 至少为 3。我曾因忘记这一步，导致 MM2 重启后从头开始同步，3TB 数据重刷了 17 小时——而 checkpoint 本身只有几 KB。

2.2 MirrorMaker2 的架构演进：从单点管道到分布式协同引擎

MM1 是典型的单进程模型：一个 JVM 进程，一个 Consumer 实例，一个 Producer 实例，所有主题、所有分区都在这个进程里轮询处理。它的扩展性瓶颈非常明显——当源集群有 500 个主题、2000 个分区时，单个 MM1 进程的 GC 压力会让吞吐量断崖式下跌，且无法水平扩容。你只能靠增加多个 MM1 实例，每个实例负责一部分主题，但这样又带来新的问题：offset 同步不同步、配置管理分散、故障定位困难。

MM2 彻底转向分布式架构，核心是Connector-Task 模型，完全复用 Kafka Connect 的运行时框架。它把整个同步任务拆解为三个核心 Connector：

• MirrorSourceConnector：负责从源集群拉取数据，每个 Task 对应一个或多个 TopicPartition；
• MirrorCheckpointConnector：专门负责消费者组 offset 的同步，独立于数据流，保证 offset 更新的强一致性；
• MirrorHeartbeatConnector：定期向目标集群发送心跳消息（写入mm2-heartbeats.internal主题），用于监控集群连通性和同步延迟。

这三个 Connector 可以部署在同一 Connect 集群，也可以分拆到不同集群——比如把MirrorCheckpointConnector单独部署在高可用性要求更高的管理网段，避免数据流阻塞影响 offset 同步。每个 Connector 下的 Task 可以动态伸缩：当你新增一个主题同步需求，只需更新 Connector 配置，Connect 框架会自动 rebalance Task 分配。我们在线上环境做过压测：单个 MM2 Connect 集群（3 节点）稳定支撑 1200+ TopicPartition 的同步，平均端到端延迟（source timestamp → target timestamp）控制在 800ms 内，P99 < 2.3s。

这个架构带来的最大收益是故障隔离。以前 MM1 挂了，整个同步链路中断；现在即使MirrorSourceConnector的某个 Task 因网络抖动失败，Connect 框架会自动将其迁移到其他节点重试，而MirrorCheckpointConnector仍在持续更新 offset，确保数据不丢。我们还利用 Connect 的 REST API 开发了自动化巡检脚本：每 5 分钟调用/connectors/{name}/status，检查所有 Task 的state是否为RUNNING，trace字段是否包含org.apache.kafka.connect.errors.RetriableException，一旦发现就自动触发告警并推送钉钉消息——这套机制让我们在去年双十一流量洪峰期间，提前 22 分钟发现了源集群某台 broker 的磁盘 IO 异常，避免了一次潜在的数据同步延迟事故。

3. 配置不是填空题，而是需要计算的工程决策

3.1 吞吐量参数：别迷信文档里的默认值，先算你的 P99 消息大小

MM2 的tasks.max、consumer.overrides.max.poll.records、producer.override.batch.size这些参数，网上教程全在告诉你“设大点吞吐更高”，但没人告诉你：设错一个，可能让整个集群的 GC 时间翻倍。我们线上环境的真实数据：平均消息大小 1.2KB，P99 消息大小 8.7KB（因为有用户上传的 base64 编码截图），峰值 TPS 24000。如果按文档建议把max.poll.records设为 500，那么单次 poll 最大拉取字节数 = 500 × 8.7KB ≈ 4.2MB。而我们的 Consumer 实例堆内存只有 4GB，年轻代 1.5GB，这意味着每秒可能产生 5~6 次 Minor GC——实测下来，GC 吞吐量下降 37%，同步延迟 P99 从 1.2s 涨到 4.8s。

解决方案是反向计算：先确定你可接受的单次 poll 字节数上限。我们设定目标为 ≤ 1.5MB（留出 50% 冗余），再除以 P99 消息大小：1.5MB / 8.7KB ≈ 172。所以max.poll.records=170是更安全的值。同理，batch.size也不能盲目设大。Producer 的 batch 是按字节累积的，batch.size=16384（默认 16KB）在小消息场景很高效，但对我们 1.2KB 平均大小的消息，意味着每批最多装 13 条，而网络往返（RTT）是 3.2ms，实际吞吐受限于linger.ms。我们最终设为batch.size=65536（64KB），配合linger.ms=5，实测每批平均装 52 条，吞吐提升 2.1 倍，且 GC 压力平稳。

注意：max.poll.records和fetch.max.wait.ms是联动参数。如果max.poll.records设太小，而fetch.max.wait.ms又设太大（比如 500ms），会导致 Consumer 长时间空转等待数据，浪费连接资源。我们线上统一设为fetch.max.wait.ms=100，确保在无数据时快速返回，降低连接维持成本。

3.2 Offset 同步精度：毫秒级时间戳对齐的代价与取舍

MM2 默认使用offset-syncs.topic.replication.factor=3，但很多人忽略了一个关键细节：offset 同步的频率，决定了你在故障切换时最多丢失多少数据。MM2 的 offset 同步不是实时的，而是按周期批量提交。默认配置offset.flush.interval.ms=60000（1 分钟），意味着如果此时发生故障，你最多丢失最近 1 分钟的 offset 更新，消费者从备份集群启动时，会从 1 分钟前的位置开始重放——这对风控指令可能是灾难性的。

我们把它调到了offset.flush.interval.ms=5000（5 秒），但立刻遇到性能问题：每 5 秒就要向mm2-checkpoints.internal主题写入数千条 offset 记录，导致该 topic 的 segment 切换过于频繁，broker 磁盘 IO 持续 92%。最终方案是分级刷新：对核心主题（如risk_decision_v3）单独配置offset.flush.interval.ms=1000，对普通日志主题（如app_log_v1）保持60000，通过topics.exclude和topics.include配合正则实现。这个操作需要修改 MM2 的MirrorSourceConnector配置，添加offset.flush.interval.ms作为 connector-level override，而不是全局配置。

更深层的问题是时间戳语义。Kafka 消息有CreateTime和LogAppendTime两种时间戳类型。源集群用LogAppendTime（broker 接收时打的时间戳），目标集群如果也用LogAppendTime，那么两条消息在源集群间隔 10ms，在目标集群可能变成 500ms（因为网络延迟+broker 处理排队）。这会导致基于时间窗口的 Flink 作业计算错误。我们的解法是在 MM2 的transforms链中插入自定义 SMT（Single Message Transform）：TimestampPreserver，它读取消息 header 中的x-source-timestamp（我们在生产端写入），并强制设置为CreateTime，确保时间戳在传输中零衰减。这个 transform 的代码只有 47 行 Java，但让下游所有时间敏感型作业的准确率从 99.2% 提升到 100%。

3.3 安全与权限：ACL 同步不是可选项，而是容灾生效的前提

很多团队在测试环境验证 MirrorMaker 时一切顺利，一上生产就失败，90% 的原因是 ACL（Access Control List）没同步。MM2 默认不同步任何 ACL 规则，这意味着：即使你成功把payment_events主题同步到了目标集群，但目标集群的payment-consumer-group用户没有READ权限，消费者启动时直接报TopicAuthorizationException。

MM2 提供了SyncGroupAcls和SyncTopicAcls两个内置 transform，但它们只同步与同步任务相关的 ACL（比如MirrorSourceConnector用的 service account 的权限），不会同步原始生产者/消费者的权限。真正的解法是：在 MM2 启动前，用 Kafka AdminClient 扫描源集群所有 ACL，生成目标集群的等效规则，再通过kafka-acls.sh --add批量导入。我们写了 Python 脚本acl_sync.py，核心逻辑是做权限映射：

• 源集群 ACL：User:CN=payment-prod,O=Finance,C=CN→TOPIC→payment_events→READ
• 目标集群 ACL：User:CN=payment-prod-mirror,O=Finance,C=CN→TOPIC→payment_events→READ

注意 CN（Common Name）后缀加-mirror，这是为了隔离权限，避免生产账号误操作备份集群。脚本还会自动跳过User:*这类宽泛规则，防止权限爆炸。执行频率设为每天凌晨 2 点，配合 Ansible 自动化部署，确保权限始终与源集群保持最终一致性。

提示：SyncTopicAclstransform 有个隐藏坑——它只同步ALLOW规则，忽略DENY。而 Kafka ACL 是 deny-by-default，所以漏掉DENY规则可能导致越权访问。我们禁用了这个 transform，全部走脚本同步，虽然麻烦点，但安全可控。

4. 实操全流程：从零搭建高可用 MM2 集群的 7 个关键步骤

4.1 环境准备：硬件、网络、JVM 的硬性门槛

别急着下载二进制包，先确认你的基础设施是否达标。我们踩过的最大坑是：在一台 8C16G 的虚拟机上部署 MM2，结果发现 CPU 被kafka-coordinator-heartbeat-thread占满，同步延迟飙升。根本原因是 MM2 的MirrorCheckpointConnector需要高频心跳，而虚拟机的 CPU steal time 过高（>15%），导致心跳超时，触发频繁 rebalance。

硬件建议：

• CPU：至少 16 核（物理核），避免超线程干扰，/proc/cpuinfo中cpu MHz≥ 2600
• 内存：≥ 32GB，JVM 堆内存设为 16GB（-Xms16g -Xmx16g），年轻代 6GB（-XX:NewRatio=2）
• 磁盘：SSD，IOPS ≥ 10000，/var/lib/kafka-mirror单独挂载，预留 50% 空间用于 checkpoint 日志滚动

网络要求：

• 源集群与目标集群间 RTT ≤ 15ms（跨机房专线），丢包率 < 0.01%
• 必须开通双向 9092（Kafka client port）和 9093（Kafka inter-broker port）端口
• DNS 必须能双向解析，禁止用 IP 地址硬编码（否则 SSL 证书校验失败）

JVM 调优（connect-distributed.properties中追加）：

# 关键 GC 参数，避免 CMS 被废弃后的 G1 停顿 jvm.args=-Xms16g -Xmx16g -XX:NewRatio=2 -XX:MaxGCPauseMillis=200 -XX:+UseG1GC -XX:G1HeapRegionSize=4M -XX:G1ReservePercent=15 # 禁用显式 GC，防止 MM2 内部调用 System.gc() 导致 STW jvm.args=-XX:+DisableExplicitGC # 开启 GC 日志，便于问题定位 jvm.args=-Xloggc:/var/log/kafka-mirror/gc.log -XX:+PrintGCDetails -XX:+PrintGCDateStamps

实测对比：未调优时 Full GC 频率 12 次/小时，调优后降至 0.3 次/小时，同步延迟 P95 从 3.8s 降至 0.9s。

4.2 配置文件生成：用模板引擎避免手工编辑失误

MM2 有 3 个核心配置文件：connect-distributed.properties（Connect 框架）、mm2.properties（MM2 全局）、connector-configs.json（Connector 实例）。手工编辑极易出错，比如漏掉.或写错大小写。我们用 Jinja2 模板自动生成：

mm2.j2模板片段：

# {{ cluster_name }} 集群配置 {{ cluster_name }}.bootstrap.servers={{ bootstrap_servers }} {{ cluster_name }}.security.protocol={{ security_protocol }} {% if security_protocol == "SSL" %} {{ cluster_name }}.ssl.truststore.location=/etc/kafka/secrets/{{ cluster_name }}-truststore.jks {{ cluster_name }}.ssl.truststore.password={{ truststore_password }} {{ cluster_name }}.ssl.keystore.location=/etc/kafka/secrets/{{ cluster_name }}-keystore.jks {{ cluster_name }}.ssl.keystore.password={{ keystore_password }} {% endif %}

执行命令：

jinja2 mm2.j2 -D cluster_name=source -D bootstrap_servers="kafka-src-01:9092,kafka-src-02:9092" -D security_protocol=SSL > mm2-source.properties

这样生成的配置文件，100% 符合 Kafka 配置语法规范，且支持变量注入（如密码从 HashiCorp Vault 获取），杜绝了ssl.truststore.location路径错误这类低级失误。

4.3 Connector 创建：REST API 调用的幂等性设计

不要用curl -X POST手动创建 Connector，必须封装成幂等脚本。因为 MM2 的 REST API 在创建已存在 Connector 时会返回 409 Conflict，而非静默覆盖。我们用 Python 写了create_connector.py：

def create_connector(connector_name, config): # 先 GET 检查是否存在 resp = requests.get(f"http://mm2-connect:8083/connectors/{connector_name}") if resp.status_code == 200: print(f"Connector {connector_name} exists, updating config...") # PUT 更新配置 requests.put(f"http://mm2-connect:8083/connectors/{connector_name}/config", json=config) else: # POST 创建新 Connector requests.post("http://mm2-connect:8083/connectors", json={ "name": connector_name, "config": config })

关键点：PUT /connectors/{name}/config是幂等的，多次调用效果相同。我们把所有 Connector 配置存放在 Git 仓库，每次上线前运行此脚本，确保环境配置与代码库完全一致。这个实践让我们在 3 个月内实现了 12 次配置变更零失误。

4.4 同步验证：不止看“running”，要看 offset 差值和延迟直方图

MM2 的/connectors/{name}/status返回"state": "RUNNING"只代表进程活着，不代表数据在同步。必须建立三层验证：

第一层：基础连通性

# 检查 source connector 是否在消费 kafka-consumer-groups.sh --bootstrap-server kafka-src:9092 \ --group connect-mirror-source-source \ --describe | grep "CURRENT-OFFSET" # 检查 target connector 是否在生产 kafka-consumer-groups.sh --bootstrap-server kafka-tgt:9092 \ --group connect-mirror-source-target \ --describe | grep "CURRENT-OFFSET"

第二层：offset 差值监控我们用 Prometheus + Grafana，采集 JMX 指标kafka.connect:type=connector-metrics,connector=.*下的offset.lag。阈值设为：核心主题 lag < 1000，普通主题 lag < 5000。超过阈值自动触发告警。

第三层：端到端延迟分析在源集群生产消息时，写入 headerx-source-timestamp=System.currentTimeMillis()；在目标集群消费时，计算System.currentTimeMillis() - header.get("x-source-timestamp")。我们用 Flink SQL 实时统计：

SELECT TUMBLING_WINDOW(processing_time, INTERVAL '1' MINUTE) as window, COUNT(*) as count, MAX(delay_ms) as max_delay, PERCENTILE_CONT(0.99) WITHIN GROUP (ORDER BY delay_ms) as p99_delay FROM ( SELECT CAST(event_time AS BIGINT) - CAST(headers['x-source-timestamp'] AS BIGINT) as delay_ms FROM kafka_source_table ) GROUP BY TUMBLING_WINDOW(processing_time, INTERVAL '1' MINUTE)

这张图比任何文字描述都直观：如果 P99 延迟曲线突然抬升，说明网络或 broker 出现瓶颈；如果 max_delay 持续 > 5s，说明linger.ms或batch.size配置不合理。

4.5 故障模拟与切换演练：每月一次的“压力测试”

我们坚持每月最后一个周五下午 3 点，进行真实故障演练。流程严格按 SOP 执行：

1. 准备阶段：通知所有相关方，暂停非紧急发布，备份当前 MM2 配置
2. 注入故障：用iptables在 MM2 节点上屏蔽源集群 9092 端口 5 分钟
3. 观察阶段：记录offset.lag增长速率、checkpoint topic 的写入延迟、GC 日志中的 Full GC 次数
4. 恢复阶段：开放端口，观察 MM2 是否自动恢复，lag 是否在 2 分钟内归零
5. 验证阶段：用kafka-console-consumer.sh对比源/目标集群最新 10 条消息的timestamp和value，确保完全一致

去年 8 月的演练中，我们发现MirrorCheckpointConnector在网络恢复后，花了 4 分 32 秒才重新同步 offset，原因是offset.flush.interval.ms设得太小（1000ms），导致大量小批次写入，触发了目标集群的request.timeout.ms超时。于是我们调整为offset.flush.interval.ms=3000，并在producer.override.request.timeout.ms=30000，问题解决。

注意：演练必须在业务低峰期进行，且首次演练前，务必在测试环境完整跑通一遍。我们曾因跳过测试环节，导致演练时误删了生产集群的mm2-checkpoints.internaltopic，花了 47 分钟重建。

5. 常见问题与独家排查技巧实录

5.1 “同步延迟越来越高，但 CPU 和磁盘都很空闲”——真相是 GC 在偷偷吃掉时间

现象：offset.lag持续增长，Prometheus 显示 JVMgc.pause.time每分钟 200ms，但process_cpu_seconds_total只有 0.3。看起来 CPU 很空，实则 GC 线程在疯狂工作。

根因：G1 GC 的Mixed GC阶段会并发标记，但Evacuation阶段是 Stop-The-World 的。当老年代碎片化严重时，Evacuation时间会指数级增长。我们用jstat -gc <pid>发现G1OldGen使用率 92%，G1MixedGCTime平均 1200ms。

解法：不是加大堆内存，而是优化对象生命周期。MM2 的MirrorCheckpointConnector会缓存大量OffsetAndMetadata对象，我们通过-XX:G1HeapRegionSize=4M强制增大 region size，减少 region 数量，从而降低标记开销；同时-XX:G1ReservePercent=15预留 15% 堆空间给 GC 使用，避免 evacuation 失败触发 Full GC。

5.2 “主题同步了，但消费者组 offset 没变”——检查你的 heartbeat topic 是否可写

现象：MirrorSourceConnector状态正常，MirrorCheckpointConnector状态也是 RUNNING，但目标集群的消费者组 offset 始终为 0。

排查路径：

1. 查看MirrorCheckpointConnector的日志，搜索Failed to write offsets to checkpoint topic
2. 检查mm2-checkpoints.internal主题的 ACL：kafka-acls.sh --authorizer-properties zookeeper.connect=zookeeper:2181 --list --topic mm2-checkpoints.internal
3. 确认 MM2 使用的 service account 是否有WRITE权限

我们遇到的真实案例：ZooKeeper ACL 里mm2-checkpoints.internal的WRITE权限只给了User:CN=mm2-service,O=IT，但 MM2 进程实际用的是User:CN=mm2-prod,O=IT（证书 CN 不同）。修复后，offset 同步立即恢复正常。

5.3 “同步了几天，突然所有 lag 归零，然后又暴涨”——checkpoint topic 被自动删除

现象：某天凌晨 2 点，所有offset.lag突然归零，随后 10 分钟内涨到百万级。

根因：mm2-checkpoints.internal主题的retention.ms被设为默认值 604800000（7 天），而我们忘了手动创建它，Kafka 自动创建时用了 broker 默认配置。当 checkpoint 数据超过 7 天，旧 segment 被删除，MM2 重启后只能从头同步。

解法：在 MM2 首次启动前，手动创建 checkpoint topic：

kafka-topics.sh --bootstrap-server kafka-tgt:9092 \ --create \ --topic mm2-checkpoints.internal \ --partitions 12 \ --replication-factor 3 \ --config cleanup.policy=compact \ --config retention.ms=2592000000 \ --config segment.bytes=1073741824

注意cleanup.policy=compact，这是必须的，否则 offset 会被定期清理。

5.4 “为什么我的 transform 不生效？”——SMT 加载顺序的隐式规则

现象：在mm2.properties中配置了transforms=TimestampPreserver，但日志里看不到 transform 的 debug 输出。

真相：MM2 的 transforms 是按字符串字典序加载的，而TimestampPreserver会依赖HeaderInjector（注入x-source-timestamp）。如果HeaderInjector名字是AddHeader，而TimestampPreserver名字是PreserveTs，那么AddHeader会先执行，PreserveTs后执行——但如果名字是TimestampPreserver和HeaderInjector，字典序HeaderInjector<TimestampPreserver，顺序就对了。

解法：在transforms配置中显式指定顺序：

transforms=inject,timestamp transforms.inject.type=com.example.HeaderInjector transforms.timestamp.type=com.example.TimestampPreserver

这样无论类名如何，执行顺序都由配置项名决定。

5.5 “跨集群同步后，消息顺序乱了”——分区键（key）哈希不一致的陷阱

现象：源集群user_events主题按user_id分区，目标集群同样主题，但相同user_id的消息被分到不同分区，导致下游按 user_id 聚合时结果错误。

根因：Kafka 的分区算法是hash(key) % num_partitions，但不同版本 Kafka 的 hash 算法略有差异（比如 2.8 用 Murmur2，3.0 用 XXHash）。如果源集群是 Kafka 2.8，目标集群是 3.0，同样的 key 会得到不同的 hash 值。

解法：在 MM2 的transforms中加入KeyConverter，强制使用统一 hash 算法：

transforms=keyhash transforms.keyhash.type=org.apache.kafka.connect.transforms.HashKey transforms.keyhash.field=user_id transforms.keyhash.algorithm=MURMUR2

这样无论目标集群 Kafka 版本如何，key 的 hash 结果都与源集群一致，分区顺序得以保持。

6. 运维手册：日常巡检清单与应急响应 SOP

6.1 每日巡检清单（5 分钟完成）

6.2 应急响应 SOP：当同步完全中断时

Step 1：快速定界（≤2 分钟）

• 执行curl http://mm2-connect:8083/connectors，确认所有 Connector 是否存活
• 如果全部UNASSIGNED，立即检查 Connect 集群节点数是否 < 3（quorum 丢失）
• 如果单个 ConnectorFAILED，查看其trace字段，提取关键词（如TimeoutException,SSLHandshakeException）

Step 2：临时恢复（≤5 分钟）

• 对FAILED的 Connector，执行curl -X POST "http://mm2-connect:8083/connectors/{name}/restart"
• 如果 restart 失败，手动删除 Connector：curl -X DELETE "http://mm2-connect:8083/connectors/{name}"，再用幂等脚本重建

Step 3：根因分析（事后）

• 检查/var/log/kafka-mirror/connect.log，搜索ERROR和WARN
• 重点看org.apache.kafka.connect.runtime.WorkerSinkTask和org.apache.kafka.connect.mirror.MirrorCheckpointTask的堆栈
• 如果是TimeoutException，检查网络连通性：telnet kafka-src:9092，mtr kafka-src
• 如果是SSLHandshakeException，检查证书有效期：openssl x509 -in /etc/kafka/secrets/mm2-keystore.jks -noout -dates

Step 4：知识沉淀

• 将本次故障的trace、log snippet、root cause、fix记录到内部 Wiki 的 “MM2 故障知识库”
• 更新巡检清单，增加针对该问题的检查项（例如：新增证书有效期检查）

这个 SOP 让我们团队的平均 MTTR（平均修复