Spring Kafka Stream应用如何跳过非法消息避免AVRO序列化异常

错误根因 #

你遇到的序列化NPE是两个原因共同导致的：

• Avro序列化器默认不允许序列化含null值的对象（除非你在AVRO Schema中显式将对应字段定义为["null", "原类型"]的联合类型）
• 你当前的逻辑是把无效消息映射为null AVRO对象后继续向下游传递，而不是直接将这类消息从流中剔除

推荐方案1：使用flatMap操作直接过滤无效消息 #

这是最符合Kafka Stream设计规范的实现方式，直接将无效消息从流中移除，不会向下游传递，也不会触发后续序列化逻辑：

@StreamListener("Processor-input-channel")
@SendTo("Processor-output-channel")
public KStream<String, AVROClass> process(KStream<String, String> input){
    KStream<String, AVROClass> kStream = input
        // 用flatMapValues替代mapValues，无效消息返回空列表，有效消息返回单元素列表
        .flatMapValues(v -> {
            AVROClass avro = service.getAVROResponse(v);
            return avro != null ? Collections.singletonList(avro) : Collections.emptyList();
        })
        .map((k,v) -> KeyValue.pair(Long.toString(System.currentTimeMillis()), v));

    kStream.foreach((k, v) -> log.info("Key: {}, Value: {}", k, v));
    
    return kStream;
}

如果是较新版本的Kafka Stream，也可以直接用filter操作：在map之后加.filter((k, v) -> v != null)即可过滤掉null值消息。

兜底方案2：配置生产异常处理器避免线程崩溃 #

如果确实存在部分合法消息也可能出现Avro序列化失败的场景，可以配置Kafka Stream的生产异常处理器，遇到序列化错误时直接跳过该消息，记录日志后继续处理后续消息，不会导致流线程终止：
在application.yml中添加如下配置：

spring:
  cloud:
    stream:
      kafka:
        streams:
          binder:
            configuration:
              default:
                production.exception.handler.class: org.apache.kafka.streams.errors.DefaultProductionExceptionHandler
              # 也可以自定义异常处理器实现更复杂的逻辑，比如将错误消息发到死信队列
              # production.exception.handler.class: com.yourpackage.CustomAvroExceptionHandler

自定义异常处理器示例：

public class CustomAvroExceptionHandler implements ProductionExceptionHandler {
    private static final Logger log = LoggerFactory.getLogger(CustomAvroExceptionHandler.class);

    @Override
    public ProductionExceptionHandlerResponse handle(ProducerRecord<byte[], byte[]> record, Exception exception) {
        log.error("序列化消息失败，跳过该消息，topic:{}, 异常信息:{}", record.topic(), exception.getMessage());
        // 返回CONTINUE表示继续处理后续消息，FAIL表示终止线程
        return ProductionExceptionHandlerResponse.CONTINUE;
    }

    @Override
    public void configure(Map<String, ?> configs) {
        // 初始化配置逻辑
    }
}

额外建议 #

如果需要留存无效消息方便后续排查，可以在过滤无效消息的同时，将这类消息发送到专门的死信Topic，避免丢失原始数据。

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