使用SCDF与Kafka时，Aggregator输出Base64编码消息的原因及解决方法

问题原因 #

• 聚合器（aggregator processor 2021.1.x）搭配Redis存储时，默认将消息payload以字节数组形式序列化存储，未利用拆分消息携带的contentType: application/json头进行反序列化处理。
• 聚合完成释放消息时，存储的字节数组被直接转为Base64字符串，最终输出Base64字符串列表，而非还原为原始JSON对象组成的JSON数组。

解决方案 #

通过以下配置让聚合器正确处理JSON格式消息，确保存储和输出阶段都完成JSON序列化/反序列化：

1. 补充聚合器部署属性 #

在SCDF部署配置中添加以下内容：

# 指定输入输出的内容类型为JSON
app.aggregator.spring.cloud.stream.bindings.input.content-type=application/json
app.aggregator.spring.cloud.stream.bindings.output.content-type=application/json

# 配置消息存储使用JSON格式序列化
app.aggregator.aggregator.message-store.content-type=application/json

# 配置Kafka序列化/反序列化器，保证JSON消息正确解析
app.aggregator.spring.cloud.stream.kafka.binder.configuration.value.serializer=org.springframework.kafka.support.serializer.JsonSerializer
app.aggregator.spring.cloud.stream.kafka.binder.configuration.value.deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
app.aggregator.spring.cloud.stream.kafka.binder.configuration.spring.json.trusted.packages=*

2. 验证修复效果 #

重新部署数据流应用后，再次发送测试请求：

http post --target http://<ip-http-source>:8181 --file data/test.json --contentType "application/json; charset=utf-8"

此时聚合后的消息应输出预期的JSON数组：

[
  {"author": "Nigel Rees", "title": "Sayings of the Century"},
  {"author": "Evelyn Waugh", "title": "Sword of Honour"},
  {"author": "Herman Melville", "title": "Moby Dick"},
  {"author": "J. R. R. Tolkien", "title": "The Lord of the Rings"}
]

额外检查项 #

若问题仍存在，可排查：

• 通过Redis客户端确认存储的消息是否已按JSON格式序列化；
• 确认聚合器版本（2021.1.x）与SCDF 2.9.6的兼容性（二者兼容）；
• 再次验证拆分后的消息contentType头是否为application/json（已通过Kowl验证可跳过）。

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