从AWS MSK集群镜像到Apache Kafka的架构与配置咨询

工具选型结论 #

直接使用**connect-mirror-maker（MirrorMaker 2.0）**，理由如下：

• 基于Kafka Connect生态，完美兼容你已添加的AWS Glue Schema Registry转换器，能在镜像流程中直接完成Avro→JSON的格式转换
• 相比legacy版kafka-mirror-maker，MM2支持更灵活的主题映射、转换逻辑，以及内置的故障恢复能力
• 无需单独配置普通的源/ sink连接器，MM2自带镜像专用的连接器逻辑，简化配置

核心配置示例（mm2.properties） #

替换占位符为你的实际信息即可使用：

# 定义集群别名
clusters = source, target

# -------------------------- 源集群（AWS MSK）配置 --------------------------
source.bootstrap.servers = <MSK_BOOTSTRAP_SERVERS>
source.security.protocol = SASL_SSL
source.sasl.mechanism = AWS_MSK_IAM
source.sasl.jaas.config = software.amazon.msk.auth.iam.IAMLoginModule required;
source.sasl.client.callback.handler.class = software.amazon.msk.auth.iam.IAMClientCallbackHandler

# -------------------------- 目标集群配置 --------------------------
target.bootstrap.servers = <TARGET_KAFKA_BOOTSTRAP_SERVERS>
target.security.protocol = PLAINTEXT # 根据目标集群的安全协议调整（如SASL_SSL）

# -------------------------- MM2 核心配置 --------------------------
# MM2内部使用的存储主题，需确保目标集群允许自动创建或提前创建
offset.storage.topic = mm2-offsets
config.storage.topic = mm2-configs
status.storage.topic = mm2-statuses
replication.factor = 3 # 匹配目标集群的broker可用数量

# 指定要镜像的主题，支持通配符（如topic-*）
topics = <YOUR_SOURCE_AVRO_TOPIC>
# 目标主题命名规则，可自定义（比如改成${topic}-json）
topic.rename.format = ${topic}

# -------------------------- 源端反序列化（Avro → 结构化数据） --------------------------
source->target.enabled = true
# 解开Glue Schema Registry封装的Avro信封（如果你的数据带信封的话）
source->target.source.transforms = unwrap
source->target.source.transforms.unwrap.type = io.confluent.connect.transforms.UnwrapFromEnvelope
# 使用AWS Glue Schema Registry转换器反序列化Avro数据
source->target.source.converter = com.amazonaws.services.schemaregistry.kafkaconnect.AWSKafkaConnectSchemaRegistryConverter
source->target.source.converter.schema.registry.url = https://glue.<AWS_REGION>.amazonaws.com
source->target.source.converter.region = <AWS_REGION>
source->target.source.converter.auto.register.schemas = false # 源端已注册Schema，无需重复注册
source->target.source.converter.schema.name = <GLUE_SCHEMA_NAME> # 对应源主题绑定的Glue Schema名称

# -------------------------- 目标端序列化（结构化数据 → JSON） --------------------------
# 使用JSON转换器输出纯JSON格式
source->target.sink.converter = org.apache.kafka.connect.json.JsonConverter
source->target.sink.converter.schemas.enable = false # 输出不带Schema的纯JSON，如需带Schema可设为true

启动命令 #

在你的Confluent Connect镜像中，执行以下命令启动MM2：

connect-mirror-maker /path/to/mm2.properties

关键注意事项 #

• 确认所有AWS相关jar包已在Connect的类路径中（你已将jar包放入/usr/share/java/kafka，可通过echo $CLASSPATH验证）
• 运行Connect的身份需具备：
  • AWS MSK的IAM访问权限（kafka:DescribeCluster、kafka:ReadData等）
  • AWS Glue Schema Registry的访问权限（glue:GetSchemaVersion、glue:GetSchema等）
• 目标集群需允许MM2创建内部存储主题（或提前手动创建mm2-offsets、mm2-configs、mm2-statuses，确保分区数和副本数符合需求）

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