Spring Cloud Stream函数错误处理器正确实现方法咨询（2021.0.3版本）

环境信息 #

• Spring Cloud 版本：2021.0.3
• Spring Boot 版本：2.7.2

问题现象 #

按照官方文档实现的自定义Consumer<ErrorMessage>错误处理器未生效，消息触发3次重试后由默认处理器记录日志并丢弃；改用@ServiceActivator绑定错误通道的方式虽能触发，但无法正确获取原始消息，出现类型转换异常。

已尝试的实现及配置 #

1. 初始函数式错误处理器实现 #

@Bean
public Function<Message<MessageDto>, Message<MessageDto>> process() {
  return streamMessage -> {
    LOG.info("received message: {}", streamMessage.getPayload());

    final MessageDto message = streamMessage.getPayload();
    final MessageHeaders headers = streamMessage.getHeaders();

    throw new InvalidJsonException(); // 自定义RuntimeException模拟业务错误

    LOG.info("send message: {}", message);

    return MessageBuilder
        .withPayload(message)
        .copyHeadersIfAbsent(headers)
        .build();
  };
}

@Bean
public Consumer<ErrorMessage> myErrorHandler() {
  return v -> {
    LOG.error("houston we got a problem: {}", v);
  };
}

2. 对应配置文件 #

spring:
  cloud:
    stream:
      default:
        error-handler-definition: myErrorHandler
      function:
        definition: process
      bindings:
        process-in-0:
          destination: inbound
          binder: kafka1
        process-out-0:
          destination: outbound
          binder: kafka1
      binders:
        kafka1:
          type: kafka
          environment:
            spring:
              cloud:
                stream:
                  kafka:
                    binder:
                      brokers: kafka:9092
                      zkNodes: localhost:2181
                      replicationFactor: 1
                      requiredAcks: -1
                      configuration:
                        max.poll.interval.ms: 259200000 # 3d
                        max.poll.records: 1

3. 日志输出 #

received message: MessageDto{text='Hello World'}
received message: MessageDto{text='Hello World'}
received message: MessageDto{text='Hello World'}
o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessageHandlingException: error occurred in message handler [org.springframework.cloud.stream.function.FunctionConfiguration$FunctionToDestinationBinder$1@610d37e1]; nested exception is InvalidJsonException
o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessageHandlingException: error occurred in message handler [org.springframework.cloud.stream.function.FunctionConfiguration$FunctionToDestinationBinder$1@610d37e1]; nested exception is InvalidJsonException
o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessageHandlingException: error occurred in message handler [org.springframework.cloud.stream.function.FunctionConfiguration$FunctionToDestinationBinder$1@610d37e1]; nested exception is InvalidJsonException

4. 尝试的@ServiceActivator实现（存在类型转换问题） #

@SuppressWarnings("unchecked")
@ServiceActivator(inputChannel = ERROR_CHANNEL_BEAN_NAME)
public void handleError(final ErrorMessage error) {
  if (error.getOriginalMessage() == null) {
    LOG.error("original stream message is null!");
    return;
  }
  
  final Message<MessageDto> streamMessage = (Message<MessageDto>) error.getOriginalMessage();
  final MessageDto message = streamMessage.getPayload();
  final MessageHeaders headers = error.getHeaders();
  final Throwable e = error.getPayload();
  
  if (e instanceof InvalidJsonException) {
    LOG.error("We got our invalid json exception");
  }
}

正确实现方案 #

方案一：修复函数式错误处理器不生效问题 #

在Spring Cloud Stream 2021.0.x版本中，函数式模型下的错误处理器配置需调整：

1. 修改配置文件，将错误处理器指定从default.error-handler-definition改为function.error-handler：

spring:
  cloud:
    stream:
      function:
        definition: process
        error-handler: myErrorHandler # 指定自定义错误处理器的bean名称
      bindings:
        process-in-0:
          destination: inbound
          binder: kafka1
          consumer:
            max-attempts: 1 # 可选：关闭重试，直接进入错误处理器（默认重试3次）
        # 其他绑定配置...

2. 保持Consumer<ErrorMessage>的bean定义，可优化为支持原始消息解析：

@Bean
public Consumer<ErrorMessage> myErrorHandler() {
  return errorMessage -> {
    LOG.error("houston we got a problem: {}", errorMessage);
    // 安全获取原始消息
    Message<?> originalMsg = errorMessage.getOriginalMessage();
    if (originalMsg != null) {
      Object payload = originalMsg.getPayload();
      if (payload instanceof MessageDto) {
        LOG.error("Original message payload: {}", payload);
      }
    }
    // 处理自定义异常
    Throwable cause = errorMessage.getPayload();
    if (cause instanceof InvalidJsonException) {
      LOG.error("Handled InvalidJsonException", cause);
    }
  };
}

方案二：修复@ServiceActivator的类型转换问题 #

通过类型判断避免强制转换失败，同时支持Kafka绑定器的序列化场景：

import org.springframework.messaging.support.ErrorMessage;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.Message;
import com.fasterxml.jackson.databind.ObjectMapper;

@ServiceActivator(inputChannel = "errorChannel")
public void handleError(ErrorMessage errorMessage) {
  Throwable throwable = errorMessage.getPayload();
  Message<?> originalMessage = errorMessage.getOriginalMessage();
  
  if (originalMessage != null) {
    Object payload = originalMessage.getPayload();
    // 处理不同类型的payload
    if (payload instanceof MessageDto) {
      LOG.error("Original message: {}", payload);
    } else if (payload instanceof byte[]) {
      // Kafka绑定器下可能是字节数组，需反序列化
      try {
        MessageDto messageDto = new ObjectMapper().readValue((byte[]) payload, MessageDto.class);
        LOG.error("Original message (deserialized): {}", messageDto);
      } catch (Exception e) {
        LOG.error("Failed to deserialize payload", e);
      }
    } else {
      LOG.error("Unexpected payload type: {}", payload.getClass().getName());
    }
  }
  
  if (throwable instanceof InvalidJsonException) {
    LOG.error("Handled InvalidJsonException", throwable);
  } else {
    LOG.error("Unhandled error", throwable);
  }
}

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