Spring Cloud Stream Kafka中如何启用deliveryAttemptHeader获取重试次数

我正在使用spring-cloud-stream-binder-kafka，并通过DefaultErrorHandler实现了有状态重试。了解到启用容器属性的deliveryAttemptHeader后，可从消息头中获取重试次数，但无法成功启用该配置。

我尝试过以下代码：

@Bean
public ContainerCustomizer<String, Message, ConcurrentMessageListenerContainer<String, Message>> containerCustomizer(
    ConcurrentKafkaListenerContainerFactory<String, Message> factory) {
  ContainerCustomizer<String, Message, ConcurrentMessageListenerContainer<String, Message>> custCustomizer = container -> {
    container.getContainerProperties().setDeliveryAttemptHeader(true);
  };
  factory.setContainerCustomizer(custCustomizer);
  return custCustomizer;
}

启动应用后调试KafkaMessageListenerContainer.java#L1078时，发现deliveryAttemptHeader仍处于禁用状态，且我创建的ContainerCustomizer实例也未被调用。

原因分析 #

Spring Cloud Stream Kafka binder不会直接使用你自定义的ConcurrentKafkaListenerContainerFactory，而是通过自身的配置链路创建容器实例，所以直接给该工厂设置ContainerCustomizer不会被binder识别和应用。

正确解决方案 #

方案1：使用KafkaBinderCustomizer定制容器工厂 #

通过实现KafkaBinderCustomizer介入binder的容器工厂创建流程，确保配置被应用：

@Bean
public KafkaBinderCustomizer kafkaBinderCustomizer() {
    return (binderName, configuration, containerFactory) -> {
        if (containerFactory instanceof ConcurrentKafkaListenerContainerFactory) {
            ConcurrentKafkaListenerContainerFactory<?, ?> factory = 
                (ConcurrentKafkaListenerContainerFactory<?, ?>) containerFactory;
            factory.setContainerCustomizer(container -> 
                container.getContainerProperties().setDeliveryAttemptHeader(true)
            );
        }
    };
}

方案2：直接通过配置文件启用 #

无需编写代码，直接在application.yml中添加配置：

spring:
  cloud:
    stream:
      kafka:
        binder:
          configuration:
            deliveryAttemptHeader: true

如果使用application.properties格式：

spring.cloud.stream.kafka.binder.configuration.deliveryAttemptHeader=true

方案3：使用ListenerContainerCustomizer（Spring Boot 2.4+） #

若你的项目基于Spring Boot 2.4及以上版本，可以直接定义ListenerContainerCustomizer Bean，它会自动作用于所有Kafka监听容器：

@Bean
public ListenerContainerCustomizer<ConcurrentMessageListenerContainer<String, Message>> listenerContainerCustomizer() {
    return (container, destination, group) -> 
        container.getContainerProperties().setDeliveryAttemptHeader(true);
}

配置完成后重启应用，再次调试KafkaMessageListenerContainer即可看到deliveryAttemptHeader已启用。重试时，可通过message.getHeaders().get("kafka_deliveryAttempt")获取当前的重试次数。

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