如果 KGroupedStream returns 的聚合器为空会发生什么？

Question

KStream<Integer, Integer> stream;
KGroupedStream<Integer, Integer> grouped = stream.groupByKey();
KTable<Integer, Integer> aggregated = grouped.aggregate(
    () -> 0,
    (k, i, agg) -> {
       if (agg == null)
         agg = 0;
       Integer sum = agg + i;
       return sum > 100 ? null : sum;
    });

我的流中的消息是：

(1, 50)
(1, 75)
(1, 50)

当第二条消息到达聚合器时 returns 为空。 KTable aggregated 是否接收到 (1, null) 并删除其 key=1 的状态？

当 message#3 到达时 agg 为 null 或再次调用 Initializer 并将 agg 设置为 0？

如果我使用 reduce 而不是 aggregate，如果 Reducer returns null 下一条消息会通过 Reducer 还是会被使用 'as is' 就像它是组中的第一条消息一样？

谢谢，大卫

Answer 1

When the second message arrives the Aggregator returns null. Does KTable aggregated receive (1, null) and deletes its state for key=1?

是的。

When message#3 arrives is agg null or is the Initializer called again and sets agg to 0?

再次调用初始化程序。

What if I use reduce instead of aggregate, if the Reducer returns null will the next message go through the Reducer or will it be used 'as is' like it was the first message in the group?

Reduce 像聚合一样工作。因此，如果您 return null，则以下消息将被视为第一条消息。

元评论：你为什么不直接运行代码并尝试一下？？？

如果 KGroupedStream returns 的聚合器为空会发生什么？

What happens if the aggregator of a KGroupedStream returns null?

apache-kafka

apache-kafka-streams