使用Spring Cloud Stream模拟Azure事件中心生产者错误与路由配置

基于你当前使用的Kafka协议对接模式，生产者侧会触发错误的场景主要分为以下几类：

• 认证与权限错误：连接字符串配置错误、身份凭证过期、当前身份无目标Event Hub的写入权限，这类错误无法通过重试恢复，一般在客户端建立连接或首次发送时就会抛出
• 网络与服务可用性错误：网络链路中断、防火墙拦截访问、Event Hub服务端临时故障、请求超时（你当前配置的request.timeout.ms=60000，超过该时间未收到服务端ACK就会触发超时错误），这类错误可通过客户端重试尝试恢复，重试耗尽后抛出
• 服务端限流与配额错误：发送吞吐量超过Event Hub配置的吞吐量单位（TU）阈值、短时间发送请求过多触发服务端限流，这类错误退避等待后可能恢复，重试耗尽后抛出
• 消息合法性错误：单条消息大小超过Event Hub限制（标准层单条消息最大1MB，批次消息总大小最大1MB）、消息格式符合服务端校验规则，这类错误重试无效，会直接抛出
• 配置与资源错误：目标Event Hub实例不存在（你当前配置autoCreateTopics=false，不会自动创建资源）、acks=all配置下ISR副本同步数量不满足要求、生产者客户端配置与服务端要求不匹配，这类错误部分可重试，部分需要修改配置后才能恢复。

你当前配置sync=false为异步发送模式，默认异步发送的异常不会直接透传到业务调用线程，需要通过Spring Cloud Stream Kafka binder自带的错误通道机制捕获，配置步骤如下：

1. 开启绑定级别的错误通道配置，在现有生产者配置下新增error-channel-enabled: true，配置参考：

server.port: 9876
spring:
  cloud:
    stream:
      bindings:
        output:
          destination: test
      kafka:
        bindings:
          output:
            producer:
              retries: 3
              sync: false
              # 开启生产者错误通道
              error-channel-enabled: true
        binder:
          configuration:
            request:
              timeout:
                ms: 60000
            security:
              protocol: SASL_SSL
            sasl:
              mechanism: PLAIN
              jaas:
                config: org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="<Pwd>";
          brokers: <Broker URL>
          autoCreateTopics: false
          producer-properties:
            acks: all

2. 编写错误通道监听逻辑，Spring Cloud Stream会为每个开启错误通道的生产者binding自动创建名为{bindingName}.errors的错误通道，你当前使用的binding名为output，对应错误通道为output.errors，直接编写监听方法即可：

import lombok.extern.slf4j.Slf4j;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class EventHubProduceErrorHandler {

    @ServiceActivator(inputChannel = "output.errors")
    public void handleSendError(Message<MessagingException> errorMsg) {
        MessagingException exception = errorMsg.getPayload();
        // 获取发送失败的原始消息内容
        Object failedPayload = exception.getFailedMessage().getPayload();
        // 获取根因异常
        Throwable rootCause = exception.getCause();
        
        log.error("Event Hub消息发送失败，原始消息：{}，错误信息：{}", failedPayload, rootCause.getMessage(), rootCause);
        // 此处可扩展自定义降级逻辑：比如持久化失败消息待重试、触发告警、记录审计日志等
    }
}

• 你配置的retries=3是Kafka客户端原生层面的重试，只有3次重试全部失败后，异常才会被投递到错误通道，重试过程中的临时异常不会触发错误通道逻辑
• 如果将sync改为true使用同步发送模式，异常会直接在业务调用线程抛出，可直接通过try-catch捕获，同时异常依然会被投递到错误通道，两种捕获方式互不冲突
• 如果需要统一处理所有生产者binding的发送错误，可直接监听全局内置的errorChannel通道，无需为每个binding单独编写监听，不过按binding单独监听更方便区分不同业务链路的错误
• 由于Event Hub的Kafka兼容层完全适配原生Kafka协议，上述错误通道逻辑无需额外引入Event Hub专属依赖，基于现有Kafka binder即可正常运行。

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