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的删除过程,实际上就是基于zookeeper做了一个订阅发布系统。zookeeper的客户端创建一个节点/admin/delete_topics/<topic>,由kafka controller监听到事件之后正式触发topic的删除:解除partition变更监听的listener,清除内存数据结构,删除副本数据,删除topic的相关zookeeper节点。
delete.topic.enable配置该参数为false的情况下执行了topic的删除命令,实际上未做任何动作。我们此时要彻底删除topic建议修改该参数为true,重启kafka,这样topic信息会被彻底删除,已经测试。
一般流行的做法是手动删除zookeeper的topic相关信息及磁盘数据但是这样的话会造成部分内存数据未清除。至于是否会有隐患,未测试。
好了,以上就是这篇文章的全部内容了,希望本文的内容对大家的学习或者工作具有一定的参考学习价值,如果有疑问大家可以留言交流,谢谢大家对的支持。
下一篇: JavaScript异步函数同步方法
推荐阅读