在Kafka流中，没有所谓的“最终聚集”。窗口始终保持打开状态，以处理在窗口结束时间过后到达的无序记录。然而，Windows并不是永远保存的。一旦它们的保留时间过期，它们就会被丢弃。对于窗口何时被丢弃，没有特殊的操作。
有关更多详细信息，请参阅合流文档：http://docs.confluent.io/current/streams/
因此，对于聚合的每次更新，都会生成一个结果记录（因为kafka流也会更新无序记录的聚合结果）。您的“最终结果”将是最新的结果记录（在窗口被丢弃之前）。根据您的用例，手动重复数据消除将是解决问题的一种方法（使用较低级别的api， transform() 或者 process() )
这篇博文可能也会有所帮助：https://timothyrenner.github.io/engineering/2016/08/11/kafka-streams-not-looking-at-facebook.html
另一篇不使用标点符号的博客文章：http://blog.inovatrend.com/2018/03/making-of-message-gateway-with-kafka.html
更新
使用kip-328，a KTable#suppress() 添加了操作符，允许严格抑制连续更新，并在每个窗口中发出单个结果记录；折衷的办法是增加延迟。

我面临这个问题，但我解决了这个问题，在固定窗口之后添加grace（0），并使用抑制的api

public void process(KStream<SensorKeyDTO, SensorDataDTO> stream) {

        buildAggregateMetricsBySensor(stream)
                .to(outputTopic, Produced.with(String(), new SensorAggregateMetricsSerde()));

    }

private KStream<String, SensorAggregateMetricsDTO> buildAggregateMetricsBySensor(KStream<SensorKeyDTO, SensorDataDTO> stream) {
        return stream
                .map((key, val) -> new KeyValue<>(val.getId(), val))
                .groupByKey(Grouped.with(String(), new SensorDataSerde()))
                .windowedBy(TimeWindows.of(Duration.ofMinutes(WINDOW_SIZE_IN_MINUTES)).grace(Duration.ofMillis(0)))
                .aggregate(SensorAggregateMetricsDTO::new,
                        (String k, SensorDataDTO v, SensorAggregateMetricsDTO va) -> aggregateData(v, va),
                        buildWindowPersistentStore())
                .suppress(Suppressed.untilWindowCloses(unbounded()))
                .toStream()
                .map((key, value) -> KeyValue.pair(key.key(), value));
    }

    private Materialized<String, SensorAggregateMetricsDTO, WindowStore<Bytes, byte[]>> buildWindowPersistentStore() {
        return Materialized
                .<String, SensorAggregateMetricsDTO, WindowStore<Bytes, byte[]>>as(WINDOW_STORE_NAME)
                .withKeySerde(String())
                .withValueSerde(new SensorAggregateMetricsSerde());
    }

在这里你可以看到结果

从kafka streams 2.1版开始，您可以使用 suppress .
上述apache kafka streams文档中有一个示例，当用户一小时内发生的事件少于三个时，它会发送警报：

KGroupedStream<UserId, Event> grouped = ...;
grouped
  .windowedBy(TimeWindows.of(Duration.ofHours(1)).grace(ofMinutes(10)))
  .count()
  .suppress(Suppressed.untilWindowCloses(unbounded()))
  .filter((windowedUserId, count) -> count < 3)
  .toStream()
  .foreach((windowedUserId, count) -> sendAlert(windowedUserId.window(), windowedUserId.key(), count));

正如这个答案的更新中所提到的，您应该知道这种权衡。此外，请注意suppress（）是基于事件时间的。