前言
Apache Kafka發源于LinkedIn，于2011年成為Apache的孵化項目，隨后于2012年成為Apache的主要項目之一。Kafka使用Scala和Java進行編寫。Apache Kafka是一個快速、可擴展的、高吞吐、可容錯的分布式發布訂閱消息系統。Kafka具有高吞吐量、內置分區、支持數據副本和容錯的特性，適合在大規模消息處理場景中使用。
本文依然是以kafka0.8.2.2為例講解
一，如何刪除一個topic
刪除一個topic有兩個關鍵點：
1，配置刪除參數
delete.topic.enable這個Broker參數配置為True。
2，執行
bin/kafka-topics.sh --zookeeper zk_host:port/chroot --delete --topic my_topic_name
假如不配置刪除參數為true的話，topic其實并沒有被清除，只是被標記為刪除。此時，估計一般人的做法是刪除topic在Zookeeper的信息和日志，其實這個操作并不會清除kafkaBroker內存的topic數據。所以，此時最佳的策略是配置刪除參數為true然后，重啟kafka。
二，重要的類介紹
1，PartitionStateMachine
該類代表分區的狀態機。決定者分區的當前狀態，和狀態轉移。四種狀態
2，ReplicaManager
負責管理當前機器的所有副本，處理讀寫、刪除等具體動作。
讀寫：寫獲取partition對象，再獲取Replica對象，再獲取Log對象，采用其管理的Segment對象將數據寫入、讀出。
3，ReplicaStateMachine
副本的狀態機。決定者副本的當前狀態和狀態之間的轉移。一個副本總共可以處于一下幾種狀態的一種
NewReplica：Crontroller在分區重分配的時候可以創建一個新的副本。只能接受變為follower的請求。前狀態可以是NonExistentReplica
OnlineReplica：新啟動的分區，能接受變為leader或者follower請求。前狀態可以是NewReplica, OnlineReplica or OfflineReplica
OfflineReplica：死亡的副本處于這種狀態。前狀態可以是NewReplica, OnlineReplica
ReplicaDeletionStarted：分本刪除開始的時候處于這種狀態，前狀態是OfflineReplica
ReplicaDeletionSuccessful：副本刪除成功。前狀態是ReplicaDeletionStarted
ReplicaDeletionIneligible：刪除失敗的時候處于這種狀態。前狀態是ReplicaDeletionStarted
NonExistentReplica：副本成功刪除之后處于這種狀態，前狀態是ReplicaDeletionSuccessful
4，TopicDeletionManager
該類管理著topic刪除的狀態機
1)，TopicCommand通過創建/admin/delete_topics/<topic>，來發布topic刪除命令。
2)，Controller監聽/admin/delete_topic子節點變動，開始分別刪除topic
3)，Controller有個后臺線程負責刪除Topic
三，源碼徹底解析topic的刪除過程
此處會分四個部分：
A),客戶端執行刪除命令作用
B),不配置delete.topic.enable整個流水的源碼
C),配置了delete.topic.enable整個流水的源碼
D),手動刪除zk上topic信息和磁盤數據
1，客戶端執行刪除命令
bin/kafka-topics.sh --zookeeper zk_host:port/chroot --delete --topic my_topic_name
進入kafka-topics.sh我們會看到
exec $(dirname $0)/kafka-run-class.sh kafka.admin.TopicCommand $@
進入TopicCommand里面，main方法里面
else if(opts.options.has(opts.deleteOpt)) deleteTopic(zkClient, opts)
實際內容是
val topics = getTopics(zkClient, opts)if (topics.length == 0) { println("Topic %s does not exist".format(opts.options.valueOf(opts.topicOpt)))}topics.foreach { topic => try { ZkUtils.createPersistentPath(zkClient, ZkUtils.getDeleteTopicPath(topic))在"/admin/delete_topics"目錄下創建了一個topicName的節點。
2，假如不配置delete.topic.enable整個流水是
總共有兩處listener會響應：
A),TopicChangeListener
B),DeleteTopicsListener
使用topic的刪除命令刪除一個topic的話，指揮觸發DeleteTopicListener。
var topicsToBeDeleted = { import JavaConversions._ (children: Buffer[String]).toSet}val nonExistentTopics = topicsToBeDeleted.filter(t => !controllerContext.allTopics.contains(t))topicsToBeDeleted --= nonExistentTopicsif(topicsToBeDeleted.size > 0) { info("Starting topic deletion for topics " + topicsToBeDeleted.mkString(",")) // mark topic ineligible for deletion if other state changes are in progress topicsToBeDeleted.foreach { topic => val preferredReplicaElectionInProgress = controllerContext.partitionsUndergoingPreferredReplicaElection.map(_.topic).contains(topic) val partitionReassignmentInProgress = controllerContext.partitionsBeingReassigned.keySet.map(_.topic).contains(topic) if(preferredReplicaElectionInProgress || partitionReassignmentInProgress) controller.deleteTopicManager.markTopicIneligibleForDeletion(Set(topic)) } // add topic to deletion list controller.deleteTopicManager.enqueueTopicsForDeletion(topicsToBeDeleted)}由于都會判斷delete.topic.enable是否為true，假如不為true就不會執行，為true就進入執行
controller.deleteTopicManager.markTopicIneligibleForDeletion(Set(topic))controller.deleteTopicManager.enqueueTopicsForDeletion(topicsToBeDeleted)
3，delete.topic.enable配置為true
此處與步驟2的區別，就是那兩個處理函數。
controller.deleteTopicManager.markTopicIneligibleForDeletion(Set(topic))controller.deleteTopicManager.enqueueTopicsForDeletion(topicsToBeDeleted)
markTopicIneligibleForDeletion函數的處理為
if(isDeleteTopicEnabled) { val newTopicsToHaltDeletion = topicsToBeDeleted & topics topicsIneligibleForDeletion ++= newTopicsToHaltDeletion if(newTopicsToHaltDeletion.size > 0) info("Halted deletion of topics %s".format(newTopicsToHaltDeletion.mkString(",")))}主要是停止刪除topic，假如存儲以下三種情況
* Halt delete topic if -
* 1. replicas being down
* 2. partition reassignment in progress for some partitions of the topic
* 3. preferred replica election in progress for some partitions of the topic
enqueueTopicsForDeletion主要作用是更新刪除topic的集合，并激活TopicDeleteThread
def enqueueTopicsForDeletion(topics: Set[String]) { if(isDeleteTopicEnabled) { topicsToBeDeleted ++= topics partitionsToBeDeleted ++= topics.flatMap(controllerContext.partitionsForTopic) resumeTopicDeletionThread() }}在刪除線程DeleteTopicsThread的doWork方法中
topicsQueuedForDeletion.foreach { topic =>// if all replicas are marked as deleted successfully, then topic deletion is done if(controller.replicaStateMachine.areAllReplicasForTopicDeleted(topic)) { // clear up all state for this topic from controller cache and zookeeper completeDeleteTopic(topic) info("Deletion of topic %s successfully completed".format(topic)) }進入completeDeleteTopic方法中
// deregister partition change listener on the deleted topic. This is to prevent the partition change listener// firing before the new topic listener when a deleted topic gets auto createdpartitionStateMachine.deregisterPartitionChangeListener(topic)val replicasForDeletedTopic = controller.replicaStateMachine.replicasInState(topic, ReplicaDeletionSuccessful)// controller will remove this replica from the state machine as well as its partition assignment cachereplicaStateMachine.handleStateChanges(replicasForDeletedTopic, NonExistentReplica)val partitionsForDeletedTopic = controllerContext.partitionsForTopic(topic)// move respective partition to OfflinePartition and NonExistentPartition statepartitionStateMachine.handleStateChanges(partitionsForDeletedTopic, OfflinePartition)partitionStateMachine.handleStateChanges(partitionsForDeletedTopic, NonExistentPartition)topicsToBeDeleted -= topicpartitionsToBeDeleted.retain(_.topic != topic)controllerContext.zkClient.deleteRecursive(ZkUtils.getTopicPath(topic))controllerContext.zkClient.deleteRecursive(ZkUtils.getTopicConfigPath(topic))controllerContext.zkClient.delete(ZkUtils.getDeleteTopicPath(topic))controllerContext.removeTopic(topic)
主要作用是解除掉監控分區變動的listener，刪除Zookeeper具體節點信息，刪除磁盤數據，更新內存數據結構，比如從副本狀態機里面移除分區的具體信息。
其實，最終要的是我們的副本磁盤數據是如何刪除的。我們重點介紹這個部分。
首次清除的話，在刪除線程DeleteTopicsThread的doWork方法中
{ // if you come here, then no replica is in TopicDeletionStarted and all replicas are not in // TopicDeletionSuccessful. That means, that either given topic haven't initiated deletion // or there is at least one failed replica (which means topic deletion should be retried). if(controller.replicaStateMachine.isAnyReplicaInState(topic, ReplicaDeletionIneligible)) { // mark topic for deletion retry markTopicForDeletionRetry(topic) }進入markTopicForDeletionRetry
val failedReplicas = controller.replicaStateMachine.replicasInState(topic, ReplicaDeletionIneligible)info("Retrying delete topic for topic %s since replicas %s were not successfully deleted" .format(topic, failedReplicas.mkString(",")))controller.replicaStateMachine.handleStateChanges(failedReplicas, OfflineReplica)在ReplicaStateMachine的handleStateChanges方法中，調用了handleStateChange，處理OfflineReplica
// send stop replica command to the replica so that it stops fetching from the leaderbrokerRequestBatch.addStopReplicaRequestForBrokers(List(replicaId), topic, partition, deletePartition = false)
接著在handleStateChanges中
brokerRequestBatch.sendRequestsToBrokers(controller.epoch, controllerContext.correlationId.getAndIncrement)
給副本數據存儲節點發送StopReplicaKey副本指令，并開始刪除數據
stopReplicaRequestMap foreach { case(broker, replicaInfoList) => val stopReplicaWithDelete = replicaInfoList.filter(p => p.deletePartition == true).map(i => i.replica).toSet val stopReplicaWithoutDelete = replicaInfoList.filter(p => p.deletePartition == false).map(i => i.replica).toSet debug("The stop replica request (delete = true) sent to broker %d is %s" .format(broker, stopReplicaWithDelete.mkString(","))) debug("The stop replica request (delete = false) sent to broker %d is %s" .format(broker, stopReplicaWithoutDelete.mkString(","))) replicaInfoList.foreach { r => val stopReplicaRequest = new StopReplicaRequest(r.deletePartition, Set(TopicAndPartition(r.replica.topic, r.replica.partition)), controllerId, controllerEpoch, correlationId) controller.sendRequest(broker, stopReplicaRequest, r.callback) }}stopReplicaRequestMap.clear()Broker的KafkaApis的Handle方法在接受到指令后
case RequestKeys.StopReplicaKey => handleStopReplicaRequest(request)
val (response, error) = replicaManager.stopReplicas(stopReplicaRequest)
接著是在stopReplicas方法中
{ controllerEpoch = stopReplicaRequest.controllerEpoch // First stop fetchers for all partitions, then stop the corresponding replicas replicaFetcherManager.removeFetcherForPartitions(stopReplicaRequest.partitions.map(r => TopicAndPartition(r.topic, r.partition))) for(topicAndPartition <- stopReplicaRequest.partitions){ val errorCode = stopReplica(topicAndPartition.topic, topicAndPartition.partition, stopReplicaRequest.deletePartitions) responseMap.put(topicAndPartition, errorCode) } (responseMap, ErrorMapping.NoError)}進一步進入stopReplica方法，正式進入日志刪除
getPartition(topic, partitionId) match { case Some(partition) => if(deletePartition) { val removedPartition = allPartitions.remove((topic, partitionId)) if (removedPartition != null) removedPartition.delete() // this will delete the local log }以上就是kafka的整個日志刪除流水。
4，手動刪除zk上topic信息和磁盤數據
TopicChangeListener會監聽處理，但是處理很簡單，只是更新了
val deletedTopics = controllerContext.allTopics -- currentChildrencontrollerContext.allTopics = currentChildrenval addedPartitionReplicaAssignment = ZkUtils.getReplicaAssignmentForTopics(zkClient, newTopics.toSeq)controllerContext.partitionReplicaAssignment = controllerContext.partitionReplicaAssignment.filter(p =>
四，總結
Kafka的topic的刪除過程，實際上就是基于Zookeeper做了一個訂閱發布系統。Zookeeper的客戶端創建一個節點/admin/delete_topics/<topic>，由kafka Controller監聽到事件之后正式觸發topic的刪除：解除Partition變更監聽的listener，清除內存數據結構，刪除副本數據，刪除topic的相關Zookeeper節點。
delete.topic.enable配置該參數為false的情況下執行了topic的刪除命令，實際上未做任何動作。我們此時要徹底刪除topic建議修改該參數為true，重啟kafka，這樣topic信息會被徹底刪除，已經測試。
一般流行的做法是手動刪除Zookeeper的topic相關信息及磁盤數據但是這樣的話會造成部分內存數據未清除。至于是否會有隱患，未測試。
好了，以上就是這篇文章的全部內容了，希望本文的內容對大家的學習或者工作具有一定的參考學習價值，如果有疑問大家可以留言交流，謝謝大家對VeVb武林網的支持。
新聞熱點
疑難解答
圖片精選
