如何将多Kafka集群的同名主题合并至中心集群？

一、Confluent Replicator异常排查与修复 #

1. 版本不兼容问题 #

你遇到的java.lang.IllegalAccessError核心原因是Confluent Replicator与目标Kafka集群（或Connect运行时）的版本不匹配。NewReplicatorAdminClient试图访问的ConfigEntry构造方法，在不同Kafka/Confluent版本中存在访问权限或参数列表的变化，导致类加载时权限冲突。

修复步骤：

• 确保Replicator版本与中心集群C的Kafka、Connect版本完全一致（优先使用同一Confluent Platform系列版本，如7.4.x）。
• 替换边缘节点上的Replicator插件包，删除旧版本依赖，统一使用与目标集群匹配的版本。

2. 复制到已存在主题的配置调整 #

要将边缘集群的events主题合并到中心已存在的同名主题，需调整以下配置：

• 添加dest.topic.creation.enable: false：禁用Replicator自动创建目标主题，避免与已存在主题的配置冲突。
• 移除冗余的producer.override.bootstrap.servers：dest.kafka.bootstrap.servers已指定目标集群地址，该配置属于重复定义，可能引发逻辑冲突。

调整后的核心配置片段：

{
  "name": "replicator_test",
  "config": {
    "connector.class":"io.confluent.connect.replicator.ReplicatorSourceConnector",
    "tasks.max":1,
    "key.converter":"io.confluent.connect.replicator.util.ByteArrayConverter",
    "value.converter":"io.confluent.connect.replicator.util.ByteArrayConverter",
    "src.kafka.bootstrap.servers": "192.168.1.215:9092",
    "dest.kafka.bootstrap.servers":"192.168.1.157:9092",
    "topic.whitelist":"events_test,_schemas",
    "topic.rename.format":"${topic}",
    "confluent.topic.replication.factor": 1,
    "dest.topic.creation.enable": false,
    "src.kafka.timestamps.topic.replication.factor": 1
  }
}

3. 无消息写入的排查点 #

若调整后仍无消息写入，检查以下内容：

• 边缘集群的events_test主题是否有新消息产生（可通过kafka-console-consumer.sh工具验证）。
• Replicator日志是否存在新错误（比如目标集群是否允许边缘节点的Replicator写入events_test主题）。
• 中心集群主题的分区数、副本数配置是否与Replicator生产者配置兼容（比如生产者acks设置是否匹配）。

二、边缘数据收集的替代健壮方案 #

1. 开源MirrorMaker 2.0（MM2） #

MM2是Apache Kafka官方的跨集群复制工具，完全开源，支持单向/双向复制、主题合并，无需商业许可。每个边缘节点部署MM2实例，配置单向复制到中心集群，核心配置示例：

# mm2.properties
clusters=edge,center
edge.bootstrap.servers=192.168.1.215:9092
center.bootstrap.servers=192.168.1.157:9092

# 复制规则：将edge的events_test主题复制到center的同名主题
edge->center.enabled=true
edge->center.topics=events_test,_schemas
edge->center.topic.rename.format=${topic}
edge->center.replication.factor=1

2. Kafka Connect Sink Connector #

每个边缘节点部署独立的Kafka Connect集群，使用KafkaSinkConnector将本地events_test主题的数据直接写入中心集群的同名主题。这种方式轻量，无需额外Replicator/MM2组件，核心配置：

{
  "name": "edge-to-center-sink",
  "config": {
    "connector.class": "org.apache.kafka.connect.sink.KafkaSinkConnector",
    "tasks.max": 1,
    "topics": "events_test",
    "key.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
    "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
    "bootstrap.servers": "192.168.1.157:9092",
    "producer.acks": "all"
  }
}

3. 边缘缓冲+批量上传 #

针对网络波动频繁的场景，可在边缘节点添加本地缓冲层（如RocksDB或本地文件系统），当网络恢复时批量将数据上传到中心集群。可结合Kafka Streams或自定义Producer实现该逻辑，确保数据不丢失。

内容的提问来源于stack exchange，提问作者Kjell van Straaten