为什么我没有看到任何来自Kafka Streams reduce方法的输出？

Question

给定以下代码：

KStream<String, Custom> stream =  
    builder.stream(Serdes.String(), customSerde, "test_in");

stream
    .groupByKey(Serdes.String(), customSerde)
    .reduce(new CustomReducer(), "reduction_state")
    .print(Serdes.String(), customSerde);

在reduce语句之后我需要什么才能看到reduce的结果？如果一个值被推到输入，我不希望看到任何东西。如果推送具有相同键的第二个值，我希望还原器应用（它确实应用了），并且我还希望还原的结果继续到处理管道中的下一个步骤。正如所描述的，我没有在管道的后续步骤中看到任何东西，我不明白为什么。

Answer 1

从Kafka0.10.1.0起，所有聚合运算符都使用内部去重复缓存来减少结果KTable changelog流的负载。例如，如果您对两个具有相同键的记录进行计数和处理，则完整的changelog流将为 、 。

使用新的缓存功能，缓存将接收 并存储它，但不会立即向下游发送它。计算 时，它将替换缓存的第一个条目。根据缓存大小、不同键数、吞吐量和提交间隔，缓存向下游发送条目。这种情况要么发生在单个键项的缓存驱逐时，要么发生在缓存的完全刷新时（向下游发送所有项）。因此，KTable变更日志可能只显示 （因为 已删除重复）。

您可以通过流配置参数streamconfig.cache_max_bytes_buffering_config控制缓存的大小。如果将该值设置为零，则完全禁用缓存，KTable更改日志将包含所有更新（有效地提供pre0.10.1.0行为）。

Confluent文档中有一节更详细地解释了缓存：

• http://docs.confluent.io/current/streams/architecture.html#记录-缓存
• http://docs.confluent.io/current/streams/developer-guide.html#streams-developer-guide-memory-management