仅当在流上接收到新事件时才抑制触发事件

Question

我正在使用 Kafka 流 2.2.1。

我正在使用抑制来阻止事件，直到窗口关闭。我正在使用事件时间语义。 但是，触发消息仅在流上有新消息可用时才会触发。

提取以下代码对问题进行示例：

        KStream<UUID, String>[] branches = is
            .branch((key, msg) -> "a".equalsIgnoreCase(msg.split(",")[1]),
                    (key, msg) -> "b".equalsIgnoreCase(msg.split(",")[1]),
                    (key, value) -> true);

    KStream<UUID, String> sideA = branches[0];
    KStream<UUID, String> sideB = branches[1];

    KStream<Windowed<UUID>, String> sideASuppressed =
            sideA.groupByKey(
                    Grouped.with(new MyUUIDSerde(),
                    Serdes.String()))
            .windowedBy(TimeWindows.of(Duration.ofMinutes(31)).grace(Duration.ofMinutes(32)))
            .reduce((v1, v2) -> {
                return v1;
            })
            .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
            .toStream();

消息仅在新消息到达“sideA”流时从“sideASuppressed”流式传输（到达“sideB”的消息不会导致抑制发出任何消息，即使窗口关闭时间已经过去很久了）。 尽管在生产中由于容量很大，问题可能不会发生太多，但在很多情况下，不必等待进入“sideA”流的新消息。

提前致谢。

Answer 1

根据 Kafka 流文档：

只有在所有输入主题的所有输入分区都有可用的新数据（具有更新的时间戳）时，流时间才会提前。如果至少一个分区没有任何新数据可用，则流时间不会提前，因此如果指定了 PunctuationType.STREAM_TIME，则不会触发 punctuate()。此行为与配置的时间戳提取器无关，即使用 WallclockTimestampExtractor 不会启用 punctuate() 的挂钟触发。

我不确定为什么会出现这种情况，但是，它解释了为什么仅当消息在它使用的队列中可用时才发出被抑制的消息。

如果有人对为什么会这样实施有答案，我将很乐意学习。这种行为导致我的实现发出消息只是为了让我的抑制消息及时发出，并导致代码的可读性大大降低。