問題描述：

線上出現(xiàn)一臺服務器特別慢，于是關閉了服務器上的kafka broker. 關閉后發(fā)現(xiàn)一些kafka consumer無法正常消費數(shù)據(jù)了, 日志錯誤：
o.a.kakfa.clients.consumer.internals.AbstractCordinator Marking the coordinator (39.0.2.100) as dead.

在邱縣等地區(qū)，都構建了全面的區(qū)域性戰(zhàn)略布局，加強發(fā)展的系統(tǒng)性、市場前瞻性、產(chǎn)品創(chuàng)新能力，以專注、極致的服務理念，為客戶提供網(wǎng)站制作、成都網(wǎng)站制作 網(wǎng)站設計制作定制網(wǎng)站制作,公司網(wǎng)站建設,企業(yè)網(wǎng)站建設,品牌網(wǎng)站制作,網(wǎng)絡營銷推廣,成都外貿(mào)網(wǎng)站建設公司,邱縣網(wǎng)站建設費用合理。

原因：

經(jīng)過一番排查，發(fā)現(xiàn)consumer group信息:
(kafka.coordinator.GroupMetadataMessageFormatter類型):
groupId::[groupId,Some(consumer),groupState,Map(memberId -> [memberId,clientId,clientHost,sessionTimeoutMs], ...->[]...)],
存到了KAFKA內(nèi)部topic: __consumer_offsets里, , 它的key是 groupId.
同時發(fā)現(xiàn)broker 參數(shù) offsets.topic.replication.factor 錯誤地被設置為1. 這個參數(shù)表示TOPIC: __Consumer_offsets 的副本數(shù). 這樣一旦某個broker被關閉, 如果被關閉的Broker 是__Consumer_offsets的某些partition的Leader. 則導致某些consumer group 不可用. 如果一旦broker已經(jīng)啟動, 需要手工通過命令行來擴展副本數(shù).

reassignment.json:
{"version":1,
 "partitions": [{"topic": "xxx", "partition": 0, "replicas": {brokerId1, brokerId2}}]
}
kafka-reassign-partitions  --zookeeper localhost:2818 --reassignment-json-file  reassignment.json --execute

客戶端尋找Consumer Coordinator的過程:
客戶端 org.apache.kafka.clients.consumer.internals.AbstractCoordinator
如果Coordinator 未知 (AbstractCoordinator.coordinatorUnknown()), 發(fā)起請求 lookupCoordinator，向負載最低的節(jié)點發(fā)送FindCoordinatorRequest

服務端 KafkaApis.handleFindCoordinatorRequest 接收請求：
首先調(diào)用 GroupMetaManager.partitionFor(consumerGroupId) consunerGroupId 的 hashCode 對 __consumer_offsets 總的分片數(shù)取模獲取partition id 再從 __consumer_offset 這個Topic 中找到partition對應的 Partition Metadata, 并且獲取對應的Partition leader 返回給客戶端

引伸思考

KAFKA 的failover機制究竟是怎么樣的？假使 __consumer_offset 設置了正確的副本數(shù)，重選舉的過程是怎樣的. 如果broker宕機后導致某些副本不可用, 副本會自動遷移到其他節(jié)點嗎？帶著這些問題稍微閱讀了一下KAFKA的相關代碼:

當一個Broker 被關掉時, 會有兩步操作：
KafkaController.onBrokerFailure ->KafkaController.onReplicasBecomeOffline
主要是通過 PartitionStateMachine.handleStateChanges 方法通知Partition狀態(tài)機將狀態(tài)置為offline. ReplicaStateMachine.handleStateChanges方法會將Replica 狀態(tài)修改為OfflineReplica, 同時修改partition ISR. 如果被關閉broker 是partition leader 那么需要重新觸發(fā)partition leader 選舉，最后發(fā)送LeaderAndIsrRequest獲取最新的Leader ISR 信息.
KafkaController.unregisterBrokerModificationsHandler 取消注冊的BrokerModificationsHandler 并取消zookeeper 中broker 事件的監(jiān)聽.

當ISR請求被發(fā)出,KafkaApis.handleLeaderAndIsrRequest() 會被調(diào)用. 這里如果需要變更leader的partition是屬于__consumer_offset這個特殊的topic,取決于當前的broker節(jié)點是不是partition leader. 會分別調(diào)用GroupCoordinator.handleGroupImmigration 和 GroupCoordinator.handleGroupEmmigration. 如果是partition leader, GroupCoordinator.handleGroupImmigration -> GroupMetadataManager.loadGroupsForPartition 會重新從 __consumer_offset 讀取group數(shù)據(jù)到本地metadata cache, 如果是partition follower, GroupCoordniator.handleGroupImigration -> GroupMetadataManager.removeGroupsForPartition 會從metadata cache中移除group信息. 并在onGroupUnloaded回調(diào)函數(shù)中將group的狀態(tài)變更為dead. 同時通知所有等待join或者sync的組成員.

KAFKA在Broker關閉時不會自動做partition 副本的遷移. 這時被關閉的Broker上的副本變?yōu)閡nder replicated 狀態(tài). 這種狀態(tài)將持續(xù)直到Broker被重新拉起并且追上新的數(shù)據(jù), 或者用戶通過命令行 手動復制副本到其他節(jié)點.

官方建議設置兩個參數(shù)來保證graceful shutdown. controlled.shutdown.enable=true auto.leader.rebalance.enable=true前者保證關機之前將日志數(shù)據(jù)同步到磁盤，并進行重選舉. 后者保證在broker重新恢復后再次獲得宕機前l(fā)eader狀態(tài). 避免leader分配不均勻?qū)е伦x寫熱點.

Reference

https://blog.csdn.net/zhanglh046/article/details/72833129
https://blog.csdn.net/huochen1994/article/details/80511038
https://www.jianshu.com/p/1aba6e226763

另外有需要云服務器可以了解下創(chuàng)新互聯(lián)scvps.cn，海內(nèi)外云服務器15元起步，三天無理由+7*72小時售后在線，公司持有idc許可證，提供“云服務器、裸金屬服務器、高防服務器、香港服務器、美國服務器、虛擬主機、免備案服務器”等云主機租用服務以及企業(yè)上云的綜合解決方案，具有“安全穩(wěn)定、簡單易用、服務可用性高、性價比高”等特點與優(yōu)勢，專為企業(yè)上云打造定制，能夠滿足用戶豐富、多元化的應用場景需求。

當前題目：一次KAFKA消費者異常引起的思考-創(chuàng)新互聯(lián)
當前URL：http://aaarwkj.com/article28/jcsjp.html

成都網(wǎng)站建設公司_創(chuàng)新互聯(lián)，為您提供動態(tài)網(wǎng)站、面包屑導航、關鍵詞優(yōu)化、品牌網(wǎng)站制作、網(wǎng)站建設、云服務器

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)