迁移手册

# 前言

本文档为金蝶Apusic分布式消息队列for Kafka（Apusic Distributed Message Queue，简称：ADMQ for Kafka）的迁移手册，详细介绍了如何将Kafka迁移到ADMQ for Kafka。

# 适用对象

本文档适用于IT信息化业务负责人、研发经理、软件项目经理、软件架构师、运维工程师。

# 相关文档

了解更多ADMQ for Kafka产品相关的信息，请参阅以下ADMQ for Kafka产品手册文档集：

序号 手册文档 说明
1 金蝶Apusic分布式消息队列for Kafka 快速使用手册 简单介绍了如何快速上手使用ADMQ for Kafka 。
2 金蝶Apusic分布式消息队列for Kafka 安装手册 详细介绍如何在各操作系统上安装ADMQ for Kafka，以及ADMQ for Kafka服务启停等操作。
3 金蝶Apusic分布式消息队列for Kafka 消息引擎用户手册 详细介绍 ADMQ for Kafka 消息引擎相关功能的使用、配置、管理及配套工具的使用方法。
4 金蝶Apusic分布式消息队列for Kafka 管控台用户手册 详细介绍ADMQ for Kafka管控台相关功能的使用和操作说明。
5 金蝶Apusic分布式消息队列for Kafka 开发手册 详细介绍基于各开发语言进行ADMQ for Kafka客户端应用开发的说明。
6 金蝶Apusic分布式消息队列for Kafka 迁移手册 详细介绍从Kafka迁移到ADMQ for Kafka的说明。
7 金蝶Apusic分布式消息队列for Kafka 运维手册 详细介绍ADMQ for Kafka的监控、运维、安全加固等运维说明。
8 金蝶Apusic分布式消息队列for Kafka 性能优化手册 详细介绍ADMQ for Kafka性能调优的说明。

# 技术支持

ADMQ for Kafka产品提供全面的技术支持服务，您可以通过以下方式获得技术支持：

  • 网址：www.apusic.com
  • 电话：400-855-5800
  • 邮箱：support@apusic.com
  • 金蝶云社区：https://vip.kingdee.com/?productId=73&productLineId=14&lang=zh-CN

您在取得技术支持时，请提供如下信息：

  1. 您的姓名

  2. 公司信息与联系方式

  3. 操作系统及其版本

  4. 产品版本号

  5. 出现异常及错误的日志、截图等详细信息

# 概述

本文档详细介绍将现有 Kafka 集群迁移到 ADMQ for Kafka V2.0.6 的最佳实践，包含兼容版本、配置迁移、元数据迁移和数据迁移。

# 兼容的 Kafka 版本

ADMQ for Kafka V2.0.6 兼容以下版本的 Kafka：

源 Kafka 版本 兼容性 说明
2.4.x 完全兼容 推荐迁移版本
2.5.x 完全兼容 推荐迁移版本
2.6.x 完全兼容 推荐迁移版本
2.7.x 完全兼容 推荐迁移版本
2.8.x 完全兼容 推荐迁移版本
3.0.x 完全兼容 推荐迁移版本
3.1.x 完全兼容 推荐迁移版本
3.2.x 完全兼容 推荐迁移版本
3.3.x 完全兼容 推荐迁移版本
3.4.x 完全兼容 推荐迁移版本
3.5.x 完全兼容 推荐迁移版本
3.6.x 完全兼容 推荐迁移版本
3.7.x 完全兼容 推荐迁移版本
3.8.x 完全兼容 推荐迁移版本

# 迁移策略

策略 适用场景 优点 缺点
滚动升级 同版本内小版本升级 无需停机 不适用于跨大版本
双集群迁移 异构集群迁移 迁移风险低 需要双倍资源
渐进式迁移 生产环境迁移 逐步验证 迁移周期长

# 迁移前准备

# 1. 环境评估

# 检查源集群状态
kafka/bin/kafka-broker-api-versions.sh --bootstrap-server <source-kafka>:9092

# 查看集群信息
kafka/bin/kafka-topics.sh --list --bootstrap-server <source-kafka>:9092
kafka/bin/kafka-consumer-groups.sh --list --bootstrap-server <source-kafka>:9092
1
2
3
4
5
6

# 2. 数据量评估

# 查看 Topic 数量
kafka/bin/kafka-topics.sh --list --bootstrap-server <source-kafka>:9092 | wc -l

# 查看消息总量
kafka/bin/kafka-run-class.sh kafka.admin.ConsumerGroupCommand \
  --bootstrap-server <source-kafka>:9092 \
  --list
1
2
3
4
5
6
7

# 3. 安装 ADMQ

