Flink 在应用程序重启/错误修复时是否考虑 Kafka 偏移？

Question

我在 Kafka-Flink 应用程序中遇到了一个代码错误，我需要重新部署整个 Flink 应用程序。我知道检查点和保存点，但是由于我的应用程序必须重新启动，它将丢失所有 flink 的内部检查点，并且我没有创建任何保存点。

但是，我的

commit.offsets.on.checkpoint

标志是true，所以即使我的内部flink检查点丢失，kafka代理也会有最后提交的偏移量。

因此，在这种情况下，我重新启动的 flink 应用程序会从该 kafka 代理的偏移量中读取，还是会从 kafka 队列中一直重新启动，因为它丢失了所有内部检查点？

Answer 1

如果 Flink 从检查点或保存点重新启动，则 Flink 将使用该快照中的偏移量。否则它将使用 KafkaSource 或 FlinkKafkaConsumer 中配置的起始位置。如果您没有明确配置起始位置，那么 KafkaSource 将从最早的偏移量开始，并且我不确定 FlinkKafkaConsumer 的默认值是什么。

如果您想从检查点或保存点重新启动，同时忽略已保存的偏移量（以便获取已保存状态的其余部分），则更改 Kafka 源运算符上的 UID，配置特定的起始位置，并设置

 --allowNonRestoredState

旗帜。