如何使用Kafka MirrorMaker复制Schema及跨集群同步主题Schema？

我来帮你梳理下用Kafka MirrorMaker（优先推荐2.x版本，也就是MirrorMaker2——它比老版1.x集成度更高，对Schema Registry的支持也更原生）同步本地已注册Schema的主题到AWS Kafka集群的完整流程，包括Avro Schema的复制方案：

要实现完整同步，你需要兼顾主题数据同步和Schema Registry的Schema同步两部分，步骤如下：

1. 环境准备与前置检查 #

• 确保本地和AWS Kafka集群网络互通（可以通过VPN、专线或者公网访问，注意安全组/防火墙开放Kafka默认端口9092、MirrorMaker2用的9094，以及Schema Registry默认端口8081）
• 确认本地和AWS都部署了Confluent兼容的Schema Registry服务（如果用AWS MSK，也可以用它的Managed Schema Registry）
• 安装并配置MirrorMaker2（通常随Confluent Platform一起安装，或者单独下载部署）

2. 配置MirrorMaker2实现主题同步 #

创建MirrorMaker2的配置文件（比如mm2.properties），核心配置要包含本地源集群和AWS目标集群的连接信息，以及主题同步规则：

# 定义集群别名，方便后续配置
clusters = onprem, aws

# 本地源集群配置
onprem.bootstrap.servers = <本地Kafka集群地址，多个用逗号分隔>
onprem.schema.registry.url = <本地Schema Registry地址>

# AWS目标集群配置
aws.bootstrap.servers = <AWS MSK或自建Kafka地址>
aws.schema.registry.url = <AWS上的Schema Registry地址>

# 同步策略：复制所有主题（也可以指定前缀，比如onprem.*）
topics = .*
# 排除Kafka内部主题，避免不必要的同步
topics.blacklist = __consumer_offsets, __transaction_state

# 启用Schema Registry自动同步（关键配置）
sync.schema.registry.enabled = true
# 指定Schema Registry存储Schema的内部主题名（默认就是_schemas，保持一致即可）
schema.registry.sync.topic.name = _schemas

# MirrorMaker2的基础存储主题配置（首次启动会自动创建）
offset.storage.topic = mm2-offsets
config.storage.topic = mm2-configs
status.storage.topic = mm2-statuses

启动MirrorMaker2的命令：

bin/kafka-mirror-maker.sh --config mm2.properties

3. 验证同步结果 #

• 查看AWS集群上是否出现镜像主题（默认会带上源集群前缀，比如onprem.<原主题名>；如果需要去掉前缀，可以额外配置topic.mapping规则自定义名称）
• 生产一条Avro格式的消息到本地主题，检查AWS镜像主题是否能消费到正确的消息，同时登录AWS的Schema Registry，确认对应的Schema已经同步过来

MirrorMaker2提供了两种主要的Schema同步方式，你可以根据场景选择：

1. 自动同步（推荐） #

通过上面配置中的sync.schema.registry.enabled = true，MirrorMaker2会自动监听本地Schema Registry的_schemas内部主题，实时将新增/更新的Schema同步到AWS的Schema Registry。

关键注意点：

• 确保两个Schema Registry的兼容性设置一致（比如都是BACKWARD兼容），避免同步时出现兼容性校验错误
• 如果AWS上的Schema Registry已经有同名的Schema，MirrorMaker2会自动检查版本兼容性，只有兼容的版本才会同步
• 同步的Schema会保留原版本号，确保消费端能匹配到正确的Schema版本

2. 手动/工具同步（适合特殊场景） #

如果自动同步有问题，或者你需要更精细的控制（比如只同步特定版本的Schema），可以手动导出本地Schema再导入到AWS：

• 导出本地主题的最新Schema：

# 导出本地主题的value Schema（如果是key Schema，把-value改成-key）
curl -X GET <本地Schema Registry地址>/subjects/<主题名>-value/versions/latest > schema.json

• 导入到AWS Schema Registry：

# 导入Schema到AWS，同时指定兼容性规则
curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"schema": "'$(cat schema.json | tr -d '\n')'", "compatibility": "BACKWARD"}' \
  <AWS Schema Registry地址>/subjects/<主题名>-value/versions

额外注意事项 #

• 如果你用的是AWS MSK的Managed Schema Registry，要确认它和本地Schema Registry的API兼容性（MSK的Registry完全兼容Confluent的Schema Registry API）
• 同步过程中如果出现Schema同步失败，优先查看MirrorMaker2的日志，通常是兼容性问题或者权限不足（比如AWS IAM权限要允许MirrorMaker2访问MSK和Schema Registry）
• 对于已有的历史主题，MirrorMaker2会自动同步主题的配置（比如分区数、副本数），但要注意AWS集群的存储和副本配置是否能支持

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