spring-boot - Spring Cloud Stream：Headers kafka_acknowledgement 值为 null

Question

我试图根据应用程序中消息的处理状态来控制 kafka 主题的偏移量提交。如果消息成功，则可以提交偏移量。为此，我试图在我的方法中获取标题，如果可以手动确认该消息

  spring:
   cloud:
    stream:
      default:
        contentType: application/json
      default-binder: binder1-kafka
      bindings:
        myChannel:
          binder: binder1-kafka
          destination: my_topic
          content-type: text/plain
          consumer:
            autoCommitOffset: false

        outChannel:
          binder: binder2-kafka
          destination: my_topic
          content-type: text/plain
          consumer:
            autoCommitOffset: false    


      binders:
        #Connection config to different clusters
        binder1-kafka:
          type: kafka
          defaultCandidate: true
          environment:
            spring:
              cloud:
                stream:
                  kafka:
                    binder:
                      brokers: some-url1:9092
        binder2-kafka:
          type: kafka
          defaultCandidate: false
          environment:
            spring:
              cloud:
                stream:
                  kafka:
                    binder:
                      brokers: some-url2:9092 

但是当使用监听器作为

@StreamListener(target = IBrokerChannel.myChannel )
public void handlePayload(@Payload MyPayload payload, @Headers Map<String, Object> headers) {
  Acknowledgment acknowledgment= (Acknowledgment) headers.get("kafka_acknowledgment"); // acknowledgment object is always null.
  acknowledgment.acknowledge();
}

acknowledgment总是null。我正在使用 kafka 生产者 cli 将消息发送到主题。spring-boot版本是1.5.10.RELEASE

标签： spring-bootspring-integrationspring-cloud-streamspring-kafka

Answer 1

您的配置属性定义中缺少kafka分支。它必须是这样的：

spring:
   cloud:
    stream:
      default-binder: kafka
      kafka:
         bindings:
           myChannel:
             consumer:
               autoCommitOffset: false 

https://docs.spring.io/spring-cloud-stream/docs/Fishtown.BUILD-SNAPSHOT/reference/htmlsingle/#kafka-consumer-properties