Kafka/MSK迁移⽅案
⽅案概览
⼯具：KafkaMirrorMaker2
配置参数
代码块
1 # 从最早消息开始同步
2 consumer.override.auto.offset.reset=earliest
3 # 同步 topic 级别配置信息
4 sync.topic.configs.enabled = true
5 # 同步 consume group 的 offset
6 sync.group.offsets.enabled = true
7 # 同步后的 topic 去除前缀
```
8 replication.policy.class =
org.apache.kafka.connect.mirror.IdentityReplicationPolicy
配置样例
代码块
1 #Apache Kafka clusters
2 clusters = source, target
3 source.bootstrap.servers = b-2.pbqa.bmzqqf.c3.kafka.ap-northeast-
1.amazonaws.com:9092,b-1.pbqa.bmzqqf.c3.kafka.ap-northeast-
1.amazonaws.com:9092,b-3.pbqa.bmzqqf.c3.kafka.ap-northeast-1.amazonaws.com:9092
4 target.bootstrap.servers = b-1.awsjpdevltpmm2dem.4os4ku.c3.kafka.ap-northeast-
1.amazonaws.com:9092,b-2.awsjpdevltpmm2dem.4os4ku.c3.kafka.ap-northeast-
1.amazonaws.com:9092,b-3.awsjpdevltpmm2dem.4os4ku.c3.kafka.ap-northeast-
1.amazonaws.com:9092
6 source->target.enabled = true
8 #Source and target cluster configuration
9 source.config.storage.replication.factor = 3
10 target.config.storage.replication.factor = 3
11 source.offset.storage.replication.factor = 3
12 target.offset.storage.replication.factor = 3
14 #Mirror Maker configuration
16 replication.policy.class =
org.apache.kafka.connect.mirror.IdentityReplicationPolicy
18 offset-sync.topic.replication.factor = 3
19 heartbeat.topic.replication.factor = 3
20 checkpoint.topic.replication.factor = 3
22 topics = .*
23 groups = .*
24 consumer.override.auto.offset.reset=earliest
26 tasks.max = 1
27 replication.factor = 3
28 refresh.topics.enabled = true
29 sync.topic.configs.enabled = true
30 sync.group.offsets.enabled = true
32 #Enable heartbeats and checkpoints
33 source->target.emit.heartbeats.enabled = true
34 source->target.emit.checkpoints.enabled = true
问题及限制
https://docs.redpanda.com/redpanda-cloud/develop/managed-connectors/create-mmaker-
checkpoint-connector/
🎨 TheMirrorMaker2Checkpointconnectordoesnotmigrateconsumergroupoffsetsthat
arelowerthanthehighestoffsetssyncedbytheMirrorMaker2Sourceconnectorbythe
timetheMirrorMaker2Checkpointconnectorisstarted.
测试
1. 启动MM2：
sh bin/connect-mirror-maker.sh mm2.properties
2. Source集群写⼊消息，并启动消费
3. Source集群停⽌消费，并继续写⼊消息
4. 查看Source端LAG信息，落后3个消息
5. 查看Target端LAG信息，可⻅于Source集群保持⼀致，Offset,Lag信息保持⼀致
6. Target集群启动消费，基于Source集群offset继续消费，未出现重复消费及遗漏消息
```
---
*本文由OpenClaw自动从PDF转换生成*
*源文件: Kafka_MSK 迁移方案.pdf*