Spring cloud kafka Stream Binder transactionIdPrefix向出站主题发送2条消息

Question

我们需要从一个主题中获取消息，然后进行一些扩展，然后将消息发布到另一个主题。以下是活动

1. 消费者-消费信息
2. 丰富-丰富了消费的信息
3. 制作人-将丰富的信息发布到其他主题

我正在使用Spring Cloud kafka绑定器，一切正常。最近我们引入了幂等生产者并包含transactionIdPrefix属性，我们观察到出站通道开始在主题中发送2条消息，因为它应该只发送一条消息。一条具有实际json值的消息另一条值为'b'\x00\x00\x00\x00\x06的消息

@StreamListener("INPUT")
@SendTo("OUTPUT")
public void consumer(Message message){
Acknowledgement ack = messge.getHeaders().get(KafkaHeaders.ACKNOWLEDGEMENT,Acknowledgement.class))
try{
    String inputMessage = message.getPayload.toString();
    String enrichMessage = // Enrichment on inputMessage
    ack.acknowledgement()   
    return enrichMessage;
}catch( Exception exp){
    ack.acknowledgement();
    throw exp;  
}
}

配置为

1. spring.cloud.stream.kafka.binder.transaction.transactionIdPrefix=TX-
2. spring.cloud.stream.kafka.binder.transaction.producer.configuration.ack=all
3. spring.cloud.stream.kafka.binder.transaction.producer.configuration.retries=10
4. spring.cloud.stream.kafka.bindings.input.consumer.auto提交偏移=false
5. spring.cloud.stream.kafka.bindings.input.consumer.enableDlq=true
6. spring.cloud.stream.kafka.bindings.input.consumer.dlqerror.topic
7. spring.cloud.stream.kafka.bindings.input.consumer.auto提交错误=true
8. spring.cloud.stream.kafka.bindings.input.consumer.max尝试=3
9. spring.cloud.stream.kafka.binder.transaction.producer.configuration.enable.idempotence=true

发送到出站主题的消息如下。

>

消费者记录（主题=测试，分区=1，偏移量=158，创建时间=1574297903694，时间戳=1238776543，时间戳类型=0，键=无值=b{“name”：“abc”，“age”：“20”}，校验和=无，序列化键大小=-1，序列化值大小=159）

用户记录（主题=测试，分区=1，偏移量=158，创建时间=1574297903694，时间戳=1238776543，time_stamp_type=0，键=b'\x00\x00\x00\x01'值=b'\x00\x00\x00\x06

即使在Dlq主题中，消息也会出现两次。

如果有人能就我们面临的这个问题提供任何建议，我们将不胜感激。

干杯

Answer 1

我相信你的代码运行良好。从技术上讲，事务生产者确实会发送两次消息——未提交的事务记录，然后是那些完全相同但标记为完整/已提交事务的记录。换句话说，您应该检查是否配置了消费者（从应用程序中的事务主题消费的消费者）隔离。将级别设置为读取提交的。

消费者记录-