【发布时间】：2019-06-04 02:37:12
【问题描述】：
我有一个 Kafka Streams 应用程序，每当我重新启动它时，它正在使用的主题的偏移量都会被重置。因此，对于所有分区，延迟都会增加，应用需要重新处理所有数据。
更新： 输出主题正在接收应用程序重新启动后已经处理的突发事件，而不是像我在上一段中所说的那样重置输入主题偏移量。但是，内部主题 (KTABLE-SUPPRESS-STATE-STORE) 偏移量正在重置，请参阅下面的 cmets。
我已确保重启前每个分区的延迟为 1（这是针对输出主题的）。 属于该消费者组 ID (app-id) 的所有消费者都处于活动状态。 立即重启，大约需要 30 秒。
应用只使用一次作为处理保证。
我已阅读此答案How does an offset expire for an Apache Kafka consumer group?。
我已尝试使用 auto.offset.reset = latest 和 auto.offset.reset = early。
这些主题的偏移量似乎没有有效地提交，（但我不确定）。
我假设在重新启动后，应用程序应该从该消费者组的最新提交偏移量中获取。
更新： 我假设这是内部主题（KTABLE-SUPPRESS-STATE-STORE）
Kafka Stream 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(..., ...));
使用 Kafka Stream API 创建的 KTABLE-SUPPRESS-STATE-STORE 内部主题总是会发生偏移重置（重启后）。
我已尝试仅一次和至少一次处理保证。
再一次，我将非常感谢有关此的任何线索。
更新： 这已在版本 2.2.1 (https://issues.apache.org/jira/browse/KAFKA-7895)
中解决【问题讨论】：
-
你说你不确定偏移量是否真的被提交了。您是否查看过 consumer__offsets 主题来验证这一点？这篇博文可能有助于找出答案：medium.com/@felipedutratine/…。
-
您是否尝试过它是否可以在没有完全一次处理保证的情况下工作？这是一个相对较新的功能。
-
感谢@user152468 的评论，是的，我已经使用
bin/kafka-consumer-groups.sh工具进行了检查。我没有尝试过其他处理保证，因为在我的情况下我需要一个，但值得尝试只是放弃这可能是问题。但是，如果是的话，我会感到惊讶。 -
偏移量应该在
streams.close()上提交——尝试验证这一点。此外，如果启用了完全一次，则默认情况下应在 100 毫秒内提交所有偏移量。不知道为什么在启动时没有再次获取偏移量。我建议检查日志——也许将日志级别提高到 DEBUG 以获取更多信息。 -
您好@MatthiasJ.Sax 感谢您的回复，我在日志中找到了这个：INFO [MI-APP-ID-xxx-StreamThread-4] internals.StoreChangelogReader (StoreChangelogReader.java:215) - 流线程 [MI-APP-ID-xxx-StreamThread-4] 未找到检查点 用于任务 1_5 状态存储 KTABLE-SUPPRESS-STATE-STORE-0000000011 更改日志 MI-APP-ID-KTABLE- SUPPRESS-STATE-STORE-0000000011-changelog-5 开启 EOS。 重新初始化任务并从头恢复其状态。 ...关注下一条评论
标签： apache-kafka apache-kafka-streams