在Bash中实现Kafka控制台消费者输出实时转发至控制台生产者

你之前的方法失效，核心原因是Kafka控制台消费者默认输出会附带分区、偏移量等元数据，而生产者只能解析纯消息内容，直接管道会导致生产者把元数据当成消息的一部分，甚至无法正确处理。另外< <(producer)的写法逻辑搞反了——它是把生产者的输出作为消费者的输入，和你想要的方向完全相反。

以下是可行的解决方案：

1. 确保消费者只输出纯消息内容 #

使用--property参数屏蔽所有额外元数据，只输出消息的key或value（根据你的需求调整）：

仅转发消息Value的命令 #

./kafka-console-consumer.sh \
  --bootstrap-server <你的Kafka Broker地址> \
  --topic <源主题名> \
  --group <自定义消费者组名> \
  --property print.key=false \
  --property print.value=true \
  --property print.offset=false \
  --property print.partition=false \
  --property print.timestamp=false \
| ./kafka-console-producer.sh \
  --bootstrap-server <你的Kafka Broker地址> \
  --topic <目标主题名>

需转发消息Key+Value的命令 #

如果要保留key-value结构，需要指定分隔符，让生产者能正确识别：

./kafka-console-consumer.sh \
  --bootstrap-server <你的Kafka Broker地址> \
  --topic <源主题名> \
  --group <自定义消费者组名> \
  --property print.key=true \
  --property print.value=true \
  --property key.separator=, \  # 自定义key和value的分隔符，确保生产者能解析
  --property print.offset=false \
  --property print.partition=false \
  --property print.timestamp=false \
| ./kafka-console-producer.sh \
  --bootstrap-server <你的Kafka Broker地址> \
  --topic <目标主题名> \
  --property parse.key=true \
  --property key.separator=,  # 和消费者用相同的分隔符

2. 避免重复消费的关键配置 #

• 必须指定--group <自定义消费者组名>：Kafka会自动记录该组的消费偏移量，重启脚本后不会重复消费已处理过的消息
• 如果需要从头开始消费源主题，在消费者命令后追加--from-beginning，但注意：只有当该消费者组从未消费过目标主题时才会生效，否则需要先重置偏移量

3. 处理特殊消息场景 #

如果消息包含换行符、非UTF-8编码或二进制内容，需要调整参数确保传输完整：

• 消费者端：追加--property value.separator=（避免默认的分隔符干扰）
• 生产者端：使用二进制行读取器，命令追加--line-reader org.apache.kafka.common.utils.ByteBufferLineReader

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