Kafka Topic 完整迁移方案

核心概念 Kafka Topic 迁移分同集群和跨集群两大类，使用 kafka-reassign-partitions.sh 和 MirrorMaker2 实现不停服零丢失迁移。

1. 前置说明

迁移分两大类：

1. 同集群内迁移：Topic 分区跨节点/跨机架重平衡；
2. 跨集群迁移：源集群A → 目标集群B（业务最常用，机房迁移、集群拆分、版本升级）；
   工具：kafka-reassign-partitions.sh（同集群）、MirrorMaker2（MM2，官方标准跨集群迁移）、Filebeat/自定义消费生产（小流量临时方案）。

一、场景1：同集群迁移Topic分区（节点更换、机架均衡、磁盘扩容）

适用：集群机器增减、热数据挪高速SSD节点、分片打散避免单机热点。
工具：kafka-reassign-partitions.sh 分区重分配工具，不丢消息、不影响读写。

2. 完整步骤

2.1. 生成迁移分配json

# 模板生成：指定topic、目标broker列表
kafka-reassign-partitions.sh \
--bootstrap-server 127.0.0.1:9092 \
--topics-to-move-json-file topics.json \
--broker-list 1,2,3 \
--generate

topics.json 内容：

{"topics": [{"topic": "order_topic"}], "version":1}

输出一份 proposed-assignment.json，里面是每个分区+副本分配到哪些broker。

2.2. 执行迁移

kafka-reassign-partitions.sh \
--bootstrap-server 127.0.0.1:9092 \
--reassignment-json-file proposed-assignment.json \
--execute

后台开始同步分区副本数据，流量小无感知。

2.3. 实时查看迁移进度

kafka-reassign-partitions.sh \
--bootstrap-server 127.0.0.1:9092 \
--reassignment-json-file proposed-assignment.json \
--verify

输出 completed successfully 代表迁移完成。

3. 关键优化避坑

1. 迁移限流，防止打满集群网卡IO

# server.properties 临时调小同步带宽
replica.fetch.max.bytes=10485760

2. 多副本同时迁移会暴涨IO，分批迁移Topic；
3. 不要业务高峰执行迁移，选择凌晨低峰；
4. 迁移过程生产者消费者无需停服，正常读写。

二、场景2：跨集群迁移（官方标准 MirrorMaker 2.0 MM2）

适用：机房搬迁、旧集群下线、Kafka版本升级、多集群数据同步，支持双向同步、自动同步偏移量、无缝切流。
架构：源集群Source → MM2同步器 → 目标集群Target，Topic名称可加前缀/同名。

4. 部署步骤

4.1. 创建MM2配置文件 mm2.properties

# 1. 两端集群地址
clusters=source,target
source.bootstrap.servers=10.0.1.1:9092
target.bootstrap.servers=10.0.2.1:9092

# 2. 集群别名映射
source.cluster.alias=src
target.cluster.alias=dst

# 3. 同步规则：匹配topic正则
topics=.*
topics.exclude=__consumer_offsets,__transaction_state

# 4. 同步消费者偏移量（核心，切流不丢数据）
sync.offsets.enabled=true
sync.offsets.interval.seconds=30

# 5. 目标topic自动创建、副本数同步
auto.create.topics.enable=true
replication.factor=3

# 6. 流量限流，保护集群
replication.throttled.rate=10485760

4.2. 启动MirrorMaker2

kafka-mirror-maker.sh mm2.properties

MM2内部封装消费者+生产者，持续拉取源集群消息写入目标集群。

4.3. 校验同步状态

1. 目标集群查看topic是否自动创建：

kafka-topics.sh --bootstrap-server target:9092 --list

2. 两端生产消息，消费验证数据一致；
3. 查看偏移同步任务：src-dst-offsets-sync 内部topic。

4.4. 业务无缝切换流程（零丢失核心）

1. MM2持续双向同步，两端数据完全对齐；
2. 业务生产者切到目标集群；
3. 等待源集群积压消息全部同步完成；
4. 消费者切到目标集群，MM2同步的偏移量保证从头续读；
5. 稳定运行7天无异常，关停旧集群MM2、下线源集群。

5. MM2优势

1. 官方组件，支持事务消息、幂等；
2. 自动同步消费者offset，不用手动记录位点；
3. 支持心跳、故障重试、限流；
4. 支持跨版本Kafka迁移（0.10→2.8→3.x）。

三、场景3：小流量临时迁移（控制台消费生产，无组件依赖）

适用：单Topic、数据量小、一次性迁移、临时测试，不适合生产长期同步。

6. 操作流程

1. 开启控制台消费源集群，输出标准输出；
2. 管道输入目标集群生产者；

# 持续同步所有消息（从头开始 --from-beginning）
kafka-console-consumer.sh --bootstrap-server source:9092 --topic order_topic --from-beginning \
| kafka-console-producer.sh --bootstrap-server target:9092 --topic order_topic

缺陷：

• 不保存offset，中断需要从头重新同步；
• 不支持批量、限流、事务；
• 大流量极易丢消息，仅临时应急。

四、场景4：离线全量迁移（存量历史数据一次性迁移，增量走MM2）

业务场景：Topic存在大量历史存量数据，先一次性同步历史，再MM2同步实时增量。

7. 方案：kafka-dump + 导入

1. 源集群导出全量消息到本地文件：

kafka-dump-log.sh --bootstrap-server source:9092 --topic order_topic --output file.dat

2. 文件批量导入目标集群：

kafka-console-producer.sh --bootstrap-server target:9092 --topic order_topic < file.dat

3. 导出完成后启动MM2同步实时增量，保证数据完整。

五、迁移关键生产保障（防丢消息核心规范）

8. 数据零丢失配置

1. 生产者开启幂等+事务：acks=all、retries=3；
2. Topic副本数≥2，min.insync.replicas=2；
3. MM2开启offset同步，切换消费不重复/不丢失；
4. 迁移期间禁止删除源集群Topic，观察7天业务稳定再下线。

9. 流量限流保护集群

1. 同集群分区迁移：调整replica.fetch.max.bytes；
2. MM2迁移：配置replication.throttled.rate限制同步带宽；
3. 业务低峰执行迁移，避开早高峰/晚高峰。

10. 迁移前后校验清单

1. 两端Topic分区数、副本数完全一致；
2. 消息条数对比（kafka-run-class.sh 统计offset最大值）；
3. 消费者分组offset同步正常；
4. 业务生产消费无报错、延迟平稳；
5. 无重复消息、无丢失消息。

六、常见坑与故障解决

1. MM2同步延迟持续上涨

• 目标集群磁盘IO不足、带宽限流过小、目标topic分区少于源；

2. 分区重分配导致集群CPU飙升

• 迁移并发过高，分批执行，调小同步带宽；

3. 切换消费者后重复消费/丢失

• 未开启MM2 offset同步，手动导出导入消费者offset；

4. 跨版本MM2同步报错协议不兼容

• MM2部署高版本Kafka客户端，向下兼容低版本源集群；

5. 同集群迁移时业务写入超时

• 副本同步抢占IO，业务高峰暂停迁移。

七、方案选型总结

1. 同集群换节点/均衡分区 → kafka-reassign-partitions.sh
2. 跨机房、跨集群长期实时同步、版本升级 → MirrorMaker2（MM2，生产首选）
3. 少量数据一次性临时迁移 → console-consumer + console-producer
4. 海量历史存量离线迁移 → 日志dump导出导入 + MM2增量同步