Kafka中如何刪除topic

小編給大家分享一下Kafka中如何刪除topic，相信大部分人都還不怎么了解，因此分享這篇文章給大家參考一下，希望大家閱讀完這篇文章后大有收獲，下面讓我們一起去了解一下吧！

前言

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

該類代表分區的狀態機。決定者分區的當前狀態，和狀態轉移。四種狀態

• NonExistentPartition

• NewPartition

• OnlinePartition

• OfflinePartition

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 --= nonExistentTopics
if(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 created
partitionStateMachine.deregisterPartitionChangeListener(topic)
val replicasForDeletedTopic = controller.replicaStateMachine.replicasInState(topic, ReplicaDeletionSuccessful)
// controller will remove this replica from the state machine as well as its partition assignment cache
replicaStateMachine.handleStateChanges(replicasForDeletedTopic, NonExistentReplica)
val partitionsForDeletedTopic = controllerContext.partitionsForTopic(topic)
// move respective partition to OfflinePartition and NonExistentPartition state
partitionStateMachine.handleStateChanges(partitionsForDeletedTopic, OfflinePartition)
partitionStateMachine.handleStateChanges(partitionsForDeletedTopic, NonExistentPartition)
topicsToBeDeleted -= topic
partitionsToBeDeleted.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 leader
brokerRequestBatch.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 -- currentChildren
controllerContext.allTopics = currentChildren

val addedPartitionReplicaAssignment = ZkUtils.getReplicaAssignmentForTopics(zkClient, newTopics.toSeq)
controllerContext.partitionReplicaAssignment = controllerContext.partitionReplicaAssignment.filter(p =>

以上是“Kafka中如何刪除topic”這篇文章的所有內容，感謝各位的閱讀！相信大家都有了一定的了解，希望分享的內容對大家有所幫助，如果還想學習更多知識，歡迎關注億速云行業資訊頻道！