SCDF中Redis存储型Aggregator触发NotSerializableException及解码异常咨询

启用Redis消息存储的报错原因 #

• 配置app.aggregator.message-store-type=redis后，聚合器会切换为RedisMessageStore作为消息组暂存组件，该组件默认使用JdkSerializationRedisSerializer做序列化，要求待存储的完整Message（包含payload、所有Header字段）都必须实现Serializable接口。
• 从报错栈可以定位到具体失败点：传入聚合器的消息Header中携带kafka_consumer字段，字段值是DefaultKafkaConsumerFactory$1类的匿名实例，该实例没有实现Serializable接口。JDK序列化器在递归遍历MessageHeaders做对象序列化时，碰到这个不可序列化的运行时对象，就会抛出NotSerializableException，最终导致消息存储流程中断。
• 该kafka_consumer字段是Spring Cloud Stream Kafka绑定器默认注入的上下文信息，原本用于同进程内的消费偏移量管理，不满足远程存储的序列化要求。

未配置Redis存储时输出Base64的原因 #

• 不指定message-store-type时，聚合器默认使用内存版SimpleMessageStore暂存消息，不会触发序列化流程。观察到的输出["YQ==","Yg==","Yw=="]本质不是解码错误：
  • splitter拆分字符串"a,b,c"得到的三个单字符片段，在经过Kafka topic传输时，会被绑定器序列化为byte[]类型的payload
  • 配置的聚合表达式#this.![payload]会提取消息组内所有消息的payload组成列表，当列表元素是byte[]类型时，Jackson JSON序列化器默认会将字节数组转为Base64字符串输出，刚好对应a（YQ==）、b（Yg==）、c（Yw==）的Base64编码结果。

参考解决方向 #

• 修复Redis存储报错：可以给RedisMessageStore配置自定义序列化器（如StringRedisSerializer），仅序列化必要的业务字段，跳过不可序列化的运行时Header；也可以在splitter与aggregator之间增加消息头过滤逻辑，将kafka_consumer这类非业务的运行时Header移除后再传入聚合器。
• 修复Base64输出问题：在splitter节点配置类型转换，将拆分得到的byte[]直接转为String类型后再传输；也可以在聚合逻辑中增加解码步骤，将Base64格式的字节数组还原为原字符串后输出。

内容的提问来源于stack exchange，提问作者何子洋