本文為翻譯flink作者之一Fabian Hueske的文章
原文：https://www.ververica.com/blog/how-apache-flink-manages-kafka-consumer-offsets
在Flink Friday Tip這集中，我們通過一步步的樣例解釋了Apache Flink是如何協(xié)作Apache Kafka保障從Kafka Topics里處理并保證exactly-once。
Checkpointing是Apache Flink用來故障恢復(fù)的內(nèi)部機(jī)制。一次checkpoint是一次Flink應(yīng)用state的統(tǒng)一拷貝并且包括輸入端的讀取位置。在一次失敗case中，F(xiàn)link恢復(fù)應(yīng)用通過加載應(yīng)用checkpoint的state并且從恢復(fù)的位置繼續(xù)讀取就像什么都沒有發(fā)生一樣。你可以理解checkpoint為電腦游戲中的保存當(dāng)前游戲進(jìn)度。如果在你保存游戲位置后發(fā)生了些事情，你可以回退到過去并且重試。
Checkpoints讓Apache Flink有了容錯并且確保了流處理應(yīng)用在發(fā)生失敗時的語義。Checkpoints是可以通過配置定時觸發(fā)的。
Flink 中的Kafka Consumer被整合進(jìn)了Flink的checkpointing機(jī)制，像一個存儲了所有Kafka Partitions的offset狀態(tài)的狀態(tài)算子。當(dāng)checkpoint觸發(fā)，每個partition的offsets被存儲到checkpoint。FLink的checkpoint機(jī)制確保存儲的全部算子的任務(wù)是統(tǒng)一的，即它們基于同樣的輸入源數(shù)據(jù)。一次checkpoint當(dāng)所有的算子任務(wù)成功存儲狀態(tài)后記作是完成。于是，系統(tǒng)在從潛在的系統(tǒng)失敗重啟時，提供了exactly-once狀態(tài)保障。
下面我們通過一步步的指示描述了Apache Flink 如何處理checkpoints 在Kafka consumer offsets場景。在我們的例子里，數(shù)據(jù)存儲在FLink job Master。要注意的是這里是在POC或生產(chǎn)用例，數(shù)據(jù)通常存儲在外部文件系統(tǒng)比如HDFS或者S3。
Step1
下面例子從Kafka topic 兩個partitions讀取數(shù)據(jù)，每個partition包含A，B，C，D，E消息。我們設(shè)置每個partitions的offset為0。
Step2
第二步，kafka consumer開始讀取partition 0消息。消息A在飛行(in-flight)被處理并且第一個consumer offset改為1.
Step3
第三步，消息A到達(dá)Map任務(wù)。每個consumer讀取各自的下個記錄（在partition0中是消息B，在partition1中是消息A）。offsets分別被更新為2和1在兩個partition。在同時，F(xiàn)link的Job Master判定觸發(fā)一次checkpoint在源。
Step4
接下來階段，kafka consumer的任務(wù)已經(jīng)創(chuàng)建了一個屬于它們狀態(tài)的快照（offset=2,1）在Job Master中。源頭發(fā)送一次checkpoint barrier（不懂可以google下）用來對齊所有算子任務(wù)和保證總體checkpoint的一致性。消息A到達(dá)map任務(wù)時頭部consmer繼續(xù)讀取下條消息（消息C）
Step5
這個階段展示了flink map任務(wù)從兩個源接收barrier，并且checkpoint它的狀態(tài)到Job Master。與此同時，consumer繼續(xù)從kafka partition讀取更多事件。
Step6
這個階段展示Map任務(wù)一旦checkpoint state完成后與Job Master的交互。當(dāng)job的所有任務(wù)checkpoint完成和響應(yīng)時，Job Master完成checkpoint。從現(xiàn)在開始，checkpoint可以用來故障恢復(fù)。值得一提的是Apache FLink 不依賴kafka的offsets機(jī)制來恢復(fù)潛在的系統(tǒng)錯誤
Recovery in case of a failure
在失敗例子中（例如，一個worker失敗）所有的算子任務(wù)被重啟并且它們的狀態(tài)被重置到最后一次checkpoint。像下圖描述的一樣。
Kafka源開始消費分別在2和1的位點，在已完成checkpoint的offset。當(dāng)job重啟時我們期望一個普通系統(tǒng)就像沒有錯誤發(fā)生一樣。
