Spring Kafka Streams中KTable配置自定义序列化器及异常处理

当前出现的ClassCastException，本质是Kafka Streams拓扑中触发shuffle的中间节点没有显式指定序列化器，默认继承了全局的String类型序列化规则，和实际Avro类型的value不匹配：

• 当KTable执行groupBy、reduce操作时，框架会自动生成两类中间主题：用于数据重平衡的repartition主题、用于状态恢复的changelog主题
• 仅给toTable操作指定自定义Avro Serdes，这部分配置仅对toTable自身生成的changelog主题生效，无法覆盖后续groupBy、reduce节点生成的repartition、聚合状态存储的序列化规则，最终导致Avro对象被错误传入StringSerializer触发类型转换异常。
• 之前误以为需要传入Bytes类型的KeyValueBytesStoreSupplier才能适配自定义Serdes，这是认知误区：只要在Materialized配置中传入正确的对象类型Serdes，框架会自动适配底层状态存储的序列化逻辑，不需要手动指定Bytes类型的存储实现。

当前使用的Spring Cloud Stream Kafka Binder编程模型，不需要手动实例化StreamsBuilder，直接在每个触发序列化的拓扑节点显式传入对应Serdes即可完成适配：

1. 给toTable节点保留已配置的Materialized.with(自定义key Serde, 自定义value Serde)，覆盖toTable自身changelog主题的序列化规则
2. 给groupBy操作补充Grouped参数，显式指定repartition主题的key、value序列化规则
3. 给reduce操作补充Materialized参数，显式指定聚合状态存储、对应changelog主题的序列化规则
4. 移除所有手动定义的KeyValueBytesStoreSupplier相关代码，避免覆盖框架自动生成的序列化适配逻辑

将原有拓扑中涉及KTable转换、聚合的代码替换为如下实现：

KTable<String, NotificationAvro> convertedTable = notificationAvroKStream
        // toTable节点：显式指定changelog主题的序列化规则
        .toTable(Materialized.with(CustomSerdes.String(), CustomSerdes.NotificationAvro()))
        // groupBy节点：显式指定repartition主题的序列化规则
        .groupBy(
                (cardNo, notificationAvro) -> KeyValue.pair(cardNo, notificationAvro),
                Grouped.with(CustomSerdes.String(), CustomSerdes.NotificationAvro())
        )
        // reduce节点：显式指定聚合状态、对应changelog主题的序列化规则
        .reduce(
                (aggValue, newValue) -> {
                    newValue.setTotalLoyaltyPoints(newValue.getEarnedLoyaltyPoints() + aggValue.getTotalLoyaltyPoints());
                    return newValue;
                },
                (aggValue, oldValue) -> oldValue,
                Materialized.with(CustomSerdes.String(), CustomSerdes.NotificationAvro())
        );

• 报错中提到的ChangedSerializer是Kafka Streams内部对KTable变更流的包装类，不需要手动配置：只要给对应节点指定了正确的业务类型Serde，框架会自动用内部的ChangeSerializer包装你的Avro Serde完成序列化，不会出现类型不兼容问题。
• 自定义的CustomSerdes实现没有问题，继承原生Serdes类后可直接复用内置的String Serde，无需额外重复实现字符串序列化逻辑。
• 不需要为了适配序列化去全局修改默认Serdes，逐节点指定的方式完全兼容现有其他拓扑的JSON、AVRO混合序列化需求。
• 完全不需要手动引入KeyValueBytesStoreSupplier：当在Materialized中传入业务类型的Serdes时，框架会自动将对象序列化后存入底层状态存储，手动传入Bytes类型的StoreSupplier反而会破坏自动适配逻辑，导致类型转换报错。

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