按照安装手册部署 ADMQ for Kafka 集群，确保：

  • 网络与源集群互通
  • 配置与源集群兼容的参数

# 配置迁移

# Broker 配置迁移

需要迁移的关键配置：

配置项 说明 迁移建议
broker.id Broker ID 保持一致或重新分配
listeners 监听地址 根据新环境调整 IP
log.dirs 数据目录 重新指定目录
zookeeper.connect ADCC for zk 地址 指向新 ADCC for zk 集群
advertised.listeners 广播地址 配置为新集群地址
compression.type 压缩类型 保持原配置
log.retention.* 日志保留 按需调整
num.partitions 默认分区数 保持一致

# 配置映射示例

# 源集群配置 (server.properties)
broker.id=0
listeners=PLAINTEXT://192.168.1.10:9092
log.dirs=/data/kafka
zookeeper.connect=zk1:2181,zk2:2181,zk3:2181/kafka

# ADMQ 配置 (kafka-broker.conf)
broker.id=0
listeners=PLAINTEXT://192.168.1.20:9092
log.dirs=/apusic/admq-kafka/data
zookeeper.connect=192.168.1.20:2181,192.168.1.21:2181,192.168.1.22:2181/admq
1
2
3
4
5
6
7
8
9
10
11

# JAAS 配置迁移

如果使用 SASL 认证：

# 迁移 JAAS 配置（以下密码仅为示例，生产环境请使用强密码）
KafkaServer {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="admin"
  password="admin-secret"
  user_admin="admin-secret"
  user_producer="producer-secret"
  user_consumer="consumer-secret";
};
1
2
3
4
5
6
7
8
9

# 元数据迁移

# Topic 迁移

# 方法一：使用 kafka-reassign-partitions（推荐）

# 1. 生成迁移方案
kafka/bin/kafka-reassign-partitions.sh \
  --bootstrap-server <source-kafka>:9092 \
  --topics-to-move-json-file topics.json \
  --broker-list "0,1,2" \
  --generate

# topics.json 格式
{
  "topics": [{"topic": "topic1"}, {"topic": "topic2"}]
}

# 2. 执行迁移
kafka/bin/kafka-reassign-partitions.sh \
  --bootstrap-server <source-kafka>:9092 \
  --reassignment-json-file reassign.json \
  --execute

# 3. 验证迁移进度
kafka/bin/kafka-reassign-partitions.sh \
  --bootstrap-server <source-kafka>:9092 \
  --reassignment-json-file reassign.json \
  --verify
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# 方法二：导出导入 Topic 配置

# 导出 Topic 配置
kafka/bin/kafka-configs.sh --bootstrap-server <source-kafka>:9092 \
  --entity-type topics \
  --entity-name my-topic \
  --describe > topic-config-backup.txt

# 在 ADMQ 中创建同名 Topic
kafka/bin/kafka-topics.sh --create \
  --bootstrap-server <admq-kafka>:9092 \
  --topic my-topic \
  --partitions 6 \
  --replication-factor 3
1
2
3
4
5
6
7
8
9
10
11
12

# ACL 迁移

# 导出 ACL
kafka/bin/kafka-acls.sh --bootstrap-server <source-kafka>:9092 \
  --list > acls-backup.txt

# 在 ADMQ 中创建相同 ACL
kafka/bin/kafka-acls.sh --bootstrap-server <admq-kafka>:9092 \
  --add --allow-principal User:producer \
  --operation Write --topic my-topic
1
2
3
4
5
6
7
8

# 消费者组 Offset 迁移

# 导出消费者组 Offset
kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server <source-kafka>:9092 \
  --group my-group \
  --describe > offset-backup.txt

# 重置 Offset（适用于需要保留消费进度的场景）
kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server <admq-kafka>:9092 \
  --group my-group \
  --topic my-topic \
  --reset-offsets --to-offset <offset-number> \
  --execute
1
2
3
4
5
6
7
8
9
10
11
12
13

# 数据迁移

# 使用 MirrorMaker 2 迁移

MirrorMaker 2 是 Kafka 官方推荐的数据迁移工具，支持：

  • 主题数据同步
  • ACL 同步
  • 配置同步
  • 消费者组 Offset 同步

# 配置 MirrorMaker 2

# mm2.properties
clusters=source,target
source.bootstrap.servers=<source-kafka>:9092
target.bootstrap.servers=<admq-kafka>:9092

source->target.enabled=true
source->target.topics.regex=.*

# 内部 Topic 配置
source->target.emit.syncs.interval.seconds=60
source->target.sync.group.offsets.enabled=true
source->target.sync.group.offsets.interval.seconds=60

# 命名空间配置
source->target.renames=sourcePrefixTransformerClass=org.apache.kafka.connect.transforms.LogRename
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

