企业官网建设流程全解析

先说个具体场景。上季度我配合一个实时数据团队做大集群搬迁，旧集群在自建机房，新集群跑在容器环境里。数据量不算夸张，但全是高吞吐日志和事件流，加起来两百多个 Topic，业务要求双跑切换、不允许停机。当时团队里有人提议直接用 Consumer + Producer 写搬迁脚本，也有人建议把分区副本跨机房搬过去。最后我们都没走这两条路，而是把整套迁移压在了 MirrorMaker2（下文简称 MM2）上。

这篇不是官方文档的翻译，也不是照着 QuickStart 复读。是我完整跑完一次 Kafka 集群数据迁移后，回头把选型思考、内部机制、配置细节、切换流程和踩过的坑整理成的一份实践笔记。如果你正在犹豫要不要用 MM2，或者已经被内部 Topic、位点同步机制折腾到头晕，这篇应该能帮你省掉不少试错时间。

1. 为什么最终选了 MM2 而不是自研数据搬运脚本

1.1 三种方案摆在一起，各自边界在哪

先说结论：自研脚本不是不能用，而是把问题想简单了。群里同事最早提的方案很朴素：写一个 Java 程序，从旧集群每个分区拉全量数据，再一条条发到新集群同名 Topic 里。听起来逻辑确实不难，但落地会发现几个问题：

• 增量怎么追：程序第一次启动后，旧集群还在持续写入。全量搬完不等于数据一致，你得再开一个增量消费循环，而且这个增量循环要从“全量快照结束的位置”接着读，这里就有个经典 bug：快照时读到 offset 100，全量搬完后旧顶 offset 已经到 200，增量从 100 继续还算对，但如果程序崩溃重启，offset 提交和实际发送很容易错位，漏数据往往就是这里发生的。
• 位点怎么映射：旧集群的消费者组在新集群没有对应的 committed offset。你得自己把 groupId 和 topic partition 的位点关系导出来，再手工重建一份，否则下游切过来要么从头读，要么从最新读，总有一边不满意。
• 失败恢复怎么做：搬了一半进程挂了，怎么知道哪些分区搬完了哪些没搬完？要不要记录“已完成的最大 offset”？这些逻辑不难，但写起来全是细节，而且测试量不小。

原生分区副本搬运其实也救不了。Kafka 的分区重分配工具解决的是同一个集群内 broker 间数据搬迁，不是跨集群迁移。你可以把旧集群的 partition 理解为存放在 A 机房的物理副本，重分配只会改变副本所在 broker，不会把数据跨机房搬到新集群的 broker 上。当然，你可以把新集群的 broker 伪装成旧集群的一部分，再挪副本，再摘掉，这是集群物理迁移玩法，实际操作风险更大，而且新旧集群版本、网络隔离稍有差异就玩不动。

1.2 MM2 的真正优势是“增量同步语义”

MM2 是建立在 Kafka Connect 框架上的连接器组合：MirrorSourceConnector 负责把源集群 Topic 复制到目标集群，MirrorCheckpointConnector 负责跟踪消费者组位点，MirrorHeartbeatConnector 负责探测两端连通性。它不是一个一次性脚本，而是常驻任务，每秒都在协调两端数据。

这意味着它天然解决了增量追踪问题：复制任务本身就一直跑着，旧集群每来一条消息，它就往新集群发一条，源端 lag 会不断收敛。迁移场景里我们只需要等它追平，然后切流，不需要自己写“先全量再增量”的复杂状态机。

其次，它把位点同步做成了标准机制。通过内部 Topic 周期性记录每个消费者组在源端的位置，虽然这套机制不能直接让普通 Consumer 无缝续读（后面会细说），但至少提供了拿到映射关系的通道，比手工维护强多了。

第三个优势是回滚能力。同一个 MM2 配置，把方向反过来就能让新集群数据回流旧集群。真出问题时，消费者切回去还有数据可读，这在生产迁移里是保命技能。

表格列一下当时比选结论：

2. 动手前必须搞懂的四个核心机制：重命名、内部 Topic、位点同步与过滤

2.1 默认的重命名策略会坑掉下游消费者

MM2 默认在目标集群复制的 Topic 名不是原名，而是会加一个前缀：<源集群别名>.<原Topic名>。比如源集群别名配成dc-old，原 Topic 叫order-events，复制到新集群后名字就是dc-old.order-events。

这个设计的本意是支持多集群汇聚：北京集群的 topic 和上海集群的 topic 都同步到同一个中心集群时，用别名区分来源。但在我们这种“整集群搬迁、希望下游无感知改订阅名”的场景里，它就是纯负担。

