SCDF 2.11处理器：事务生产者+手动ACK消费者配置异常排查

我正在使用SCDF 2.11，尝试实现无消息丢失的处理器。要求生产者具备事务性，消费者仅在处理成功时，或错误发生且消息已进入DLQ时才进行ACK。

以下是我的配置：

# Producer config
spring.cloud.stream.kafka.binder.transaction.transaction-id-prefix=my-tx-
spring.cloud.stream.kafka.binder.transaction.producer.configuration.acks=all
spring.cloud.stream.kafka.binder.transaction.producer.configuration.retries=5
spring.cloud.stream.kafka.binder.transaction.producer.configuration.max.block.ms=5000
spring.cloud.stream.kafka.binder.transaction.producer.configuration.delivery.timeout.ms=4500
spring.cloud.stream.kafka.binder.transaction.producer.configuration.request.timeout.ms=2000
spring.cloud.stream.kafka.binder.transaction.producer.configuration.linger.ms=0
spring.cloud.stream.kafka.binder.transaction.producer.configuration.batch.size=0

# Consumer config
spring.cloud.stream.kafka.binder.consumer.isolation.level=read_committed
spring.cloud.stream.kafka.bindings.input.consumer.ackMode=MANUAL_IMMEDIATE
spring.cloud.stream.kafka.bindings.input.consumer.enableDlq=true
spring.cloud.stream.bindings.input.group=mygroup

但当处理器出错时，消费偏移量的ACK操作失败，导致消息被重复消费，最终DLQ中出现大量重复消息。以下是堆栈信息片段：

o.s.k.t.KafkaTransactionManager          : Participating in existing transaction
o.s.c.s.b.k.KafkaMessageChannelBinder    : Sent to DLQ  a message with key='mykey' and payload='byte[1281]' received from 0: mytopic@152
o.s.t.support.TransactionTemplate        : Initiating transaction rollback on application exception

java.lang.NullPointerException: Cannot invoke "org.springframework.kafka.support.Acknowledgment.acknowledge()" because the return value of "org.springframework.messaging.MessageHeaders.get(Object, java.lang.Class)" is null
    at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder$DlqSender.sendToDlq(KafkaMessageChannelBinder.java:1639) ~[spring-cloud-stream-binder-kafka-4.1.5.jar:4.1.5]
    at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.lambda$handleRecordForDlq$10(KafkaMessageChannelBinder.java:1273) ~[spring-cloud-stream-binder-kafka-4.1.5.jar:4.1.5]
    at org.springframework.transaction.support.TransactionOperations.lambda$executeWithoutResult$0(TransactionOperations.java:68) ~[spring-tx-6.1.16.jar:6.1.16]
    at org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:140) ~[spring-tx-6.1.16.jar:6.1.16]
    at org.springframework.transaction.support.TransactionOperations.executeWithoutResult(TransactionOperations.java:67) ~[spring-tx-6.1.16.jar:6.1.16]
    at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.handleRecordForDlq(KafkaMessageChannelBinder.java:1272) ~[spring-cloud-stream-binder-kafka-4.1.5.jar:4.1.5]
    at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.lambda$getErrorMessageHandler$6(KafkaMessageChannelBinder.java:1136) ~[spring-cloud-stream-binder-kafka-4.1.5.jar:

经过排查，发现消息缺失kafka_acknowledgment头。此外，我使用以下代码创建输出消息，本应不会丢失头信息：

MessageBuilder.withPayload(object).copyHeaders(inMsg.getHeaders());

请问我的配置中缺少了什么？

这个问题的核心是事务性生产者配置会触发Spring Cloud Stream的头过滤机制，导致kafka_acknowledgment这类内部控制头被自动移除，即便调用copyHeaders也无法保留。

关键原因 #

当启用spring.cloud.stream.kafka.binder.transaction.transaction-id-prefix时，Binder会默认启用头过滤逻辑，仅保留符合Kafka协议的标准头（如kafka_messageKey），而kafka_acknowledgment属于Spring Cloud Stream的内部控制头，会被过滤丢弃。当消息出错需要转发到DLQ时，Binder找不到这个头就会抛出NPE，进而触发事务回滚、消费偏移量未提交，最终导致消息重复消费。

修复步骤 #

1. 禁用内部头过滤
   添加配置强制Binder保留所有头信息，包括内部控制头：

   spring.cloud.stream.kafka.binder.headers=*
   

2. 修正消息构建逻辑
   确认消息构建时完整生成消息实例，避免遗漏头设置：

   MessageBuilder.withPayload(object).copyHeaders(inMsg.getHeaders()).build();
   

   注意必须调用build()方法完成消息构建，否则头信息不会被正确绑定到消息实例。

3. 确保DLQ操作参与事务
   由于使用了事务性生产者，需让DLQ的发送操作也纳入事务管理，避免出现"消息已发往DLQ但事务回滚"的不一致情况：

   spring.cloud.stream.kafka.bindings.input.consumer.dlq-producer-transacted=true
   

验证效果 #

修复完成后，当处理器出错时：

• 消息会被正确发送到DLQ，且事务正常提交
• kafka_acknowledgment头存在，Binder能正常执行ACK操作
• 消费偏移量被提交，不会出现重复消费问题

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