Spring Cloud Stream Kafka Binder：批量模式下结合DLQ时重试失效问题

环境信息 #

• Spring Cloud版本：2023.0.1
• Spring Cloud Stream版本：4.1.1

批量消费者代码 #

@Bean
Consumer<Message<List<String>>> consumer1() {
    return message -> {
        final List<String> payload = message.getPayload();
        final MessageHeaders messageHeaders = message.getHeaders();
        payload.forEach(System.out::println);
        payload.forEach(p -> {
            if(p.startsWith("a")) {
                throw new RuntimeException("Intentional Exception");
            }
        });
        System.out.println(messageHeaders);
        System.out.println("Done");
    };
}

application.yml配置 #

spring:
  cloud:
    function:
      definition: consumer1;
    stream:
      bindings:
        consumer1-in-0:
          destination: topic1
          group: consumer1-in-0-v0.1
          consumer:
            batch-mode: true
            use-native-decoding: true
            max-attempts: 3
      kafka:
        binder:
          brokers:
            - localhost:9092
        default:
          consumer:
            configuration:
              max.poll.records: 1000
              max.partition.fetch.bytes: 31457280
              fetch.max.wait.ms: 200
        bindings:
          consumer1-in-0:
            consumer:
              enableDlq: true
              dlqName: dlq-topic
              dlqProducerProperties:
                configuration:
                  value.serializer: org.apache.kafka.common.serialization.StringSerializer
                  key.serializer: org.apache.kafka.common.serialization.StringSerializer
              configuration:
                key.deserializer: org.apache.kafka.common.serialization.StringDeserializer
                value.deserializer: org.apache.kafka.common.serialization.StringDeserializer

自定义重试配置类 #

@Bean
ListenerContainerWithDlqAndRetryCustomizer cust(KafkaTemplate<?, ?> template) {
    return new ListenerContainerWithDlqAndRetryCustomizer() {

        @Override
        public void configure(AbstractMessageListenerContainer<?, ?> container, String destinationName,
                              String group,
                              @Nullable BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> dlqDestinationResolver,
                              @Nullable BackOff backOff) {

            ConsumerRecordRecoverer dlpr = new DeadLetterPublishingRecoverer(template,
                    dlqDestinationResolver);
            container.setCommonErrorHandler(new DefaultErrorHandler(dlpr, backOff));
        }

        @Override
        public boolean retryAndDlqInBinding(String destinationName, String group) {
            return false;
        }

    };
}

问题现象 #

当发生错误时，消息批次直接进入DLQ，没有进行任何重试。由于可能存在导致批次处理失败的瞬时错误，希望批次先重试几次再进入DLQ，但无法实现该功能，请问哪里配置错误？

核心问题 #

1. ListenerContainerWithDlqAndRetryCustomizer中retryAndDlqInBinding返回false，导致Spring Cloud Stream跳过绑定层面的重试配置，绑定里的max-attempts:3完全失效。
2. configure方法中传入的backOff参数为null，直接创建的DefaultErrorHandler没有重试策略，会直接将失败批次转入DLQ。

修正步骤 #

1. 调整自定义重试配置类 #

修改retryAndDlqInBinding返回true，让绑定的重试配置生效；同时如果backOff为null，手动指定退避策略控制重试次数：

@Bean
ListenerContainerWithDlqAndRetryCustomizer cust(KafkaTemplate<?, ?> template) {
    return new ListenerContainerWithDlqAndRetryCustomizer() {

        @Override
        public void configure(AbstractMessageListenerContainer<?, ?> container, String destinationName,
                              String group,
                              @Nullable BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> dlqDestinationResolver,
                              @Nullable BackOff backOff) {
            // 手动指定退避策略：重试2次，每次间隔1秒（加上初始尝试共3次）
            if (backOff == null) {
                backOff = new FixedBackOff(1000L, 2L);
            }
            ConsumerRecordRecoverer dlpr = new DeadLetterPublishingRecoverer(template,
                    dlqDestinationResolver);
            container.setCommonErrorHandler(new DefaultErrorHandler(dlpr, backOff));
        }

        @Override
        public boolean retryAndDlqInBinding(String destinationName, String group) {
            return true; // 允许绑定层面的重试配置生效
        }

    };
}

2. 批量模式重试逻辑说明 #

批量消费模式下，DefaultErrorHandler默认将整个批次视为一个单元重试。如果需要跳过单条失败消息继续处理批次其他内容，可以配置BatchErrorHandler，但如果需求是整个批次重试后再进入DLQ，上述配置即可满足。

3. 验证绑定配置的生效 #

当retryAndDlqInBinding返回true时，绑定中的max-attempts:3会自动生成对应的退避策略，此时可以不用手动创建BackOff，Spring Cloud Stream会自动处理重试次数。

额外注意事项 #

• 确认use-native-decoding: true符合业务需求，批量模式下该配置会将整个批次的ConsumerRecord转换为List<String>，配置错误可能导致消息解析异常。
• DLQ的生产者序列化配置要和原消息格式匹配，确保消息能正确写入DLQ。

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