解决办法是配置replication.policy.class为IdentityReplicationPolicy，让 MM2 保持原 Topic 名复制。这个类从 Kafka 2.6 开始提供，如果你的版本更老，就得自己写一个ReplicationPolicy实现类。代价是如果集群间做双向同步，A 同步到 B、B 又同步回 A，同名 Topic 会形成回环，所以长期容灾架构很多人反而故意保留前缀。一次性搬迁场景用 Identity 没毛病，但注意别同时开反向同步。

2.2 两类内部 Topic：MM2 自己的，和 Connect Worker 的

MM2 运行时会自动创建几个内部 Topic，第一次看到的人很容易懵。建议动手前先分清它们各自的作用：

前三个是 MM2 连接器层创建的，后三个是 Connect Worker 自身运行需要的基础设施。如果集群开的是auto.create.topics.enable=false，启动前必须手工把这几张 Topic 建出来，否则会一直报 Topic 不存在，很多人排查半天才发现是这问题。另外这些 Topic 的数据量不大，但重要，副本数和清理策略建议提前规划好。

2.3 offset 同步不是“切过去就能接着消费”

sync.group.offsets.enabled=true这个参数容易给人错觉。它开启后，MM2 会把源集群每个消费者组的位点周期性地写进mm2-checkpoints.<目标集群>.internal。但问题在于，目标集群上的普通 KafkaConsumer 并不会主动去读这张 Topic。你切到新集群用同一个 groupId 订阅同一个 Topic 时，照样从auto.offset.reset决定的位置开始读。

要让业务消费者真正从检查点续读，得用 MM2 配套的MirrorClient读取检查点并手动调用seek()，或者干脆用官方提供的一种 Consumer 封装。实话说，让每个下游业务都改消费代码，迁移成本会明显上升。我们最后的方案是：先追平数据，切流时短暂停写，消费者用latest启动，业务可接受几秒间隙。这个决策后面细说。

2.4 Topic 过滤要早确认，正则别一上来就写 .*

默认的DefaultTopicFilter会排除 Kafka 内部 Topic（以__开头）以及 MM2 自身的心跳、检查点 Topic。在此基础上，你可以用topic.whitelist指定要同步哪些。最省事的写法是.*，但那是双刃剑：

• 目标集群如果已经有一批业务 Topic，正则会把你不想同步的 Topic 也卷进来。
• 源集群如果有一个超大日志 Topic，七天的数据量几百 TB，而迁移窗口根本不想碰它，正则就该把这类 Topic 排除掉。

我们当时手工维护了一份迁移清单，然后按 Topic 前缀和几个明显不需要同步的 Topic，把topic.whitelist写成了有明确边界的方向。等启动之后才想起来改配置，通常会导致连接器全量重建，非常折腾。所以过滤规则宁可花十分钟提前列清楚。

3. mm2.properties 完整配置拆解与启动前检查清单

3.1 环境与版本选型

Kafka 版本尽量选 2.7 以上的，MM2 在 2.4 引入，2.6 加入了IdentityReplicationPolicy，2.7 算是修了不少连接器缺陷。当时源集群是 2.5，新集群是 2.8，MM2 跑在独立的 2.8 生态里，兼容性没问题。如果你源集群特别老，比如 Kafka 1.x，MM2 未必能直接读，最好先做连通性验证。

运行环境方面，启动 MM2 的机器要能同时访问旧集群和新集群的 bootstrap server，并且带宽要留足。如果两个集群跨机房，专线带宽可能成为瓶颈，实测中同步大 Topic 时一度吃满整个链路。JDK 版本别太老，推荐和 Kafka 运行版本匹配的 JDK 8 或 11。

启动方式也有讲究：别再用老的kafka-mirror-maker.sh，那是 MM1 时代的脚本。要用connect-mirror-maker.sh，它底层会把一套 Connect Worker 和 MM2 连接器一起拉起来。也可以把连接器配置提交到已有的 Connect 集群，但生产上推荐独立跑一套专用 Worker，别跟其他 Connector 混在一起，避免互相挤占线程和内存。

3.2 一套亲测可用的核心配置

下面这份配置基本涵盖了迁移需要的关键项，直接复制改成自己的集群信息就能起步：

