Spring Cloud Stream Kafka Streams反序列化异常崩溃 无法跳过坏记录如何解决

你配置的deserializationExceptionHandler（反序列化异常处理器）仅会在Kafka Serde执行反序列化操作的阶段生效，也就是deserialize方法主动抛出异常时才会被触发。
你当前遇到的ClassCastException不属于反序列化阶段异常：你的CotsEventSerde已经成功将消息反序列化为per.shades.model.events.CotsEvent对象，但绑定接口中声明的输入流泛型为KStream<String, StatsMetadata>，Spring Cloud Stream在将反序列化结果传递给你的业务处理器时发生了类型强制转换失败，这个异常属于业务处理阶段的异常，不会被反序列化异常处理器捕获，因此你的配置和自定义处理器都不会触发，最终导致Kafka Streams线程全部退出，应用关闭。

方案1：修正类型声明，手动过滤不匹配消息 #

如果你确认Topic中的消息本身就是CotsEvent类型，只是部分消息可以转换为StatsMetadata，直接修改绑定接口和处理器的泛型声明，在业务逻辑中手动做类型校验过滤：

1. 修改绑定接口

@Configuration
public interface StatsStreamBindings
{
    String statsInput = "statsInput";

    @Input(statsInput)
    KStream<String, CotsEvent> statsInput();
}

2. 修改处理器逻辑，添加类型过滤

public void aggregateStats(KStream<String, CotsEvent> inputStream) {
    inputStream
        // 过滤类型不匹配的消息
        .filter((key, value) -> {
            if (value instanceof StatsMetadata) {
                return true;
            }
            // 此处可添加日志打印、手动写入死信队列的逻辑
            log.warn("跳过类型不匹配的消息，实际类型：{}，期望类型：StatsMetadata", value.getClass().getName());
            return false;
        })
        // 转换为目标类型
        .mapValues(value -> (StatsMetadata) value)
        // 后续继续执行原有的业务聚合逻辑
        // ...
}

方案2：自定义Serde，将类型不匹配转为反序列化异常 #

如果需要保留原有绑定接口的StatsMetadata泛型声明，同时希望复用现有的反序列化异常处理器和DLQ配置，可以自定义Serde在反序列化阶段就做类型校验，不匹配直接抛出反序列化异常：

1. 自定义Serde

public class StatsMetadataSerde implements Serde<StatsMetadata> {
    private final CotsEventSerde delegateSerde = new CotsEventSerde();

    @Override
    public Deserializer<StatsMetadata> deserializer() {
        return (topic, data) -> {
            Object deserializedObj = delegateSerde.deserializer().deserialize(topic, data);
            if (deserializedObj instanceof StatsMetadata) {
                return (StatsMetadata) deserializedObj;
            }
            // 类型不匹配直接抛出反序列化异常，触发后续的logAndContinue和DLQ逻辑
            throw new SerializationException(String.format("消息类型不匹配，期望StatsMetadata，实际为%s", deserializedObj.getClass().getName()));
        };
    }

    @Override
    public Serializer<StatsMetadata> serializer() {
        // 如果不需要序列化逻辑可空实现，或根据业务需求实现
        return (topic, data) -> delegateSerde.serializer().serialize(topic, data);
    }

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        delegateSerde.configure(configs, isKey);
    }

    @Override
    public void close() {
        delegateSerde.close();
    }
}

2. 修改配置文件中的valueSerde配置为自定义Serde

spring:
  cloud:
    stream:
      kafka:
        streams:
          bindings:
            statsInput:
              consumer:
                valueSerde: 你的包路径.StatsMetadataSerde

修改完成后，类型不匹配的消息会在反序列化阶段抛出异常，你已经配置的deserializationExceptionHandler: logAndContinue和dlqName都会自动生效，异常会被打印，坏消息会自动写入死信队列，应用不会退出。

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