启用Kafka事务后，Spring Cloud Stream消费者可重试异常如何配置？

问题背景 #

在Spring Cloud Stream中结合Kafka事务、死信队列配置可重试异常时，一旦通过spring.cloud.stream.kafka.binder.transaction.transaction-id-prefix启用事务，default-retryable和retryable-exceptions配置会被忽略。即使配置了default-retryable: false并指定IllegalArgumentException为不可重试，消息仍会被重试5次；但关闭事务后，配置能正常生效。

复现示例 #

消费者代码 #

@Configuration
@Slf4j
public class ConsumerConfig {
    @Bean
    public Consumer<String> consumeMessage() {
        return s -> {
            log.info("Consuming {}", s);
            throw new IllegalArgumentException(s);
        };
    }
}

配置文件 application.yml #

spring:
  cloud:
    function:
      definition: consumeMessage
    stream:
      kafka:
        binder:
          transaction:
            transaction-id-prefix: transaction-
          required-acks: all
          configuration:
            key.serializer: org.apache.kafka.common.serialization.StringSerializer
            key.deserializer: org.apache.kafka.common.serialization.StringDeserializer
        bindings:
          consumeMessage-in-0:
            consumer:
              enable-dlq: true
      bindings:
        consumeMessage-in-0:
          group: my-group
          destination: my-topic
          consumer:
            default-retryable: false
            max-attempts: 5
            back-off-initial-interval: 100
            retryable-exceptions:
              java.lang.UnsupportedOperationException: true
              java.lang.IllegalArgumentException: false

依赖配置 #

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-kafka-streams</artifactId>
</dependency>
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <scope>provided</scope>
</dependency>

问题原因 #

查看KafkaMessageChannelBinder源码可知：

• 未启用事务时，Binder会通过buildRetryTemplate(properties)构建包含重试异常规则的RetryTemplate；
• 启用事务后，Binder改用AfterRollbackProcessor处理重试，但仅传入退避策略（BackOff），未复用default-retryable和retryable-exceptions的配置，导致规则失效。

解决方案 #

要在事务模式下实现自定义重试异常规则，需手动配置AfterRollbackProcessor并注入重试异常判断逻辑：

1. 自定义AfterRollbackProcessor #

@Configuration
public class KafkaTransactionRetryConfig {

    @Bean
    public AfterRollbackProcessor<String, String> customAfterRollbackProcessor(
            KafkaBindingProperties bindingProperties,
            ObjectProvider<DeadLetterPublishingRecoverer> deadLetterPublishingRecoverer) {

        // 获取目标消费者的配置参数
        ConsumerProperties consumerProps = bindingProperties.getBindings()
                .get("consumeMessage-in-0")
                .getConsumer();

        // 构建异常分类器，复用配置中的重试规则
        BinaryExceptionClassifier exceptionClassifier = BinaryExceptionClassifier.builder()
                .defaultValue(consumerProps.getDefaultRetryable())
                .classifiedExceptions(consumerProps.getRetryableExceptions())
                .build();

        // 创建带异常规则的AfterRollbackProcessor
        DefaultAfterRollbackProcessor<String, String> processor = new DefaultAfterRollbackProcessor<>(
                deadLetterPublishingRecoverer.getIfAvailable(),
                consumerProps.getBackOff(),
                exceptionClassifier
        );

        // 设置最大重试次数
        processor.setMaxAttempts(consumerProps.getMaxAttempts());
        return processor;
    }
}

2. 绑定自定义处理器到消费者容器 #

通过容器工厂配置，将自定义处理器关联到目标消费者：

@Configuration
public class KafkaConsumerContainerConfig {

    @Autowired
    private AfterRollbackProcessor<String, String> customAfterRollbackProcessor;

    @Bean
    public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            ConsumerFactory<Object, Object> kafkaConsumerFactory) {

        ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        configurer.configure(factory, kafkaConsumerFactory);
        // 替换默认处理器为自定义实现
        factory.getContainerProperties().setAfterRollbackProcessor(customAfterRollbackProcessor);
        return factory;
    }
}

3. 验证效果 #

重启应用后发送消息，IllegalArgumentException会在一次消费失败后直接进入死信队列，不再重试；UnsupportedOperationException会按配置重试指定次数后进入死信队列，符合预期规则。

注意事项 #

• 多消费者绑定场景下，需为每个绑定单独配置对应的AfterRollbackProcessor，或通过动态逻辑适配多个绑定；
• 不同Spring Cloud版本的AfterRollbackProcessor构造参数可能存在差异，需根据实际版本调整代码；
• 确保DeadLetterPublishingRecoverer已正确配置（即enable-dlq: true），否则重试耗尽后消息会被丢弃。

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