KAFKA的ISR的伸縮過程是什么

196次閱讀

沒有評論

共計 3058 個字符，預計需要花費 8 分鐘才能閱讀完成。

本篇內容介紹了“KAFKA 的 ISR 的伸縮過程是什么”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓丸趣 TV 小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學有所成！

一、前言

知識點總結

二、概述

我們了解 ISR 列表是不斷伸縮的，在副本失效后及時踢出 ISR 列表，在副本趕上進度之后重新將副本加入到 ISR 列表中，后面我們就會按照這個思路來看下其中細節。

三、什么是失效副本？

功能失效：節點宕機，在該節點上的副本都屬于功能失效副本。
同步失效：follower 副本所在的 broker 因為帶寬或者負載等因素無法及時完成同步，導致被踢出 ISR。

四、ISR 伸縮控制參數了解嗎？

在 0.9x 版本之前，有一個控制參數：replica.lag.max.messages 默認值為 4000，表示如果 follower 的消息個數落后 leader 個數 4000，那么就會被踢出 ISR 列表；
我們可以想一下這種直接指定條數的方式是否合理呢？顯然是不合理的，原因入下：
高吞吐的場景：瞬間就幾萬條消息，可能 follower 就滯后個幾秒鐘就被判定為失效從而被踢出，可能導致 ISR 列表頻繁的變動，以及元數據的頻繁更新。
低吞吐的場景：可能一天就幾條消息，那可能 follower 都滯后好幾天了依舊存在于 ISR 中，那 ISR 不就失去意義了嗎？

因此 0.9x 版本開始，移除了該參數，取而代之的參數是 replica.lag.time.max.ms 該參數默認值是 10000ms，也就是 10s。
也就是說如果 follower 在 10s 都沒能追上 leader 的 LEO，就會被認定為失效，從而踢出 IS 列表。

五、ISR 是如何將失效副本剔除的？

我們知道了 ISR 是如何判定失效副本后，再來看下，到底是怎么把這個失效的副本踢出去的呢？
1、每個 broker 在啟動的時候都會啟動兩個定時任務：

isr-expiration：定時檢查當前 broker 上的 eader 對應的副本失效信息，也就是看當前 Leader 的 ISR 列表中是否存在失效副本，默認執行周期為 replica.lag.time.max.ms / 2 = 5s。

isr-change-propagation：定時檢查內存 isrChangeSet 中是否有新的變更數據，固定執行周期為 2.5s

2、判斷副本失效:
isr-expiration 任務會根據當前時間 now，減去某 follower 的 lastCaughtUpTimeMs，如果大于 replica.lag.time.max.ms 值，則說明失效。
而 lastCaughtUpTimeMs 這個值，在 follower 的 LEO 與 leader 的 LEO 相等時 (Leader 中維護了 follower 的 LEO 信息)，被更新。
也就是說，只有當 follower 完全追上了 Leader 才更新，而不是每 Fetch 一次就更新。

關于為什么不是每次 Fetch 的時候就更新該值呢？
我們試想一下，如果 leader 的寫入速率遠大于 follower 的同步速率，可能 leader 已經寫了 10w 條數據了，follower 由于網絡 / 負載為原因還在慢悠悠的同步，但是因為 Fetch 請求是正常發送的，就每次都更新 lastCaughtUpTimeMs 值，從而認為該 follower 是有效的，那這不就導致 leader 和 follower 之間在這種場景下存在巨大的數據差異了嘛？從而影響數據的可靠性。

3、這個 ISR 變化的信息如何傳遞呢？

由 leader 所在的 broker 的 isr-expiration 定時任務，去檢查失效副本和更新 zk 的 /state 節點數據，同時寫入 isrChangeSet。

isr-change-propagation 去檢查 isrChangeSet 是否有新增數據，如果有，則往 zk 中的 /isr_change_notification 節點下創建子節點。

而 Controller 對這個節點有一個 Watcher，如果發現新增了子節點，那么 Controller 就會重新從 zk 中獲取到最新的元數據，然后通知所有 Broker 更新元數據。

從上述過程中，我們還可以知道，實際上這個變更的數據會在內存中停留一段時間，假如這個時候我們對應的 broker 宕機了，那么不就是改了 zk 卻沒有讓其他 broker 更新元數據嗎？
其實不是，因為這種情況下，broker 宕機會觸發 controller 在 zk 下的 brokers/ids 下對應的節點被刪除，因此 Controller 也會讓其他的 broker 更新元數據，所以無論如何都會更新。

最后我們來總結一下整個 ISR 剔除的過程：
每個 leader 在啟動的時候都會啟動兩個定時檢查任務，每隔一段時間檢查是否存在失效副本。
如果某個 follower 的 lastCaughtUpTimeMs 10s 那么就會被判定為失效副本
如果定時任務掃描到存在失效副本時，就會往 zk 的 /state 節點下更新最新的 ISR 列表數據，同時將變更數據寫入到內存中的 isrChangeSet 中。
然后另外一個傳播任務會定時檢查 isrChangeSet 是否存在需要變更的任務，如果感知到就往 zk 的 /isr_change_notification 節點下創建子節點。
最終由 Controller 感知到節點的變化，然后從 zk 中獲取最新的元數據，然后通知所有的 Broker 更新元數據，完成整個 ISR 列表的數據更新。

六、追趕上的副本是如何重新加入 ISR 中的？

在看完第五小節之后，第六小節就會顯得非常簡單，無非是需要知道什么時候一個副本會重新判定為同步副本呢？那就是：當前失效 follower 的 LEO 等于 leaderHW 的時候，即被判斷可以重新加入 ISR。

那么隨之而來的一個問題就是在哪兒去判斷 followerLEO == leaderHW 的呢？
這里和上面的剔除 ISR 成員不一樣，并不是由定時任務去檢測的，而是在處理完 Fetch 請求的時候，如果判斷 Fetch 請求是 follower 發過來的的（replicaId = 0），那么就會去看下當前這個 follower 的 LEO 是多少（其實就是 Fetch 請求帶過來的），是不是趕上了當前的 leaderHW，如果是的那么就執行擴張 ISR 操作。
擴張 ISR 操作流程就和上面流程一樣了，先寫 zk 下的 /state 數據，然后寫 isrChangeSet，最后由 Controller 感知到數據變化，更新集群元數據。
我們所需要記住的主要差別點在于，ISR 列表的擴張是在 Fetch 請求的時候去判斷和執行的。

七、整個伸縮過程總結

最后，我們用圖示來加深一點印象。
1、失效副本（圖源：《深入理解 kafka》）：