我有Kafka流应用程序如下：

static KafkaStreams build(AppConfig appConfig, SerdesHelper serdes) {
  final KStreamBuilder builder = new KStreamBuilder();

  builder
      .table(serdes.sourceKeySerde, serdes.sourceValueSerde, appConfig.sourceTopic)
      .groupBy(StreamBuilder::groupByMapper, serdes.intSerde, serdes.longSerde)
      .aggregate(
          StreamBuilder::initialize,
          StreamBuilder::add,
          StreamBuilder::subtract,
          serdes.sinkValueSerde)
      .to(serdes.intSerde, serdes.sinkValueSerde, appConfig.sinkTopic);

  return new KafkaStreams(builder, appConfig.streamConfig);
}

我的具体例子如下

((k, v)) -> ((k), v[])

在使用只有两个唯一键的3.000.000条消息的虚拟数据运行时，我在 sinkTopic 在不到一分钟，我希望得到4/6（根据当时我设法停止应用程序）。
如何确保只有具有最新分组值的密钥才会提交回kafka，而不是每个中间消息？