如何将K8s集群中Confluent Kafka(cp-kafka:5.4.1)迁移至Bitnami Kafka？

一、Mirror Maker 迁移完全可行，先排查之前的失效原因 #

你之前用Mirror Maker 2.8.1未复制数据且无日志，核心是配置未适配Bitnami Kafka的默认设置，和Confluent Kafka之间的迁移逻辑本身没问题，重点检查以下几点：

1. 集群访问地址 #

确保Mirror Maker使用K8s集群内的headless service地址访问源/目标集群，比如confluent-kafka-headless:9092、bitnami-kafka-headless:9092，不要用外部暴露地址。

2. Topic预创建配置 #

Bitnami Kafka默认关闭auto.create.topics.enable，需先在目标集群手动创建所有待迁移topic（保持和源集群一致的分区数、副本数），或临时开启目标集群的auto.create.topics.enable=true，迁移完成后再关闭。

3. 安全认证匹配 #

• 若源Confluent Kafka开启SASL认证，需在Mirror Maker配置中添加对应JAAS参数；
• Bitnami Kafka默认启用SASL_PLAINTEXT，必须在Mirror Maker的生产者/消费者配置中匹配security.protocol=SASL_PLAINTEXT、sasl.mechanism=PLAIN及对应的账号密码配置。

4. 日志输出配置 #

之前无日志是因为容器默认日志级别过高，启动Mirror Maker时添加参数--log4j.logger.org.apache.kafka=INFO，或挂载自定义log4j配置文件，强制输出日志便于排查。

二、正确的Mirror Maker 2.x迁移步骤（推荐） #

1. 编写MM2配置文件（mm2.properties） #

# 源Confluent Kafka集群配置
clusters.source.bootstrap.servers=confluent-kafka-headless:9092
# 若源集群有SASL认证，取消注释以下配置
# clusters.source.security.protocol=SASL_PLAINTEXT
# clusters.source.sasl.mechanism=PLAIN
# clusters.source.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin-secret";

# 目标Bitnami Kafka集群配置
clusters.target.bootstrap.servers=bitnami-kafka-headless:9092
clusters.target.security.protocol=SASL_PLAINTEXT
clusters.target.sasl.mechanism=PLAIN
clusters.target.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="user" password="bitnami-secret";

# 复制规则：同步所有topic、配置及消费者偏移量
topics=.*
sync.topic.configs=true
sync.group.offsets=true

# 开启源到目标的复制流
source->target.enabled=true
source->target.topics=.*

2. 部署MM2（K8s或Docker方式） #

• Docker方式（集群内运行）：

docker run -v $(pwd)/mm2.properties:/opt/kafka/config/mm2.properties apache/kafka:2.8.1 \
  bin/kafka-mirror-maker.sh --config /opt/kafka/config/mm2.properties

• K8s方式：将配置文件挂载为ConfigMap，用Deployment运行apache/kafka:2.8.1镜像，执行上述启动命令。

3. 验证迁移进度 #

• 检查目标集群topic数据：kafka-console-consumer.sh --bootstrap-server bitnami-kafka-headless:9092 --topic your-topic --from-beginning
• 通过MM2日志确认复制完成状态。

三、备选方案：数据导出+导入 #

如果Mirror Maker仍有问题，可采用更直接的离线迁移方式：

1. 导出源集群所有topic数据 #

# 遍历源集群所有topic并导出
for topic in $(kafka-topics.sh --list --bootstrap-server confluent-kafka-headless:9092); do
  kafka-console-consumer.sh --bootstrap-server confluent-kafka-headless:9092 --topic $topic --from-beginning --property print.key=true --property key.separator=":" > $topic.data
done

2. 在目标集群创建匹配的topic #

# 复制源集群topic的分区数、副本数
for topic in $(kafka-topics.sh --list --bootstrap-server confluent-kafka-headless:9092); do
  partitions=$(kafka-topics.sh --describe --bootstrap-server confluent-kafka-headless:9092 --topic $topic | grep -oE "PartitionCount: [0-9]+" | awk '{print $2}')
  replicas=$(kafka-topics.sh --describe --bootstrap-server confluent-kafka-headless:9092 --topic $topic | grep -oE "ReplicationFactor: [0-9]+" | awk '{print $2}')
  kafka-topics.sh --create --bootstrap-server bitnami-kafka-headless:9092 --topic $topic --partitions $partitions --replication-factor $replicas
done

3. 导入数据到目标集群 #

# 将导出的数据批量导入目标集群
for topic_file in $(ls *.data); do
  topic_name=${topic_file%.data}
  kafka-console-producer.sh --bootstrap-server bitnami-kafka-headless:9092 --topic $topic_name --property parse.key=true --property key.separator=":" < $topic_file
done

4. 同步消费者组偏移量 #

导出源集群消费者偏移量，手动适配格式后导入目标集群：

# 导出源集群偏移量
kafka-consumer-groups.sh --describe --bootstrap-server confluent-kafka-headless:9092 --all-groups > consumer-offsets.txt

# 手动整理格式后，用kafka-consumer-groups.sh --reset-offsets导入到目标集群

四、迁移后验证 #

1. 确认所有topic数据完整，消费无异常；
2. 切换业务应用到Bitnami Kafka，验证生产消费链路正常；
3. 确认业务完全切换后，关闭源Confluent Kafka集群。

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