本文为翻译flink作者之一Fabian Hueske的文章
原文：https://www.ververica.com/blog/how-apache-flink-manages-kafka-consumer-offsets
在Flink Friday Tip这集中，我们通过一步步的样例解释了Apache Flink是如何协作Apache Kafka保障从Kafka Topics里处理并保证exactly-once。
Checkpointing是Apache Flink用来故障恢复的内部机制。一次checkpoint是一次Flink应用state的统一拷贝并且包括输入端的读取位置。在一次失败case中，Flink恢复应用通过加载应用checkpoint的state并且从恢复的位置继续读取就像什么都没有发生一样。你可以理解checkpoint为电脑游戏中的保存当前游戏进度。如果在你保存游戏位置后发生了些事情，你可以回退到过去并且重试。
Checkpoints让Apache Flink有了容错并且确保了流处理应用在发生失败时的语义。Checkpoints是可以通过配置定时触发的。
Flink 中的Kafka Consumer被整合进了Flink的checkpointing机制，像一个存储了所有Kafka Partitions的offset状态的状态算子。当checkpoint触发，每个partition的offsets被存储到checkpoint。FLink的checkpoint机制确保存储的全部算子的任务是统一的，即它们基于同样的输入源数据。一次checkpoint当所有的算子任务成功存储状态后记作是完成。于是，系统在从潜在的系统失败重启时，提供了exactly-once状态保障。
下面我们通过一步步的指示描述了Apache Flink 如何处理checkpoints 在Kafka consumer offsets场景。在我们的例子里，数据存储在FLink job Master。要注意的是这里是在POC或生产用例，数据通常存储在外部文件系统比如HDFS或者S3。
Step1
下面例子从Kafka topic 两个partitions读取数据，每个partition包含A，B，C，D，E消息。我们设置每个partitions的offset为0。
Step2
第二步，kafka consumer开始读取partition 0消息。消息A在飞行(in-flight)被处理并且第一个consumer offset改为1.
Step3
第三步，消息A到达Map任务。每个consumer读取各自的下个记录（在partition0中是消息B，在partition1中是消息A）。offsets分别被更新为2和1在两个partition。在同时，Flink的Job Master判定触发一次checkpoint在源。
Step4
接下来阶段，kafka consumer的任务已经创建了一个属于它们状态的快照（offset=2,1）在Job Master中。源头发送一次checkpoint barrier（不懂可以google下）用来对齐所有算子任务和保证总体checkpoint的一致性。消息A到达map任务时头部consmer继续读取下条消息（消息C）
Step5
这个阶段展示了flink map任务从两个源接收barrier，并且checkpoint它的状态到Job Master。与此同时，consumer继续从kafka partition读取更多事件。
Step6
这个阶段展示Map任务一旦checkpoint state完成后与Job Master的交互。当job的所有任务checkpoint完成和响应时，Job Master完成checkpoint。从现在开始，checkpoint可以用来故障恢复。值得一提的是Apache FLink 不依赖kafka的offsets机制来恢复潜在的系统错误
Recovery in case of a failure
在失败例子中（例如，一个worker失败）所有的算子任务被重启并且它们的状态被重置到最后一次checkpoint。像下图描述的一样。
Kafka源开始消费分别在2和1的位点，在已完成checkpoint的offset。当job重启时我们期望一个普通系统就像没有错误发生一样。
