前言
Kafka的broker端有ISR機制，它可以看成多個副本的集合型诚，里面有l(wèi)eader副本和多個follower副本委刘，數(shù)量是我們可以配置的，只有ISR中的副本才有可能成為leader副本们拙。
我們可以通過一些參數(shù)的配置，例如ack的配置阁吝、生產(chǎn)者重試砚婆、isr最小數(shù)量，消費者改為手動提交等方式來盡可能的保障Kafka的消息可靠性突勇。
Kafka在保證數(shù)據(jù)的可靠性上使用的是‘?dāng)?shù)據(jù)冗余’的方式装盯，即將一個分區(qū)下的數(shù)據(jù)保存到多個副本中，起到備份數(shù)據(jù)的作用甲馋。這樣如果leader掛了埂奈，會重新選舉follower作為leader繼續(xù)工作，那么Kafka中l(wèi)eader副本是如何將數(shù)據(jù)同步到follower副本中的呢定躏？
broker里的高水位（High Watermark）
什么是高水位账磺？
在Kafka中芹敌，高水位是一個位置信息標(biāo)記，它是用消息位移來表征的垮抗，比如某個副本中HW=8氏捞，就是這個副本的高水位在offset=8那個位置上。
還有一個日志末端位移的概念借宵，Log End Offset，縮寫是LEO矾削。它表示副本寫入下一條消息的位移值壤玫。上圖中LEO是15，即下一條新消息的位移是15哼凯，8-14這些位置上的消息就是未提交消息欲间。同一個副本對象，其高水位值不會大于LEO值断部。
HW的作用
在Kafka中猎贴，高水位的作用主要有2個
- 定義消息的可見性，即用來標(biāo)識分區(qū)下的哪些消息是可以被消費的蝴光，比如某個分區(qū)的HW（leader的HW）是8她渴，那么這個分區(qū)只有 < 8 這些位置上消息可以被消費。即高水位之前的消息才被認為是已提交的消息蔑祟，才可以被消費趁耗。
- 幫助Kafka完成副本同步。
需要注意的是疆虚，位移值等于高水位的消息也屬于未提交消息苛败。也就是說，高水位上的消息是不能被消費者消費的
分區(qū)的高水位就是其leader副本的高水位径簿。
副本同步機制解析
首先我們先看一個例子：
開始時罢屈，follower向leader副本發(fā)送fetch同步數(shù)據(jù)請求，此時他們的HW和LEO都是0篇亭。
當(dāng)生產(chǎn)者向leader發(fā)送一條消息缠捌，且提交成功后，leader的LEO更新為1译蒂，這個時候follower發(fā)現(xiàn)有消息可以拉取了鄙币，于是follower的LEO也更新為1，但是此時leader和follower的HW都為0蹂随，它們需要在下一輪的拉取中被更新十嘿。
在新一輪的拉取請求中，因為之前位移值是0的消息已經(jīng)拉取成功了岳锁，所以follower這次請求拉取的是位移值=1的消息绩衷。leader接收到這個請求后，將遠程副本LEO更新為1，然后更新leader的HW=1咳燕，最后將HW=1發(fā)送給follower副本勿决，follower收到后將自己的高水位值更新成1。
至此招盲，一次完整的消息同步周期就結(jié)束了低缩。Kafka就是利用這樣的機制，實現(xiàn)了leader和follower之前的同步曹货。
這里看高水位機制咆繁，有點像分布式事務(wù)的二次提交。
依托于高水位顶籽，Kafka既界定了消息的對外可見性玩般，又實現(xiàn)了異步的副本同步機制。但是這里面還存在一些問題礼饱。剛才我們了解到的Kafka副本同步的過程中坏为，follower副本的高水位更新需要一輪額外的拉取請求才能實現(xiàn)。如果有多個follower副本镊绪，情況可能更糟匀伏，就需要多輪拉取請求。也就是說蝴韭，leader副本的高水位更新和follower副本高水位更新在時間上是存在一定延遲的帘撰，這樣會導(dǎo)致數(shù)據(jù)丟失或者數(shù)據(jù)不一致。
我們再來看下數(shù)據(jù)丟失的場景：
-
數(shù)據(jù)丟失
下圖中万皿，副本A是leader摧找，副本B是follower，現(xiàn)在情況是A的LEO和HW都是2牢硅，副本B中的LEO也更新成了2蹬耘，但是HW還是1，它需要再一輪請求來更新自己的HW减余，但是此時因為某些原因综苔，B重啟了，重啟完后位岔，B會執(zhí)行日志截斷功能如筛，將LEO調(diào)整到它的高水位位置，即副本B的LEO變?yōu)榱?抒抬，就是說副本B中位置為1的消息被刪除了杨刨，現(xiàn)在只有0一條消息了。
image.png
執(zhí)行完截斷操作后擦剑，副本B開始從A拉取消息妖胀，執(zhí)行正常的消息同步芥颈。如果在這個時候，副本A所在的Broker宕機了赚抡，那么kafka只能選B成為新的leader了爬坑，當(dāng)A重啟回來后，需要執(zhí)行相同的日志截斷工作涂臣，將高水位調(diào)整為B的高水位值盾计，因為規(guī)定follower的HW值不能超過leader中的HW值。A中的位移1位置的消息也被刪除了赁遗，這樣這條消息就丟失了署辉。
- 數(shù)據(jù)不一致
上圖中副本A和B，其中副本A中有兩條消息吼和，LEO是2涨薪、HW是2骑素，B的LEO和HW都是1炫乓，假設(shè)在同一時刻，A和B都宕機了献丑，然后B先醒過來末捣，那么B成了新的leader，然后他收到生產(chǎn)者發(fā)來的m3消息创橄，然后B的LEO和HW都更新成了2箩做。
當(dāng)A醒過來后，會先根據(jù)HW判斷是否需要進行日記截斷妥畏，這里HW和LEO相等邦邦，發(fā)現(xiàn)不需要進行日志截斷，然后跟B進行同步醉蚁，這個時候A和B的LEO都是2燃辖，這樣A中的消息是<m1,m2>,B中的消息是<m1,m3>,出現(xiàn)了消息不一致的情況。
Leader Epoch是什么
Leader Epoch 可以認為是leader的版本网棍，它由兩部分數(shù)據(jù)組成黔龟。（這里kafka應(yīng)該是借鑒了zk的數(shù)據(jù)同步機制ZAB）
- Epoch，一個單調(diào)增加的版本號滥玷。每當(dāng)副本領(lǐng)導(dǎo)權(quán)發(fā)生變更后氏身，都會增加該版本號。小版本號的leader被認定是過期leader惑畴。
- 起始位移蛋欣，leader副本在該Epoch值上寫入的首條消息的位移。
假設(shè)現(xiàn)在有兩個 Leader Epoch<0, 0> 和 <1, 120>如贷，那么豁状，第一個 Leader Epoch 表示版本號是 0捉偏，這個版本的 Leader 從位移 0 開始保存消息，一共保存了 120 條消息泻红。之后夭禽，Leader 發(fā)生了變更，版本號增加到 1谊路，新版本的起始位移是 120讹躯。
Kafka Broker 會在內(nèi)存中為每個分區(qū)都緩存 Leader Epoch 數(shù)據(jù)，同時它還會定期地將這些信息持久化到一個 checkpoint 文件中缠劝。當(dāng) Leader 副本寫入消息到磁盤時潮梯，Broker 會嘗試更新這部分緩存。如果該 Leader 是首次寫入消息惨恭，那么 Broker 會向緩存中增加一個 Leader Epoch 條目秉馏，否則就不做更新。這樣脱羡，每次有 Leader 變更時萝究，新的 Leader 副本會查詢這部分緩存，取出對應(yīng)的 Leader Epoch 的起始位移锉罐，以避免數(shù)據(jù)丟失和不一致的情況帆竹。
Leader Epoch是如何解決消息丟失和消息不一致的問題的
消息丟失
還是剛才那個場景，現(xiàn)在有了Leader Epoch機制的介入脓规，當(dāng)副本B重啟回來后栽连，會向A發(fā)送一個請求去獲取leader中的LEO，發(fā)現(xiàn)A的LEO=2侨舆，不比它自己的LEO值小秒紧，而且緩存中沒有保存任何起始位移值 > 2的Epoch條目，這樣B就不需要執(zhí)行日志截斷操作了挨下。
然后副本A宕機了熔恢，B成為了leader，B的Leader Epoch由原來的<0, 0> 更新成了 <1, 2>复颈，意思說是B成為了新的leader绩聘，版本號+1，這個leader的起始消息位移值為2耗啦。
A重啟回來后會向B發(fā)送請求獲取B的LEO凿菩，發(fā)現(xiàn)等于2，和自己相同帜讲，并且緩存中的Leader Epoch的起始位移值是2衅谷，也不需要進行日志截斷。這樣就不會出現(xiàn)消息丟失的問題了似将。
消息不一致
有了Leader Epoch機制的加入获黔，當(dāng)B變?yōu)閘eader后蚀苛，producer發(fā)送m3消息到B中，數(shù)據(jù)保存到磁盤上玷氏，Leader Epoch會更新為<1堵未，2>,然后A醒過來后，會先發(fā)送請求知道B的LEO值為2和自己一樣盏触，然后通過緩存的Leader Epoch值渗蟹，得知下一條要寫入的消息是1的位置，然后就會進行日志截斷赞辩，將原先的m2刪除雌芽，再將m3寫入。由此解決了消息不一致的問題辨嗽。
思考
- 這里為什么高水位的時候世落，副本B重啟要截斷offset=1的數(shù)據(jù)，而引入epoch后糟需，就不需要截斷了（這也是消息丟失問題所在）屉佳？
設(shè)計者角度思考下，沒有引入epoch的時候篮灼，副本B重啟發(fā)現(xiàn)HW與LEO不一致忘古，它并不知道其間的數(shù)據(jù)徘禁，是不是當(dāng)前l(fā)eader同步來的诅诱，并不確定是否和當(dāng)前l(fā)eader的數(shù)據(jù)一致（只有HW以前的才能保證是與leader同步好的）
引入epoch后，可以判斷當(dāng)前l(fā)eader的epoch-offset位后的數(shù)據(jù)都是當(dāng)前l(fā)eader的數(shù)據(jù)（這里再想下送朱，如果副本B的HW是更前的offset娘荡，超過了當(dāng)前l(fā)eader的epoch-offset位時要怎樣？這里就會截斷驶沼，然后同步當(dāng)前l(fā)eader的副本B的HW位后的所有數(shù)據(jù)）
- 消息不一致的問題
同樣也是會判斷當(dāng)前重啟的broker炮沐，看最新leader的epoch-offset，看是否與本身的HW位有交叉回怜，如果有大年，那需要與新leader同步數(shù)據(jù)，截斷新leader的epoch-offset后的數(shù)據(jù)（因為這部分不是老leader寫入的）
