Kafka源码系列教程之删除topic

bin/kafka-topics.sh --zookeeper zk_host:port/chroot --delete --topic my_topic_name

exec $(dirname $ 0 )/kafka-run- class .sh kafka.admin.topiccommand $@

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

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)

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 )

brokerrequestbatch.sendrequeststobrokers(controller.epoch, controllercontext.correlationid.getandincrement)

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