Flink offset 在手动重置 kafka offset 时进入不一致状态

Question

我们有一个 flink 流应用程序，它从 kafka 读取消息。由于某种原因，我们不得不通过 kafka reset 命令将 kafka 偏移量重置为最新，因为有大量堆积。我们希望 flink 应用程序跳过所有这些消息并从重置后出现的新消息开始。

问题是因为 flink 在内部管理它的偏移量，所以它不知道这个重置，它现在只从后向读取消息（重置前的偏移量点），现在也不能提交偏移量。因此，每次重新启动 flink 应用程序时，它都会再次从同一点读取。所以我们在每次重启时都有重复的消息。

我明白我们不应该在 flink kafka 应用程序中手动重置偏移量。但是我们如何从中恢复。

我已经尝试将 auto.offset.config 设置为最新，但它仍然会再次读取这些消息。

Answer 1

只有当 Flink 从故障中恢复或从保存点或检查点手动重启时，才会使用检查点或保存点中记录的偏移量。

否则，Flink Kafka 消费者将从消费者组在 Kafka 代理中提交的偏移量开始读取，或者从您在代码中明确指定的偏移量开始读取，即

myConsumer.setStartFromEarliest();     // start from the earliest record possible
myConsumer.setStartFromLatest();       // start from the latest record
myConsumer.setStartFromTimestamp(...); // start from specified epoch timestamp (msecs)
myConsumer.setStartFromGroupOffsets(); // the default behaviour

我不确定如何将这些事实与您报告的内容相协调。

Flink offset 在手动重置 kafka offset 时进入不一致状态

Flink offset went to inconsistent state on manually resetting kafka offset

apache-kafka

flink-streaming