如何将Kafka集群消息复制到另一集群并按自定义头过滤

可以通过Kafka Connect的Source + Sink连接器组合实现需求，核心是利用Kafka Connect的消息转换（Transform）功能过滤掉无指定自定义头的消息，完全不需要依赖MirrorMaker。

实现步骤 #

1. 配置Source连接器（从集群1读取并过滤消息） #

使用官方的KafkaSourceConnector从集群1拉取指定主题的数据，同时通过Filter转换规则过滤掉没有目标自定义消息头的消息。

示例配置（保存为cluster1-source.properties）：

name=cluster1-to-cluster2-source
connector.class=org.apache.kafka.connect.source.KafkaSourceConnector
tasks.max=2
# 指定要复制的集群1主题
topics=your-target-topic
# 集群1的Broker地址
bootstrap.servers=cluster1-host:9092
# 启用消息过滤转换
transforms=filterMissingHeader
transforms.filterMissingHeader.type=org.apache.kafka.connect.transforms.Filter$Value
# 替换为你的自定义消息头名称，过滤掉无此头的消息
transforms.filterMissingHeader.condition=headers['your-custom-header-key'] != null

2. 配置Sink连接器（写入集群2同名主题） #

使用官方的KafkaSinkConnector将经过过滤的数据写入集群2的同名主题。

示例配置（保存为cluster2-sink.properties）：

name=cluster1-to-cluster2-sink
connector.class=org.apache.kafka.connect.sink.KafkaSinkConnector
tasks.max=2
# 要写入的集群2主题（和源主题同名）
topics=your-target-topic
# 集群2的Broker地址
bootstrap.servers=cluster2-host:9092
# 根据你的消息格式选择转换器，这里以JSON为例
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
# 若不需要Schema，关闭schema校验
key.converter.schemas.enable=false
value.converter.schemas.enable=false

3. 启动连接器 #

将上述配置文件转为JSON格式后，通过REST API提交：

# 提交Source连接器
curl -X POST -H "Content-Type: application/json" --data @cluster1-source.json http://connect-host:8083/connectors
# 提交Sink连接器
curl -X POST -H "Content-Type: application/json" --data @cluster2-sink.json http://connect-host:8083/connectors

关键注意事项 #

• 确保Kafka Connect节点能同时访问两个集群的Broker端口（9092或自定义端口）
• 如果消息使用Avro等格式，替换对应的Converter（如io.confluent.connect.avro.AvroConverter）
• 可根据集群规模调整tasks.max参数，提升复制吞吐量
• 测试时可生产几条带自定义头和不带头的消息，验证过滤逻辑是否生效

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