Kafka源码系列之如何删除topic

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))

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)
}

controller.deleteTopicManager.markTopicIneligibleForDeletion(Set(topic))
controller.deleteTopicManager.enqueueTopicsForDeletion(topicsToBeDeleted)

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(",")))
}

def enqueueTopicsForDeletion(topics: Set[String]) {
 if(isDeleteTopicEnabled) {
 topicsToBeDeleted ++= topics
 partitionsToBeDeleted ++= topics.flatMap(controllerContext.partitionsForTopic)
    resumeTopicDeletionThread()
  }
}

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))
  }

// 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)

{
 // 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)
  }

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)

// send stop replica command to the replica so that it stops fetching from the leader
brokerRequestBatch.addStopReplicaRequestForBrokers(List(replicaId), topic, partition, deletePartition = false)

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()

case RequestKeys.StopReplicaKey => handleStopReplicaRequest(request)
val (response, error) = replicaManager.stopReplicas(stopReplicaRequest)

{
 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)
}

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
 }

val deletedTopics = controllerContext.allTopics -- currentChildren
controllerContext.allTopics = currentChildren

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