# 两套集群的别名与连接信息 clusters = dc-old, dc-new dc-old.bootstrap.servers = old-broker-1:9092,old-broker-2:9092,old-broker-3:9092 dc-new.bootstrap.servers = new-broker-1:9092,new-broker-2:9092,new-broker-3:9092 # 只开启从旧集群到新集群的单向同步 dc-old->dc-new.enabled = true # 并发与过滤 dc-old->dc-new.tasks.max = 8 dc-old->dc-new.topic.filter.class = org.apache.kafka.connect.mirror.DefaultTopicFilter dc-old->dc-new.topic.whitelist = order-events,user-action.*,metrics.* dc-old->dc-new.replication.policy.class = org.apache.kafka.connect.mirror.IdentityReplicationPolicy # 位点同步与心跳 dc-old->dc-new.sync.group.offsets.enabled = true dc-old->dc-new.emit.checkpoints.interval.seconds = 60 dc-old->dc-new.emit.heartbeats.interval.seconds = 5 dc-old->dc-new.checkpoints.topic.replication.factor = 3 dc-old->dc-new.heartbeats.topic.replication.factor = 3 dc-old->dc-new.offset-syncs.topic.replication.factor = 3 # 生产者端调优 dc-old->dc-new.producer.override.linger.ms = 20 dc-old->dc-new.producer.override.batch.size = 1048576 dc-old->dc-new.producer.override.acks = all dc-old->dc-new.producer.override.compression.type = lz4

逐项说下我踩过之后理解的关键：

• tasks.max控制这个连接器的最大 Task 数。Topic 分区多的时候，提高它能让多个 Task 并行拉取和写入，但别盲开特别大，Task 数和分区数、目标集群负载要匹配。
• replication.policy.class上面说了，想要目标端保持原 Topic 名就用IdentityReplicationPolicy。
• sync.group.offsets.enabled开不开看你的切流策略。开了不亏，能拿到检查点数据，但别指望它自动续传。
• producer.override.*是覆盖 MM2 内部生产者配置的入口。大流量迁移里，调大linger.ms和batch.size能显著降低请求次数，减轻 broker 压力。如果网络带宽紧张，compression.type也可以设成zstd，CPU 换带宽，自行取舍。

3.3 Connect Worker 自己的三张 Topic

如果集群禁止自动创建 Topic，需要手动预创建。当时我用kafka-topics.sh把connect-offsets、connect-config、connect-status都建好，每个 3 副本，单分区足够，因为 Connect Worker 的 offset 量不会很大。但如果用分布式 Connect 模式，connect-config和connect-status建议也保持多副本。

启动命令网上到处都有，但容易漏一个细节：MM2 默认的 Topic 复制是持续性的，不是“复制完就退出”。只要连接器不关，它就会一直监听源端的新消息。迁移完成后如果不再需要双向同步，记得停掉连接器，否则目标集群会一直产生无谓流量。

3.4 启动前的自我检查清单

列一个我每次跑之前都会核对的清单，照着过一遍能少踩大量基础坑：

• 新旧集群都能从启动机 telnet 通，bootstrap server 配的是全部 broker 而不是负载均衡入口。
• 目标集群没有同名 Topic 和源端分区数不一致，如果有，先扩分区或删掉重建。
• 涉及多套 ACL/SASL 认证时，确认 MM2 配置里已经带上对应集群的security.protocol、sasl.mechanism、sasl.jaas.config。集群 A 和集群 B 的认证信息可能不同，MM2 是分别给两个集群配置的。
• 源端需要同步的 Topic 的retention.ms是否足够覆盖全量追赶窗口，不够先把该 Topic 保留时间临时调大。
• 消费端切换策略已经和业务方确认，是接受短暂停写，还是必须用 checkpoint 精确续读。
• 目标集群的auto.create.topics.enable是否开启，不开启则手工建 Topic。

4. 迁移时间线实录：全量追赶、数据校验与切流回滚

4.1 第一阶段：启动 MM2，等待全量追赶

启动后做的第一件事不是看多少条数据过去了，而是看三组指标：

• MM2 在源集群使用的消费组 lag：如果 Topic 多，可以按连接器名过滤看到每个 Topic 的同步进度。
• 目标集群同名 Topic 的 LEO（Log End Offset）：判断写入是否在推进。
• Connect Worker 的 Task 状态：有没有 Task 反复失败重启。

全量追赶阶段最容易出现的现象是：小 Topic 几分钟就追平了，大 Topic 追了一天 lag 还没降。这时不要急着调参，先用kafka-consumer-groups.sh看源端消费组的CURRENT-OFFSET和LOG-END-OFFSET差多少，再结合带宽判断瓶颈。

当时的经验是：先把tasks.max拉开让不同分区由多个 Task 分担，再调linger.ms和batch.size降低请求数量。如果目标端 Topic 分区数远少于源端，写入并发上不去，就得先处理分区数对齐问题。注意目标端如果已经存在同名 Topic，MM2 不会自动扩分区，要在迁移前提前对齐好。

4.2 第二阶段：存量数据校验，别只看 end offset

