使用confluent-kafka-python实现Kafka含元数据消息跨队列转发

你的代码报错和发送失败主要有两个核心原因：

1. Producer.produce()方法的value参数仅支持字节、字符串等可直接序列化的原始类型，你直接传入confluent-kafka封装的Message对象，库无法识别该类型，自然无法发送
2. 尝试对Message对象做JSON序列化必然失败，它是C扩展封装的底层结构体，没有实现JSON序列化接口，且就算能序列化也会丢失原始消息的二进制信息，完全不需要做这一步操作

不需要序列化整个Message对象，直接调用Message自带的方法提取所有原生属性，再传入produce()方法即可完整保留所有元数据和消息内容，还可以额外追加死信场景的排查字段。

from confluent_kafka import Producer, Consumer

# 消费者初始化使用你原有配置即可，此处仅做示例
consumer = Consumer({
    'bootstrap.servers': 'host1:9092',
    'group.id': 'your-consumer-group',
    'auto.offset.reset': 'smallest'
})
consumer.subscribe(['your-business-topic'])

producer = Producer({'bootstrap.servers': "host1:9092",'client.id': 'dlq-producer'})

while True:
    msg = consumer.poll(timeout=1)
    if msg is None:
        continue
    # 触发死信的判断逻辑（消费报错/业务处理失败都可以走到死信分支）
    if msg.error():
        # 提取原消息所有原生属性
        original_value = msg.value()
        original_key = msg.key()
        # 处理原有headers，无headers则默认空列表
        original_headers = msg.headers() or []
        # 追加死信场景自定义元数据，方便后续排查
        dlq_headers = original_headers + [
            ('x-original-topic', msg.topic().encode('utf-8')),
            ('x-original-partition', str(msg.partition()).encode('utf-8')),
            ('x-original-offset', str(msg.offset()).encode('utf-8')),
            ('x-consume-error', str(msg.error()).encode('utf-8'))
        ]
        # 提取原消息时间戳
        _, ts_val = msg.timestamp()
        # 发送到死信队列
        try:
            producer.produce(
                topic='your-dlq-topic',
                value=original_value,
                key=original_key,
                headers=dlq_headers,
                timestamp=ts_val
            )
            # 触发消息发送，批量场景可以攒一批后统一调用flush()
            producer.poll(0)
        except Exception as e:
            print(f"死信消息发送失败: {str(e)}")
        continue
    
    # 正常业务消费逻辑写在此处
    # ...

• 原消息的value、key完全原样传递，没有任何转换损耗，不管原消息是二进制、JSON、纯文本格式都能完整保留
• 原生支持保留原消息的headers、时间戳，所有元数据无丢失
• 新增的自定义header字段可以直接在Kafka控制台或者消费死信队列时直接读取，不需要额外解析

• Kafka headers的所有值必须是bytes类型，字符串需要先做utf-8编码再传入
• 不要忘记调用producer.poll(0)或者定期执行producer.flush()，否则消息会缓存在客户端内存不会真正发送到集群
• 如果需要保留更多消费上下文，比如消费组ID、消费时间、业务错误信息等，都可以追加到headers里

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