# 启动 MirrorMaker 2

# 使用 connect-mirror-maker.sh
kafka/bin/connect-mirror-maker.sh mm2.properties &

# 或者使用分布式模式
kafka/bin/connect-distributed.sh connect-distributed.properties
1
2
3
4
5

# 使用 Kafka Connect 迁移

// source-connector.json
{
  "name": "source-connector",
  "config": {
    "connector.class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
    "source.cluster.alias": "source",
    "target.cluster.alias": "target",
    "source.bootstrap.servers": "<source-kafka>:9092",
    "target.bootstrap.servers": "<admq-kafka>:9092",
    "topics": ".*",
    "sync.group.offsets.enabled": "true"
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
# 创建 Connector
curl -X POST -H "Content-Type: application/json" \
  --data @source-connector.json \
  http://localhost:8083/connectors/
1
2
3
4

# 数据校验

# 验证数据完整性
kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server <admq-kafka>:9092 \
  --group mirror-consumer-group \
  --describe

# 对比源集群和目标集群消息数量
kafka/bin/kafka-run-class.sh kafka.tools.GetOffsetShell \
  --broker-list <source-kafka>:9092 \
  --topic my-topic \
  --time -2 | tail -1

kafka/bin/kafka-run-class.sh kafka.tools.GetOffsetShell \
  --broker-list <admq-kafka>:9092 \
  --topic my-topic \
  --time -2 | tail -1
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

# 客户端迁移

# 切换客户端连接地址

将客户端配置中的 bootstrap.servers 从源集群地址改为 ADMQ 地址：

# 原配置
bootstrap.servers=192.168.1.10:9092,192.168.1.11:9092

# 新配置（ADMQ）
bootstrap.servers=192.168.1.20:9092,192.168.1.21:9092
1
2
3
4
5

# 代码层面无需修改

由于 ADMQ 完全兼容 Kafka 协议，以下方面无需修改：

  • 生产者/消费者代码
  • 序列化/反序列化方式
  • 消费者组管理方式
  • 主题操作方式

# 验证客户端迁移

# 测试生产者
kafka/bin/kafka-console-producer.sh \
  --bootstrap-server <admq-kafka>:9092 \
  --topic test-migration

# 测试消费者
kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server <admq-kafka>:9092 \
  --topic test-migration \
  --from-beginning
1
2
3
4
5
6
7
8
9
10

# 迁移步骤

# 步骤 1：部署 ADMQ 集群

按照安装手册完成 ADMQ 部署。

# 步骤 2：验证集群

# 验证 ADMQ 集群健康
kafka/bin/kafka-broker-api-versions.sh --bootstrap-server <admq-kafka>:9092
kafka/bin/kafka-topics.sh --create --topic test --partitions 3 --replication-factor 3
1
2
3

# 步骤 3：配置迁移

将源集群的配置文件参数对应迁移到 ADMQ 配置中。

# 步骤 4：元数据迁移

使用工具同步 Topic、ACL、消费者组等元数据。

# 步骤 5：数据迁移

使用 MirrorMaker 2 或 Kafka Connect 同步数据。

# 步骤 6：客户端切换

分批将客户端切换到 ADMQ 集群。

# 步骤 7：验证

  • 验证消息完整性
  • 验证消费进度
  • 验证业务功能

# 步骤 8：清理

确认迁移成功后，下线源集群。

# 迁移风险控制

# 回滚方案

  1. 保留源集群一段时间
  2. 客户端配置支持快速切换
  3. 记录关键配置和状态

# 迁移窗口

数据量 预计迁移时间 说明
< 1TB 2-4 小时 小型集群
1-10TB 4-12 小时 中型集群
> 10TB 1-3 天 大型集群

# 注意事项

  1. 网络带宽：确保源集群到 ADMQ 集群的网络带宽充足
  2. 磁盘空间：ADMQ 集群需要足够的磁盘空间存储数据
  3. 业务低峰期：建议在业务低峰期进行数据迁移
  4. 监控告警：开启监控告警，及时发现异常

# 常见问题

# Topic 创建失败

原因：Topic 名称冲突或分区数超过 Broker 数量 解决：检查 Topic 名称，确保分区数 ≤ Broker 数

# 数据同步延迟

原因：网络带宽不足或同步任务配置不当 解决：检查网络，增加 MirrorMaker 并发数

# 消费进度丢失

原因：未同步消费者组 Offset 解决：启用 sync.group.offsets.enabled 配置

# ACL 不生效

原因：ACL 配置未同步 解决：手动同步 ACL 配置

编辑页面 (opens new window)
#迁移手册

← 运维手册→