Step 2:
假设 Flink Kafka consumer 从分区 0 开始读取数据 “A”，那么此时第一个 consumer 的位点从 0 变成 1。如下图所示。
Step 3:
此时数据 “A” 到达 Flink Job 中的 Map Task。两个 consumer 继续读取数据 (从分区 0 读取数据 “B” ，从分区 1 读取数据 “A”)。 offsets 分别被更新成 2 和 1。与此同时，假设 Flink 从 source 端开始执行 checkpoint。
Step 4:
到这里，Flink Kafka consumer tasks 已经执行了一次快照，offsets也保存到了 state 中(“offset = 2, 1”) 。此时 source tasks 在 数据 “B” 和 “A” 后面，向下游发送一个 checkpoint barrier。checkpoint barriers 是 Flink 用来对齐每个任务算子的 checkpoint，以确保整个 checkpoint 的一致性。分区 1 的数据 “A” 到达 Flink Map Task， 与此同时分区 0 的 consumer 继续读取下一个消息(message “C”)。
Step 5:
Flink Map Task 收到上游两个 source 的 checkpoint barriers 然后开始执行 checkpoint ，把 state 保存到 filesystem。同时，消费者继续从Kafka分区读取更多事件。
Step 6:
假设 Flink Map Task 是 Flink Job 的最末端，那么当它完成 checkpoint 后，就会立马通知 Flink Job Master。当 job 的所有 task 都确认其 state 已经 “checkpointed”，Job Master将完成这次的整个 checkpoint。 之后，checkpoint 可以用于故障恢复。
故障恢复
如果发生故障（例如，worker 挂掉），则所有任务将重启，并且它们的状态将被重置为最近一次的 checkpoint 的状态。 如下图所示。
source 任务将分别从 offset 2 和 1 开始消费。当任务重启完成， 将会正常运行，就像之前没发生故障一样。
PS: 文中提到的 checkpoint 对齐，我说下我的理解，假设一个 Flink Job 有 Source -> Map -> Sink，其中Sink有多个输入。那么当一次checkpoint的 barrier从source发出时，到sink这里，多个输入需要等待其它的输入的barrier已经到达，经过对齐后，sink才会继续处理消息。这里就是exactly-once和at-least-once的区别。
The End
原文链接: How Apache Flink manages Kafka consumer offsets
