使用Azure Event Hubs Stream时max-attempts重试配置不生效求助

1. 消费者逻辑判断错误导致异常未触发 #

你的消费者代码中直接用Message<String>对象与字符串"a"做equals比较，这是对象引用层面的对比，永远不会相等，所以异常根本不会被抛出。修正逻辑为比较消息的payload内容：

@Bean
public Consumer<Message<String>> consumer() {
    return message -> {
        String payload = message.getPayload();
        if ("a".equals(payload)) {
            throw new RuntimeException("run time exception");
        }
    };
}

2. 重试配置层级错误 #

当前配置缺少retry节点，spring-cloud-azure-stream-binder-eventhubs的重试参数需要嵌套在consumer.retry下，正确配置如下：

spring:
  stream:
    function:
      definition: consumer
    bindings:
      consumer-in-0:
        destination: test-eventhub
        group: $Default
        consumer:
          retry:
            max-attempts: 3
            initial-interval: 1000ms # 可选，首次重试间隔
            multiplier: 2.0 # 可选，重试间隔倍数
            max-interval: 5000ms # 可选，最大重试间隔
      supply-out-0:
        destination: test-eventhub

3. 全局错误处理器拦截重试流程 #

你通过@ServiceActivator(inputChannel = "errorChannel")直接捕获所有异常，会导致Spring Cloud Stream的重试机制无法触发。如果需要保留错误处理逻辑，建议使用Spring Cloud Stream的自定义重试恢复器替代直接监听errorChannel：

@Bean
public Consumer<Message<String>> consumer() {
    return message -> {
        String payload = message.getPayload();
        if ("a".equals(payload)) {
            throw new RuntimeException("run time exception");
        }
    };
}

// 重试耗尽后触发的恢复器
@Bean
public Consumer<Message<?>> consumerErrorRecoverer() {
    return failedMessage -> {
        log.error("重试耗尽，处理失败消息: {}", failedMessage.getPayload());
        // 此处可添加失败后的业务逻辑，比如投递死信队列
    };
}

同时在配置中指定恢复器：

spring:
  stream:
    function:
      definition: consumer;consumerErrorRecoverer
    bindings:
      consumer-in-0:
        destination: test-eventhub
        group: $Default
        consumer:
          retry:
            max-attempts: 3
            recoverer: consumerErrorRecoverer # 指定恢复器函数名
      supply-out-0:
        destination: test-eventhub

4. 版本兼容性检查 #

确保spring-cloud-azure-stream-binder-eventhubs版本与spring-cloud-stream版本兼容，不同版本的配置项可能存在差异。例如Spring Cloud 2023.x对应的Azure Binder版本应为5.x系列。

完成以上调整后，当消息payload为"a"时，重试机制会正常触发，重试3次后进入恢复器处理。

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