如何将两个Kafka集群不同主题的消息转发至第三个集群的指定主题

可行实现方案 #

• Kafka MirrorMaker 2.0（官方原生方案）
  你认为镜像方案不支持主题重命名属于常见误解，Kafka官方自带的MirrorMaker 2.0（MM2）原生支持自定义主题映射规则，无需额外开发。只需要单独部署两套MM2同步作业：

  1. 第一套作业负责集群A到集群C的同步，配置topic.include为incoming.dataA，新增配置replication.policy.class=org.apache.kafka.connect.mirror.IdentityReplicationPolicy，同时添加topic.alias.map=incoming.dataA:received.data即可直接将A集群的incoming.dataA同步到C集群的received.data
  2. 第二套作业负责集群B到集群C的同步，参考上述配置将源主题替换为incoming.dataB即可
     该方案自带偏移量同步、故障自动转移、Exactly Once语义支持，适合生产环境大流量场景使用。

• 自定义生产者-消费者同步程序
  轻量场景可直接写极简的自定义同步程序，核心逻辑仅需两步：

  1. 初始化两个消费者实例，分别连接集群A、集群B，订阅对应源主题
  2. 初始化一个连接集群C的生产者实例，消费到源端消息后直接投递到received.data主题，投递成功后手动提交源端消费偏移量
     以下是Python版本的核心示例代码：

  from confluent_kafka import Consumer, Producer, KafkaError
  
  # 集群A消费者配置
  consumer_a = Consumer({'bootstrap.servers': 'kafka-a:9092', 'group.id': 'sync-group-a', 'auto.offset.reset': 'earliest', 'enable.auto.commit': False})
  consumer_a.subscribe(['incoming.dataA'])
  # 集群B消费者配置
  consumer_b = Consumer({'bootstrap.servers': 'kafka-b:9092', 'group.id': 'sync-group-b', 'auto.offset.reset': 'earliest', 'enable.auto.commit': False})
  consumer_b.subscribe(['incoming.dataB'])
  # 集群C生产者配置
  producer_c = Producer({'bootstrap.servers': 'kafka-c:9092', 'acks': 'all', 'enable.idempotence': True})
  
  def delivery_report(err, msg):
      if not err:
          # 投递成功提交源端偏移量
          if msg.topic() == 'incoming.dataA':
              consumer_a.commit()
          else:
              consumer_b.commit()
  
  # 轮询消费转发
  while True:
      for consumer in [consumer_a, consumer_b]:
          msg = consumer.poll(1.0)
          if msg and not msg.error():
              producer_c.produce('received.data', value=msg.value(), key=msg.key(), timestamp=msg.timestamp()[1], on_delivery=delivery_report)
      producer_c.flush()
  

  该方案灵活度最高，可自定义消息过滤、格式转换、分区路由规则。

• 流处理框架方案
  你提到流处理方案无法满足需求大概率是未配置自定义输出主题规则，Kafka Streams、Flink等主流流处理框架都支持多集群源接入，将两个源主题的流合并后写入指定目标主题即可，无需和源主题同名。
  以Kafka Streams为例，分别配置两个源集群的消费端参数，读取两个源主题的流后调用merge()方法合并，最后调用to()方法指定输出到集群C的received.data主题即可。

• 通用数据采集组件方案
  不想写代码的场景可以用Logstash、Vector、Apache Flume这类通用数据采集组件，所有组件都支持多Kafka数据源输入、自定义目标Kafka主题配置，只需要修改配置文件即可完成同步，无需开发。

注意事项 #

• 如果需要保留原主题的消息顺序，转发时可以将源集群标识+原分区号作为目标消息的key，保证同一原分区的消息会落到目标主题的同一个分区，避免乱序
• 生产环境建议开启幂等生产者、手动提交偏移量，避免消息丢失或重复

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