等 lag 归零后，开始数据一致性校验。最粗的校验用 GetOffsetShell 对比两端每个 partition 的 end offset：

# 老集群 kafka-run-class.sh kafka.tools.GetOffsetShell \ --broker-list old-broker-1:9092,old-broker-2:9092,old-broker-3:9092 \ --topic order-events --time -1 # 新集群 kafka-run-class.sh kafka.tools.GetOffsetShell \ --broker-list new-broker-1:9092,new-broker-2:9092,new-broker-3:9092 \ --topic order-events --time -1

但端到端相同只是“基础合格”。由于迁移是持续复制，你对比时一定要选低峰期或明确源端没有写入的时间点，否则两端差异包含了正常增量。

更严的校验是抽样比对消息内容。我们当时写了一个小工具，按 partition 抽样读消息，对同一个 key 的最近一条消息做内容比对，确认业务关键字段完整。这里有个坑要提醒：复制到新集群后，消息的timestamp往往会被重写为复制时刻的时间戳，而不是原始生产时间。如果你下游有任务依赖 Kafka 消息时间戳做窗口计算，迁移后可能会出现时间数据错乱。务必提前和业务方确认，必要时在消息体内保存业务时间。

4.3 第三阶段：切流与回滚设计

我们采用的切流步骤是一个偏保守的“停写-追平-切读”方案：

1. 选业务低峰时段，通知所有生产端暂停向旧集群写入。
2. 等 MM2 把最后一批消息复制到目标集群，源端消费组 lag 归零。
3. 下游消费者指向新集群，订阅同名 Topic，auto.offset.reset设为latest。
4. 确认消费者正常消费后，恢复生产端直接写入新集群。
5. 观察一段时间，期间保留旧集群数据和 MM2 连接器不关，用于回滚。

这个方案的关键点是消费者用latest启动，所以不会把历史数据重新读一遍。因为停写期间没有新数据，所以也不会丢消息。代价是切流窗口内业务要接受写入暂停，时间通常控制在几分钟内。

如果业务完全无法停写，那就得走 checkpoint 精确续传：通过MirrorClient读取目标端检查点 Topic，拿到源端 group 的位点，再用seek()定位到新集群对应 partition。这条路更平滑，但需要每个下游都改造消费初始化逻辑，成本高。我们评估后没有采用，只在回滚预案里保留了方案。

回滚方面，旧集群没停、数据还在，只要把消费者改回旧集群并按原 group 的位点继续读即可。需要留意的是，切流后新集群产生的新数据默认不会自动回流旧集群，所以如果要在旧集群继续看到新数据，还得提前把反向 MM2（dc-new->dc-old.enabled=true）开起来。我们说开但没提前演练，后面复盘专门提了这件事。

5. 三个实战坑的完整排查链路

5.1 坑一：目标集群出现了一堆带前缀的 Topic

现象：迁移启动几小时后，业务方跑过来问“新集群里那些dc-old.xxx的 Topic 是哪来的”。我们一脸懵，再一看，原 Topic 名全都带了dc-old.前缀。

排查链路：

1. 先确认目标集群 Topic 列表，发现前缀统一的dc-old.，基本可以判断是 MM2 的默认 ReplicationPolicy 生效了。
2. 回头看 mm2.properties，确实没配置replication.policy.class。
3. 查文档确认默认行为是<源集群别名>.<原Topic名>，问题定位完成。

处理方案有两个，要么让业务消费者把订阅名改成dc-old.order-events，要么改 MM2 配置用IdentityReplicationPolicy保持原名。我们选了后者，因为这次迁移目标就是“下游无感知切换”。

但这里有个隐藏成本：改了策略之后，之前同步到dc-old.*的数据就废了，得清理掉重新同步。清理时我先把带前缀的 Topic 删掉，确认目标端没有同名旧 Topic，然后重启连接器。因为源端保留时间够长，重新同步没丢数据，但白白烧了不少带宽。这个坑的最佳规避方式就是动手前先想清楚最终消费端读的是哪个名字。

5.2 坑二：开了 sync.group.offsets 却依然从最新开始消费

现象：配置里sync.group.offsets.enabled=true，检查点 Topic 里也有数据，但下游切到新集群后，老 group 根本不会从老位点继续读。有同事直接质疑是不是检查点没生效。

排查链路：

1. 先确认检查点 Topic 确实有内容：用kafka-console-consumer.sh读mm2-checkpoints.<目标集群>.internal，能看到每个消费者组的位点记录。
2. 拉出记录和源端实际 committed offset 对比，发现位点数据本身没问题。
3. 看下游消费代码，发现用的就是普通 KafkaConsumer，没有任何读取检查点 Topic 的逻辑。
4. 查文档确认：MM2 只是把位点写入内部 Topic，标准 Consumer 不会自动消费这些数据。想要无缝续读必须显式走 MirrorClient 或自定义消费逻辑。

