为什么我的Kafka Streams应用程序的消费者组（app-id）的偏移量在应用程序重启后会被重置？

Question

我有一个Kafka Streams应用程序，每当我重新启动它时，它所消耗的主题的偏移量就会被重置。因此，对于所有分区，延迟增加，应用程序需要重新处理所有数据。

更新：输出主题接收到一系列事件，这些事件在应用程序重新启动后已经被处理，而不是像我在上一段中所说的那样，输入主题的偏移量被重置。但是，内部主题（KTABLE-SUPPRESS-STATE-STORE）偏移量正在重置，请参见下面的注释。

在重新启动之前，我已经确保每个分区的延迟为1（这是针对输出主题的）。属于该消费者组ID（app-id）的所有消费者都是活动的。重启是立即的，大约需要30秒。

我已经读过这个答案，对于一个Apache Kafka消费者群体，偏移是如何过期的？.

我尝试了auto.offset.reset=最新和auto.offset.reset=最早。

似乎这些主题的补偿没有有效地提交，（但我不确定这一点）。

Kafka流API是否确保在关闭之前提交所有消耗的偏移量？（在调用streams.close（）之后）

我会非常感激任何关于这件事的线索。

更新：

final StreamsBuilder builder = new StreamsBuilder();
final KStream<..., ...> events = builder
        .stream(inputTopicNames, Consumed.with(..., ...)
        .withTimestampExtractor(...);

events
    .filter((k, v) -> ...)
    .flatMapValues(v -> ...)
    .flatMapValues(v -> ...)
    .selectKey((k, v) -> v)
    .groupByKey(Grouped.with(..., ...))
    .windowedBy(
        TimeWindows.of(Duration.ofSeconds(windowSizeInSecs))              
            .advanceBy(Duration.ofSeconds(windowSizeInSecs))
            .grace(Duration.ofSeconds(windowSizeGraceInSecs)))
    .reduce((agg, new) -> {
        ...
        return agg;
    })
    .suppress(Suppressed.untilWindowCloses(
                  Suppressed.BufferConfig.unbounded()))
    .toStream()
    .to(outPutTopicNameOfGroupedData, Produced.with(..., ...));

再一次，我会非常感激任何关于这一点的线索。

更新：这个问题已经在2.2.1版中解决了(https://issues.apache.org/jira/browse/kafka-7895)

Answer 1

偏移量重置只是而且总是（在重新启动之后）使用Kafka Stream API创建的KTABLE-SUPPRESS-STATE-STORE内部主题进行。

这是当前（版本2.1）的预期行为，因为suppress()操作符仅在内存中工作。因此，在重新启动时，必须从changelog主题重新创建禁止缓冲区，然后才能开始处理。

注意，计划在未来的版本中允许suppress()写入磁盘（参见https://issues.apache.org/jira/browse/kafka-7224)。这将避免从changelog主题重新创建缓冲区的开销。