SCDF 聚合器仅发出最后一条记录答案

【问题标题】：SCDF Aggregator Emits only the last recordSCDF 聚合器仅发出最后一条记录
【发布时间】：2020-05-28 16:14:00
【问题描述】：

我目前正在尝试一种用例，我们必须在将多个输入消息写入文件系统之前将其聚合为一个。为此，我正在尝试使用此处提供的聚合器模块 https://repo.spring.io/release/org/springframework/cloud/stream/app/aggregator-processor-kafka/2.1.1.RELEASE/

我的环境是在 docker Dataflow 2.4.1.Release 和 Skipper 2.3.1.Release 上运行的本地 SCDF/Kafka

流定义简单如下

:inputTopic > 聚合器 > :outputTopic

在聚合器属性中提供这些属性值。

release: size() == 5
group-timeout: 5000
correlation: headers.headerName
aggregation: T(org.springframework.util.StringUtils).collectionToDelimitedString(#this.![new String(payload)],'')
message-store-type: simple

我有两个问题

虽然我给出了 5 的发布大小，如果有 10 消息，我没有看到我的输出主题收到两条消息
输出主题仅包含聚合标头的最后一条消息，不包含此列表的串联值。

请建议是否有任何其他配置可以帮助实现聚合输出。

另外一个观察是，如果我有一个流

http | aggregator > :topic

相同的聚合器配置可以正常工作并且可以根据需要进行。但是我们拥有它的方式是，数据通过使用 KStreams 的处理器放入主题中，并且不确定是否有一种方法可以使用这些默认启动应用程序生成标头并将其读取到自定义处理器。

在调试聚合器期间的另一个观察结果是，correlationKey 是通过 hashCode 而不是作为字符串获取或比较的，不确定它是错误还是功能。但这正是导致预期行为中断的原因。

【问题讨论】：

标签： spring-cloud-stream spring-cloud-dataflow

【解决方案1】：

对于任何寻找答案的人来说，我要做的就是将相关性更改为

new String[headers.headerName]

并将输入值序列化器显式配置为bytearrayserializer

但是 Spring 团队提出的问题是，如果我不必触摸聚合器，是否可以在 2.2.4 引导开发的 Kstream 应用程序的输出生成器属性中设置一个属性，以将标头作为字符串而不是字节数组？

我使用以下 API 来设置标题

context.headers().add(key, value.getBytes())

我试过了

producer.header-mode: to none/headers.

但是我的聚合器仍然没有聚合并且只给出最后一条记录。只有在更改聚合器属性后它才起作用。但是我想避免更改聚合器属性，因为该流不归我们所有，并且旧版应用程序适用于其他旧版流。

【讨论】：