Kafka2.8安裝
1.概述
最近Kafka官網發布了2.8版本，在該版本中引入了KRaft模式。鑒於新版本和新特性的引入，相關使用資料較少，那邊本篇部落格筆者將為大家介紹Kafka2.8的安裝和使用。
2.內容
2.1 版本介紹
2.1.1 目的
從Kafka2.8版本開始，可以不用Apache Zookeeper來作為Kafka的依賴組件了，官網把這種稱之為KRaft模式。目前，Kafka使用Zookeeper來存儲有關分區和Broker的元數據，並選擇一個Broker作為Kafka的Controller。現在官網打算刪除對Zookeeper的依賴，讓Kafka能夠以更具擴展性和更加強大的方式管理元數據，從而支援更多分區。
2.1.2 元數據作為事件日誌
我們經常會討論將狀態作為一系列事件進行管理的好處。單個數字（偏移量）描述了Consumer在Stream中的位置。只需要重播所有比其當前偏移量新的事件，多個Consumer就可以迅速趕上最新的狀態。該日誌在事件之間建立了清晰的順序，並確保使用者始終沿著單個時間軸移動。
但是，這些好處Kafka卻沒有享受到，對元數據的更改視為互相之間沒有關係的孤立更改。當Controller向集群中的其他Broker推送狀態更改通知（例如LeaderAndIsrRequest）時，Broker可能會獲取部分更改，但不是全部。儘管Controller重試了幾次，但最終還是放棄了。這樣會使Broker處理分歧狀態，更壞的是，儘管Zookeeper是記錄的存儲，但是Zookeeper中的狀態通常與Controller記憶體中保存的狀態不匹配。例如，當分區Leader在Zookeeper中更改了其ISR時，Controller通常在幾秒鐘內不了解這些更改。Controller沒有通用的方法來遵循Zookeeper事件日誌。儘管Controller可以設置一次Watch，但是由於性能原因，Watch的數量收到限制。監視觸發時，它不會告訴Controller當前狀態，而只是告訴狀態已更改。到Controller重新讀取znode並設置新的Watch時，狀態可能已經與Watch最初觸發時的狀態有所變化。如果沒有Watch，則Controller可能根本不了解該變化。
某些情況下，元數據應該存儲在Kafka中，而不是存儲在單獨的系統中。這將避免與Controller狀態和Zookeeper狀態之間的差異相關的所有問題。Broker不應將通知發送給Broker，而應僅使用事件日誌中的元數據事件。這樣可以確保元數據更改始終以相同的順序到達。Broker將能夠在文件中本地存儲元數據。當它們啟動時，它們只需要從Controller讀取已更改的內容，而無需讀取完整狀態。這將使我們以更少的CPU消耗支援更多的分區。
2.1.2 簡化部署和配置
Zookeeper是一個單獨的系統，具有自己的配置文件語法，管理工具和部署模式。這意味著系統管理員需要學習如何管理和部署兩個獨立的分散式系統才能部署Kafka。對於管理員來說，這可能是一項艱巨的任務，特別是如果他們對部署Java服務不是很熟悉的話。統一該系統將大大改善運行Kafka的首次體驗，並有助於擴大其應用範圍。
由於Kafka和Zookeeper配置是分開的，因此很容易出錯。例如，管理員可能在Kafka上設置了SASL，並且錯誤的認為它們已經保護了通過網路傳輸的所有數據。實際上，這樣做還必須在單獨的在Zookeeper系統中部署配置安全性。
2.2 架構介紹
KIP-500提出了可擴展的Zookeeper Kafka後的總體構想，為了展示整體情況，忽略了諸如RPC格式、磁碟格式等詳細資訊。如下圖所示：
當前，Kafka集群包含了幾個Broker節點，以及Zookeeper節點的外部選舉。在此圖中，我們描繪了4個Broker節點和3個Zookeeper節點。這是小型集群的典型規劃。Controller（用白色表示）在選定後從Zookeeper Leader中載入其狀態。從Controller延伸到Broker中其他節點的箭頭指向表示Controller推送的更新，例如LeadAndIsr和UpdateMetaData消息。
需要注意的是，除了Controller之外的其他Broker可以並且確實與Zookeeper能夠進行通訊。因此，實際上，應該從每個Broker到Zookeeper進行劃清界限。另外，外部命令行工具和應用程式可以在Zookeeper中修改狀態，而無需Controller的參與。如前所述，這些問題使得很難知道Controller上的記憶體中的狀態是否真正反映了Zookeeper中的持久狀態。
在提出的體系結構中，三個Controller節點替代了三個Zookeeper節點。Controller節點和Broker節點在單獨的JVM中運行。Controller節點為元數據分區選擇單個Leader，以白色表示。Broker不是從Controller向Broker發布更新，而是從該Leader提取元數據更新。這就是為什麼箭頭指向Controller而不是Broker的原因。
請注意，儘管Controller進程在邏輯上與Broker進程是分開的，但是它們在物理上並不需要分開。在某些情況下，將某些或者所有Controller進程與Broker進程部署在同一節點上可能是有意義的。這類似於Zookeeper進程可以與較小規模的集群中的Kafka Broker部署在相同的節點上。
2.3 Controller數量
Controller節點包括一個Raft選舉，用於管理元數據日誌。該日誌包含有關集群元數據的每次更改的資訊。當前存儲在Zookeeper中的所有內容，例如Topic，分區，ISR，配置等，都將存儲在此日誌中。
使用Raft演算法，Controller節點將在不依賴任何外部系統的情況下從他們當中選出一個Leader。元數據日誌的Leader稱之為Active Controller。Active Controller處理由Broker生成的所有RPC，Follower Controller複製寫入數據到Active Controller，並在Active Controller出現故障時用作熱備用伺服器（HA）。由於Controller現在都將跟蹤最新狀態，因此Controller故障轉移將不需要很長的重新載入時間，在此期間我們將所有狀態都轉移到新的Controller。
就像Zookeeper一樣，Raft需要大多數節點才能繼續運行。因此，三個節點Controller集群可以承受一次故障。五個節點Controller集群可以承受兩次故障，依此類推。Controller定期將元數據的快照寫到磁碟上，儘管從概念上講這與壓縮類似，但是程式碼路徑會有所不同，因為我們可以簡單的從記憶體中讀取狀態，而不是從磁碟中重新讀取日誌。
2.4 Broker元數據管理
這些Broker將通過新的MetadataFetch API從Active Controller中獲取更新，而不是Controller將更新推送給其他Broker。MetadataFetch與fetch請求類似，就像fetch請求一樣，Broker將跟蹤其獲取的最後更新的偏移量（offset），並且僅從Active Controller中請求更新。
Broker將把提取到磁碟的元數據持久化。即使存在成千上萬甚至數百萬個分區，這也可以使Broker快速啟動。（請注意，由於這種持久性是一種優化，因此如果使開發更容易，可以將其白柳在第一個版本之外）。
在大多數情況下，Broker只需要獲取增量，而無需獲取完整狀態。但是，如果Broker與Active Controller的距離太遠（落後Active Controller太遠），或者Broker完全沒有快取的元數據，則Controller將發送完整的元數據鏡像，而不是一系列增量。
Broker將定期向Active Controller請求元數據更新，該請求作為心跳，讓Controller知道Broker處理活躍狀態。
2.5 Broker狀態機制
目前，Broker在啟動後立即向Zookeeper註冊，該註冊完整了兩件事：它使Broker知道了它是否已經被選舉為Controller，並且它使其他節點知道如何聯繫它。
當前，如果Broker丟失了其Zookeeper會話，則Controller會將其從集群元數據中刪除。在以後Zookeeper時代，如果Active Controller在足夠長的時間內未發送MetadataFetch心跳，則Active Controller將從集群元數據中刪除Broker。
在當前情況下，可以聯繫Zookeeper從Controller進行分區的Broker將繼續滿足用戶的請求，但是不會接收任何元數據更新。這可能會導致一些令人困惑和困難的情況。例如，使用acks=1的生產者可能會繼續生產實際上不再是該Leader的Leader，但是無法接收到Controller的LeaderAndIsrRequest來遷移Leader。
在以後Zookeeper的世界中，集群成員身份與元數據更新集成在一起，如果Broker無法接收元數據更新，則他們將無法繼續成為集群的成員。儘管仍然可以從特定Client對Broker進行分區，但是如果從Controller進行了分區，則Broker將從集群中刪除。
- Offline：當Broker進程處理離線狀態時，它要麼根本不運行，要麼不執行啟動所需要的單節點任務，例如初始化JVM或執行日誌恢復；
- Fenced：當Broker處理Fenced狀態時，它將不會響應來自Client的RPC。當啟動並嘗試獲取最新的元數據時，Broker將處理受保護狀態。如果無法聯繫Active Controller，它將重新進入隔離狀態。發送給Client的元數據中應省略受保護的Broker；
- Online：當Broker在線時，它準備響應Client的請求；
- Stopping：Broker收到SIGINT時便進入停止狀態，這表明系統管理員要關閉Broker。當Broker停止時，它仍在運行，但是我們正在嘗試將分區Leader移除該Broker。最終，Active Controller將通過MetadataFetchResponse中返回特殊的結果程式碼來要求Broker最終離線。或者，如果Leader不能在預定時間內移動，則Broker將關閉。
另外，以前直接寫給Zookeeper的需要操作將變為Controller操作。例如，更改配置，更改默認授權者存儲的ACL等。Client的新版本應將這些操作直接發送到Active Controller，它將與新舊集群一起使用。為了保持與將這些操作發送給隨機Broker的舊Client的兼容性，Broker會將這些請求轉發到Active Controller。
在某些情況下，需要創建一個新的API來替換以前通過Zookeeper完成的操作，這樣的一個示例是，當分區的管理員想要修改同步副本集時，它當前直接就該Zookeeper。在以後Zookeeper中，管理員將改為向Active Controller進行RPC。
當前，某些工具和腳本直接與Zookeeper聯繫，在以後的Zookeeper中，這些工具必須改用為Kafka API。
3.Kafka2.8
KRaft目前在Kafka2.8版本是一個測試版本，KRaft模式不推薦使用到生產環境。當Kafka集群處理KRaft模式時，它不會將其元數據存儲在Zookeeper中，實際上根本不需要運行Zookeeper，因為它將元數據存儲在Controller節點中。
目前官網退出的KRaft模式僅用於測試，不推薦使用到生產環境，因為官方還不支援將現有的基於Zookeeper的Kafka集群升級到KRaft模式。實際上，當Kafka3.0發布時，無法將Kafka集群從2.8升級到3.0，目前該模式會有些BUG，如果嘗試KRaft使用到生產環境，會存在數據丟失的風險。
3.1 使用步驟
步驟1：生成集群ID
先生成集群ID，需要使用kafka-storage工具，命令如下：
./bin/kafka-storage.sh random-uuid
步驟2：格式化存儲目錄
接下來的步驟是格式化存儲目錄，如果你是運行單節點模式，你可以執行如下命令：
# wzhDv517Siy99Rm42vTHEA 這個是執行上面的命令獲取到的集群ID ./kafka-storage.sh format -t wzhDv517Siy99Rm42vTHEA -c ../config/kraft/server.properties
如果，你使用的多節點安裝，你需要在每個節點上執行格式化命令，並確保每個節點使用的集群ID是相同的。
步驟3：啟動Kafka服務
最後，你可以在每個節點上啟動Kafka服務，命令如下：
./kafka-server-start.sh ../config/kraft/server.properties
之後，我們可以使用jps命令查看Kafka進程是否已經成功啟動。
在啟動Kafka進程後，我們可以和之前Zookeeper+Kafka的模式一樣，通過連接9092埠來操作Topic，Consumer和Producer。例如，創建一個分區為1，副本1的Topic，命令如下：
./kafka-topics.sh --create --topic ke28 --partitions 1 --replication-factor 1 --bootstrap-server localhost:9092
3.2 Controller Server
在KRaft模式下，只有一小部分特別選定的伺服器可以充當Controller（與基於Zookeeper的模式不同，在這種模式下，任何伺服器都可以成為Controller）。特別選定的Controller伺服器將參與元數據的管理，每個Controller伺服器要麼是Active，要麼是Standby。
我們通常會為此角色選擇3台或者5台伺服器，具體取決於成本和系統應承受的並發故障數等因素。就像Zookeeper一樣，為了保持可用性，必須保持大多數Controller處理在線狀態，如果你有3個Controller，可以容忍1次故障，使用5個Controller，可以容忍2次故障。
3.3 Process Roles
每個Kafka服務現在有了一個新的配置，它被稱為「process.roles」，它有如下可選值：
- broker：它在KRaft模式中扮演一個Broker角色；
- controller：它在KRaft模式中扮演一個Controller角色；
- broker,controller：它在KRaft模式中同時扮演Broker和Controller角色；
- 如果沒有設置，則假定我們處於ZooKeeper模式。如前所述，如果不重新格式化，當前無法在ZooKeeper模式和KRaft模式之間來迴轉換。
充當Broker和Controller的節點稱為「combined」節點，該節點對簡單用例操作更簡單，並且允許避免與JVM相關的一些固定記憶體開銷。關鍵的缺點是Controller與系統的其餘部分隔離較少。例如，如果Broker上發生OOM，伺服器的Controller部分不會與該OOM隔離。
3.4 Quorum Voters
系統中的所有節點都必須設置「controller.quorum.votters」配置，這標識應使用的是選舉Controller的伺服器，必須枚舉所有的Controller。這類似在使用Zookeeper時，「zookeeper.connect」配置必須包含所有的Zookeeper伺服器。但是，與Zookeeper配置不同的是，「controller.quorum.configures」還有每個節點ID，格式為id1@host1:port1,id2@host:port2等。
所以，如果你有10個Broker和3和Controller命名為controller1，controller2，controller3，你可以按照如下進行配置：
process.roles=controller node.id=1 listeners=CONTROLLER://controller1.dn1.kafka-eagle.org:9093 controller.quorum.voters[email protected]:9093,[email protected]:9093,[email protected]:9093
每個Broker和Controller必須設置「controller.quorum.votters」。請注意，「controller.quorum.votters」配置中提供的節點ID必須與提供給伺服器的節點ID匹配。
所以在Controller1上，node.id必須設置為1，依此類推。這裡不要求Controller的ID從0或者1開始。然而，最簡單和最容易混淆的分配方法節點ID可能只是給每個伺服器一個數字ID，從0開始。
需要注意的是，Client端不需要配置「controller.quorum.votters」，只有伺服器端才需要配置。
3.5 kafka-dump-log
這裡我們可以通過kafka-dump-log.sh工具來查看metadata日誌資訊，命令如下所示：
./kafka-dump-log.sh --cluster-metadata-decoder --skip-record-metadata --files /data/soft/new/kafka2.8/data/kraft/\@metadata-0/*.log
執行結果如下圖所示：
3.5 Metadata Shell
可以通過kafka-metadata-shell.sh來查看元數據資訊，這個和Zookeeper Client操作很類似，命令如下：
./kafka-metadata-shell.sh --snapshot /data/soft/new/kafka2.8/data/kraft/\@metadata-0/00000000000000000000.log
然後，進入到一個命令行操作介面，操作步驟如下所示：
4.總結
目前Kafka2.8中的KRaft處於測試階段，大家如果是學習可以嘗試安裝部署了解和認識這個KRaft模式，生產環境暫時建議不推薦使用。Kafka2.8的主線版本還是以依賴Zookeeper為主，部落客開發的Kafka-Eagle監控工具仍然可以正常監控Kafka2.8，當然KRaft模式的兼容，部落客在Kafka-Eagle的測試分支中在進行開發，當Kafka官網將KRaft切換為主線版本時，Kafka-Eagle也會及時切換KRaft作為主線版本來監控Kafka。最後，歡迎大家使用Kafka-Eagle監控工具。
5.結束語
這篇部落格就和大家分享到這裡，如果大家在研究學習的過程當中有什麼問題，可以加群進行討論或發送郵件給我，我會盡我所能為您解答，與君共勉！
另外，部落客出書了《Kafka並不難學》和《Hadoop大數據挖掘從入門到進階實戰》，喜歡的朋友或同學， 可以在公告欄那裡點擊購買鏈接購買部落客的書進行學習，在此感謝大家的支援。關注下面公眾號，根據提示，可免費獲取書籍的教學影片。