也就是说，检查点机制是“提供了数据”，但没有“自动应用”。对大部分迁移项目来说，与其为每个消费者写一套续读逻辑，不如用我们前面的“停写追平 + latest 启动”方案简单可靠。这个坑的本质是期望管理问题：把 MM2 的位点同步当成黑魔法，容易在切流当天被打个措手不及。

5.3 坑三：大 Topic 同步速度跟不上，带宽被打满

现象：一个日增几百 GB 的核心 Topic，启动 MM2 后源端消费组 lag 一直不降，监控显示旧集群 broker 网卡出方向带宽接近饱和，线上其他业务延迟升高。

排查链路：

1. 先看这个 Topic 的分区分布，发现源端 24 个分区，目标端同名 Topic 却只有 6 个分区。这是之前目标集群里已经有人建过同名 Topic 留下的残留。
2. 分区不对等导致 MM2 的写入能力天然受限：源端 24 个分区的并发拉取，只能往 6 个目标分区塞，单个分区压力大，写入瓶颈直接卡住整体 lag 下降。
3. 把目标端 Topic 扩到 24 个分区（当时数据量还不大，可以直接kafka-topics.sh --alter；如果目标端数据已多，扩分区前要评估是否会影响键有序性）。
4. 调tasks.max从 4 提到 12，让更多 Task 参与拉取和写入。
5. 调producer.override.linger.ms到 20、batch.size到 1MB，减少小消息请求对带宽的消耗。因为源端消息本身已是压缩格式，就没有再叠压缩层，避免额外 CPU 开销。

改完重启连接器，再观察半小时，lag 开始稳定下降。这个坑最难受的不是参数不会调，而是目标集群残留了同名 Topic 这个前置条件没被发现。所以迁移启动前检查目标端是否已有同名 Topic，分区数是否一致，必须写进检查清单。

6. 迁移完成后最后悔没做的三件事

6.1 没提前把源端 Topic 的 retention.ms 调大

源端默认日志保留七天，迁移窗口因为排期和追赶速度不可控，实际用了快两周。最开始的几天，早期日志已经被清理了，导致一部分历史数据压根没进新集群。虽然业务最关心的是最近数据，但这个“数据缺口”一旦被审计或补数任务发现，解释成本极高。

正确做法是决定迁移那一刻，就把所有涉及 Topic 的retention.ms临时调大到 30 天，确认新集群数据完整后再恢复原值。改配置可以在线操作，几乎零成本，但漏掉的后果是补数要重新找数据源。

6.2 没同步对齐目标 Topic 的关键参数

迁移同步的是消息，不是 Topic 配置。如果目标集群的 Topic 设置的cleanup.policy和源端不一致，麻烦在后面。我们有一个压缩 Topic，源端设置的是compact，目标端因为自动创建用的是默认delete。同步过去后，删除策略把旧版本 key 清掉了，下游查这个表的聚合视图时发现历史状态对不上。

另外min.insync.replicas、segment.ms、index.interval.bytes这些参数也会悄悄影响行为。建议迁移前写一个脚本，把源端 Topic 的配置读取出来，逐个在目标端比对，差异项手工对齐。跳过这步，后面往往以隐蔽的方式反咬一口。

6.3 没提前演练反向同步，回滚预案变成了纸面预案

切流后我一直在想，如果新集群出了问题，怎么回滚？旧集群数据还在，消费者切回去读原 group 位点是没问题的。但问题是切流之后，新集群持续写入的新数据并不会自动回流旧集群。等到真需要回滚时，才发现反向 MM2 配置虽然写在了文档里，却没有实际启动验证过。

正确的做法是迁移 Day 0 就把反向同步开起来，让新集群数据持续流回旧集群。这样既验证了反向链路可用，也让回滚时旧集群不会丢新数据。唯一要注意的是避免回环，反向同步时 Topic 过滤规则要仔细设计。

最后说一点个人体会：数据迁移最难的不是把消息搬过去，而是把“消费语义”一起搬过去。MM2 把增量复制、位点映射这些脏活标准化了，但它不会替你做业务决策。切流之前想清楚目标端用什么 Topic 名、消费者从什么位置启动、出问题往哪回退，比调任何参数都重要。如果后面再做类似的迁移，我会把上面的检查项全部固化成脚本和 checklist，宁可多花半天验证，也不在切流当天靠人肉救火。