Kafka Need No Keeper
最近在鵝廠工作中不斷接觸到Kafka，雖然以前也使用過，但是對其架構和發展過程總是模模糊糊，所以在回學校准備末考的時候找些資料總結一下。
Kafka Need No Keeper 是一個在Kafka Submit分享的標題，我也是看了Kafka needs no Keeper（關於KIP-500的討論）這篇博客分享后才對Kafka有了初期的認識，如果想要了解細節的話可以直接閱讀該博客分享，本篇博客是一次對Kafka的自我總結，多少有些大白話和概括之意。
Kafka架構
Kafka是什么？Apache Kafka 是一款分布式流處理框架（新版本后，定位發生了改變），用於實時構建流處理應用。
Kafka的架構可以簡單分為Client和Broker兩部分。在Kafka發展過程中，Kafka都是不斷減少這兩部分對Zookeeper的依賴。
那為什么要減少對Zookeeper的依賴呢？
- Kafka在新版本后定位變成了分布式流處理框架，但是本質上還是一個消息中間件，中間件與中間件之間不應該存在依賴關系，需要降低耦合。
- Kafka與Zookeeper不斷通信，不斷寫入數據，而Zookeeper一致性要求較高，當某個數據節點信息發生變更時，會通知其他節點同步更新，半數以上完成更新才能返回，寫入性能較差，影響了Kafka的性能。
Client架構
Client一般分為三類，Consumer Client、Producer Client和Admin Tool。
舊版架構
- Producer Client 只需要向Kafka集群中發送消息，不需要連接Zookeeper
- Consumer Client 需要讀取某主題某分區內的消息，那么需要知道讀取哪條消息（讀取offset）和下一次讀哪條消息（提交offset）,所以需要和Zookeeper交互（offset保存在ZK中）
- Admin Tool 執行主題的操作，因為元數據保存在ZK中，所以需要與ZK交互
可以看出，Zookeeper在Kafka中①存儲元數據
新版架構
新版主要針對舊版中的Consumer Client和Admin Tool改進
- Offset改進：在Kafka中新建一個內部主題_consumer_offset用來保存消費者組的offset，提交和獲取offset都可以直接與Kafka集群交互獲取。
- Rebalance改進：在舊版架構中，消費者組中的消費者消費的主題分區信息都是保存在ZK中，在新版架構改進中，每一個消費組使用一個Coordinator來控制重分區過程。
- Admin改進：社區引入了新的運維工具AdminClient以及相應的CreateTopics、DeleteTopics、AlterConfigs等RPC協議，替換了原先的Admin Tool，這樣創建和刪除主題這樣的運維操作也完全移動Kafka這一端來做。
| Question | Answer |
|---|---|
| 重分區是什么？ | 如上圖，重分區就是將消費者組里訂閱主題下的分區重新分配給當前組內消費者實例的過程。 |
| 重分區發生條件是什么？ | ①消費者組消費者數量改變； ②訂閱的主題數量改變； ③訂閱的主題下分區數量改變。 |
| 怎么進行重分區？ | 真正的重分區是有Group Leader來完成的。 第一個進入Consumer Group的消費者實例為leader，它向Coordinator申請消費者組成員列表，然后按照分區策略進行分區，接着將分區的結果告訴Coordinator，最后由Coordinator告知所有的消費者分區信息。 |
| Coordinator是怎么找到的 | 消費者組向任意一個Broker發送groupCoordinatorRequest請求，集群返回一個負載最小的Broker節點使其成為當前消費者組的Coordinator。 |
| 分區策略是什么？ | ①Range分區（默認）：分塊分區，對於每一個主題而言， 首先將分區按數字順序排行序，消費者按名稱的字典序排序，然后用分區總數除以消費者總數。如果能夠除盡，平均分配；若除不盡，則位於排序前面的消費者將多負責一個分區。 ②RoundRobin分區：輪詢分區，對所有主題而言，首先將所有主題的分區組成列表，然后按照列表重新輪詢分配分區給不同的消費者。 |
Broker架構
現階段架構
在現階段結構中，Broker端是嚴重依賴Zookeeper的，基本上所有元數據信息和管理都要通過Zookeeper集群，如下圖：
可以看出，Zookeeper在Kafka中有②集群管理和③選舉Controller的作用
發展中的架構
第一步首先是隔離非Controller端對ZK的依賴；
第二步是移除Controller端對ZK的依賴，這一步可以采用基於Raft的共識算法來做(?)。
Kafka同步副本管理
基本概念
| 概念 | 簡介 |
|---|---|
| LEO | Log End Offset。日志末端位移值或末端偏移量，表示日志下一條待插入消息的 位移值。 |
| LSO | Log Stable Offset。這是 Kafka 事務的概念。如果你沒有使用到事務，那么這個值無意義。該值控制了事務型消費 者能夠看到的消息范圍。它經常與 Log Start Offset，即日志起始位移值相混淆，因為 有些人將后者縮寫成 LSO，這是不對的。在 Kafka 中，LSO就是指代 Log Stable Offset。 |
| HW | 高水位值(High watermark)。這是控制消費者可讀取消息范圍的重要字段。一 個普通消費者只能“看到”Leader 副本上介於 Log Start Offset 和 HW(不含)之間的 所有消息。水位以上的消息是對消費者不可見的。 |
| AR | Assigned Replicas。AR 是主題被創建后，分區創建時被分配的副本集合，副本個數由副本因子決定。 |
| ISR | In-Sync Replicas。Kafka 中特別重要的概念，指代的是 AR 中那些與 Leader 保持同步的副本集合。在 AR 中的副本可能不在 ISR 中，但 Leader 副本天然就包含在 ISR 中。 |
Kafka文件大小對應關系：
ISR
Leader 與 Follower
ISR中的Leader是由Controller指定，與Leader保持同步用指標來衡量就是follower中LEO落后leader中LEO的時間不超過指定時間范圍（replica.lag.time.max.ms=10s）。
（在舊版本中還有另外一個指標是落后的LEO條數，不過這樣子的話每次發送大量數據后，一開始ISR就只有leader，到后面follower跟上的才能加入ISR，這樣子會導致ZK的頻繁寫入修改性能下降）
另外在Leader掛掉后，Controller會讓ISR中的一個Follower成為Leader，並且開始同步新的Leader的Offset。這里要注意的是有可能此時ISR中並沒有Follower，所以有兩種選擇，①允許OSR的Follower成為Leader和②該分區沒有Leader。這來源於設置unclean.leader.election.enable，設置為true為選擇①，保證了系統的高可用性和損失了一致性，設置為false為選擇②，保證系統的一致性和損失高可用性。
同時一個Leader和多個Follower看上是讀寫分離的結構，但是Kafka並不支持讀寫分離。原因由兩點，①場景不合適，讀寫分離適用於讀負載很大，而寫操作不頻繁的場景，顯然Kafka不是；②同步機制，Follower和Leader之間存在不一致的窗口，很可能出現消息滯后（類似於幻讀）
ACK機制
這主要決定了Producer發送信息時，Kafka的接受機制，有三種：
| ACK | 機制 |
|---|---|
| ack = 0 | at most once，最多一次語義，Producer不需要等待Broker回發確認消息，直接發送下一批消息。 |
| ack = 1 | at least onve，最少一次語義，Producer只要Leader成功消息並且返回確認后，就可以發送下一批消息 |
| ack = -1 | Producer需要等到Leader和ISR中的Follower同步完成並且返回確認后，才能發送下一批消息 |
那么問題就來，怎么實現Exactly Once呢？
Kafka Exactly Once 和事務機制
這里討論的Exactly Once主要是針對Producer端，至於消費者的Exactly Once可以在客戶端上保留偏移量來實現（參見flink事務機制）。
單Session情況
先來討論單Session的情況，在Kafka中給每個Producer都分配了一個內部的唯一的PID，每次Producer發送信息時，帶有的主鍵是<PID ,Topic,Partition,SequenceNumber>，Leader端收到信息后對相同的<PID,Topic,Partition>的SequenceNumber進行比較，如果來的信息比Leader端的小，證明數據重復，丟棄該條信息；如果來的信息比Leader端的大1，插入該信息；吐過來的信息比Leader端的大超過1，證明發生了亂序丟棄該信息。
跨Session情況
簡單理解
在單Session的情況如果存在PID都可以保證Exactly Once，那么要是在不同的Session中我能拿到相同的PID就可以了。所以引入了一個TID（自己定義的）並且綁定了事務一開始的PID，只要事務沒有提交，那么每次都拿着這個TID去獲取對應的PID就可以保證Exactly Once了。
具體做法
內部引入了一個Transaction Coordinator用於分配PID和管理事務，並且在內置了一個主題Transaction Log用於記錄事務信息，事務的操作簡圖如下：
| 步驟 | 具體內容 |
|---|---|
| 1.請求/返回Transaction Coordinator | 由於Transaction Coordinator是分配PID和管理事務的核心，因此Producer要做的第一件事情就是通過向任意一個Broker發送FindCoordinator請求找到Transaction Coordinator的位置。 |
| 2.TID->PID | 找到Transaction Coordinator后，具有冪等特性的Producer必須發起InitPidRequest請求以獲取PID。 |
| 3 Producer生產消息 | ①Producer拿到PID后向Kafka主題發送消息 ②Transaction Coordinator會將該<Transaction, Topic, Partition>存於Transaction Log內，並將其狀態置為BEGIN |
| 4 事務完成 | ①將PREPARE_COMMIT或PREPARE_ABORT消息寫入Transaction Log。 ②以Transaction Marker的形式將COMMIT或ABORT信息寫入用戶數據日志以及_consumer_log中。 ③最后將COMPLETE_COMMIT或COMPLETE_ABORT信息寫入Transaction Log中。 |
與兩階段提交的區別
- Kafka事務機制中，PREPARE時即要指明是PREPARE_COMMIT還是PREPARE_ABORT，並且只須在Transaction Log中標記即可，無須其它組件參與。而兩階段提交的PREPARE需要發送給所有的分布式事務參與方，並且事務參與方需要盡可能准備好，並根據准備情況返回Prepared或Non-Prepared狀態給事務管理器。
- Kafka事務中，一但發起PREPARE_COMMIT或PREPARE_ABORT，則確定該事務最終的結果應該是被COMMIT或ABORT。而分布式事務中，PREPARE后由各事務參與方返回狀態，只有所有參與方均返回Prepared狀態才會真正執行COMMIT，否則執行ROLLBACK
- Kafka事務機制中，某幾個Partition在COMMIT或ABORT過程中變為不可用，只影響該Partition不影響其它Partition。兩階段提交中，若唯一收到COMMIT命令參與者Crash，其它事務參與方無法判斷事務狀態從而使得整個事務阻塞
- Kafka事務機制引入事務超時機制，有效避免了掛起的事務影響其它事務的問題
- Kafka事務機制中存在多個Transaction Coordinator實例，而分布式事務中只有一個事務管理器