Kafka Manager 簡介
- Kafka Manager 可能是現(xiàn)在能找到的最好的可視化的Kafka管理工具, 感謝Yahoo-我人生中打開的一個(gè)網(wǎng)站-的開源;
- 使用Kafka Manager, 基本上之前需要運(yùn)行Kafka相應(yīng)命令行工具的工作現(xiàn)在都可以可視化的完成:
- 創(chuàng)建Topic, 調(diào)整消息保存時(shí)長, Partition數(shù)量等等配置;
- 管理Topic, 包括Reassign Partitions, Preferred Replica Election等等;
- 消費(fèi)情況查看, 支持offset保存到zk和broker兩種方式, 列出所有消費(fèi)的group, 消費(fèi)每個(gè)partition的詳情;
- 集群的簡單健康狀態(tài)查看,包括partition分布是否均衡, leader分布是否均衡等;
- 通過JMX查看各種指標(biāo), 比如各個(gè)broker的網(wǎng)絡(luò)流量和消息進(jìn)出數(shù)據(jù), 每個(gè)Topic消息的讀寫速度等;
- 下面我們會先簡單介紹下Kafka Manager的實(shí)現(xiàn)和在使用中遇到的幾種坑;
Kafka Manager實(shí)現(xiàn)
- 實(shí)現(xiàn)語言: Scala
- 用到的框架和第三方庫:
- Play framework: Kafka-Mananger本質(zhì)上是個(gè)Web應(yīng)用, 因此使用play framework的MVC架構(gòu)實(shí)現(xiàn);
- AKKA: 用于構(gòu)建高并發(fā)、分布式和容錯(cuò)的應(yīng)用. Kafka Manager中的所有請求都使用akka來異步處理;
- Apache Curator Framework: 用于訪問zookeeper;
- Kafka Sdk: 用于獲取各Topic的last offset, 使用Admin接口實(shí)現(xiàn)各種管理功能;
- 編譯:
整個(gè)工程使用 sbt 構(gòu)建, 具體編譯流程可以在githut上找到. sbt在build過程中會加載很多第三方依賴, 這個(gè)在國內(nèi)有時(shí)會很慢, 各種同學(xué)各顯神通吧. - 實(shí)現(xiàn):
其實(shí)kafka manager的代碼還是很清晰易閱讀的, 如果熟悉scala和play的話應(yīng)該沒有難度. 不同本人也是現(xiàn)學(xué)現(xiàn)用, 好慚愧~~~. 咱們這里撿重點(diǎn)的說吧, 不分析具體代碼實(shí)現(xiàn),只講下實(shí)現(xiàn)的方法:
-
獲取集群中所有Topic
使用Curator訪問zk獲取,并監(jiān)聽zk相關(guān)節(jié)點(diǎn) /brokers/topics 的變化; -
獲取Topic的partiton, leader, replicas信息
也是從zk獲取, /brokers/topics/[topic]/partitions; -
獲取Topic的各partition的last offset
使用kafka sdk發(fā)送OffsetRequest到kafka集群來獲得, 這個(gè)獲取的動作會被封裝成Future[PartitionOffsetsCapture], 每個(gè)topic一個(gè)Future, 使用Google的LoadingCache來存儲這些future, 利用LoadingCache的超時(shí)淘汰機(jī)制來周期性的創(chuàng)建新的Future來間隔地發(fā)送OffsetRequest獲取當(dāng)前最新的last offset; -
獲取Kafka本身管理的group的消費(fèi)情況
使用kafka sdk不斷地消費(fèi)"__consumer_offsets"這個(gè)topic, 來獲取所有g(shù)roup的消費(fèi)情況,關(guān)于__consumer_offsets參考 Committing and fetching consumer offsets in Kafka -
獲取zookeeper管理的group的消費(fèi)情況
肯定是從zk上讀取, /consumers
上面的這些實(shí)現(xiàn)都在 KafkaStateActor.scala 這個(gè)文件里.
- 各種Acotr的關(guān)系簡圖,僅供參考
Kafka Manager遇到的坑
- 多個(gè)kafka manager來管理同一個(gè)kafka集群:
你會發(fā)現(xiàn)在kafka manager里無法看到所有offset使用kafka本身管理的group.
前面我們講過使用kafka sdk不斷地消費(fèi)"__consumer_offsets", 看看這段代碼(在KafkaStateActor.scala中):
props.put("group.id", "KafkaManagerOffsetCache")
props.put("bootstrap.servers", bootstrapBrokerList.list.map(bi => s"${bi.host}:${bi.port}").mkString(","))
props.put("exclude.internal.topics", "false")
props.put("enable.auto.commit", "false")
props.put("key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer")
props.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer")
props.put("auto.offset.reset", "latest")
props.put("group.id", "KafkaManagerOffsetCache")這句說明不管啟動了幾個(gè)kafka manager, 消費(fèi)"__consumer_offsets"都使用同一個(gè)group.
解決方案: group.id從配置文件中讀取,每個(gè)kafka manager使用不同的group id;
- 客戶端使用某些sdk(比如librdkafka)消費(fèi)topic, 客戶端crash后, 在kafka manager上查看其group的消費(fèi)情況, 仍然一直能看到"Consumer Instance Owner"
原因在于處理從broker返回的GroupMetadata response時(shí)沒有處理異常情況:
case GroupMetadataKey(version, key) =>
val value: GroupMetadata = readGroupMessageValue(key, ByteBuffer.wrap(record.value()))
value.allMemberMetadata.foreach {
mm =>
mm.assignment.foreach {
case (topic, part) =>
groupTopicPartitionMemberMap += (key, topic, part) -> mm
}
}
}
這里的record.value可能為空, 此時(shí)應(yīng)作清理工作:
if (null != record &&
null != record.value()) {
val value: GroupMetadata = readGroupMessageValue(key, ByteBuffer.wrap(record.value()))
value.allMemberMetadata.foreach {
mm =>
mm.assignment.foreach {
case (topic, part) =>
groupTopicPartitionMemberMap += (key, topic, part) -> mm
}
}
} else {
groupTopicPartitionMemberMap.foreach {
case ((group, topic, part), mmd) =>
if (group == key) {
var tmp = mmd
tmp.memberId = ""
tmp.clientHost = ""
groupTopicPartitionMemberMap += (key, topic, part) -> tmp
}
}
}
- Yikes! Ask timed out on [ActorSelection[Anchor(akka://kafka-manager-system/), Path(/user/kafka-manager)]] after [5000 ms]
訪問kafka manager時(shí)出現(xiàn)上面的超時(shí)提示, 遇到這個(gè)問題,好學(xué)不服輸?shù)哪憧隙〞暇W(wǎng)各種搜, 然后你會去改kafka manager的各種配置, 調(diào)大各種thread pool的容量, 增大queue size, 甚至開大jvm的使用內(nèi)存, 然而問題并沒有解決, 看來只剩下定時(shí)重啟這一招兒了.
這里提供一種解決方案: 這個(gè)超時(shí)是Actor在執(zhí)行異步請求時(shí)一直等不到返回結(jié)果造成的, 主要是前面講過的"獲取Topic的各partition的last offset的Future"沒有返回結(jié)果,這些Future是通過Await.ready來阻塞拿到result的, 然而在kafka manager中這個(gè)Await.ready沒有給timeout, 是一直等待, 那咱們就給個(gè)timeout好了, 代碼在ActorModel.scala中, 有好幾處Await.ready的調(diào)用.
找到根源: 再也不用定時(shí)重啟, 提了一個(gè)pull request到官方:Use a separate thread to get the topic offsets to fixed bug 'Yikes! Ask timed out...', 主要就是不再使用 Future[PartitionOffsetCapture]來獲取topic offset, 因?yàn)檫@個(gè)會產(chǎn)生大量的Future, 進(jìn)而會產(chǎn)生大量的task提交到ThreadExcutor, 其實(shí)只需要啟動一個(gè)單獨(dú)的線程來作這件事就好了.
Consumer offset的詳情不完整
通過上面的源碼分析我們知道km是通過消費(fèi)"__consumer_offsets"來獲取某一個(gè)組的消費(fèi)情況的,消費(fèi)這個(gè)topic,和消費(fèi)用戶自己的topic沒什么兩樣, km里使用"props.put("auto.offset.reset", "latest")"默認(rèn)offset無效時(shí)從最新位置來拉取, 如果一個(gè)group用戶已經(jīng)有段時(shí)間沒有提交offset(但還沒有完全過期), 則此時(shí)在km上看不到相應(yīng)的gorup信息, 可以簡單改為"props.put("auto.offset.reset", "earliest")"同名group消費(fèi)不同topic后,其中一個(gè)group的消費(fèi)進(jìn)程結(jié)束后, 仍可以看到其消費(fèi)詳情
這個(gè)問題是最近被發(fā)現(xiàn),之前應(yīng)該是一直存在著,沒能引起重視.
這里提供一種簡單的,hack的解決方案:
case GroupMetadataKey(version, key) =>
if (null != record &&
null != record.value()) {
val value: GroupMetadata = readGroupMessageValue(key, ByteBuffer.wrap(record.value()))
var topicSet:Set[String] = Set()
value.allMemberMetadata.foreach {
mm =>
mm.assignment.foreach {
case (topic, part) =>
topicSet += topic
groupTopicPartitionMemberMap += (key, topic, part) -> mm
}
}
groupTopicPartitionMemberMap.foreach {
case ((group, topic, part), mmd) =>
if (group == key &&
!topicSet.contains(topic)) {
var tmp = mmd
tmp.memberId = ""
tmp.clientHost = ""
groupTopicPartitionMemberMap += (key, topic, part) -> tmp
}
}
} else {
groupTopicPartitionMemberMap.foreach {
case ((group, topic, part), mmd) =>
if (group == key) {
var tmp = mmd
tmp.memberId = ""
tmp.clientHost = ""
groupTopicPartitionMemberMap += (key, topic, part) -> tmp
}
}
}
}
