kafka删除topic的方法

werew

0.8的官方文档提供了一个删除topic的命令：
kafka-topics.sh --delete 但是在运行时会报错找不到这个方法。
kafka-topics.sh最终是运行了kafka.admin.TopicCommand这个类，在0.8的源码中这个类中没有找到有delete topic相关的代码。
在kafka的admin包下，提供了一个DeleteTopicCommand的类，可以实现删除topic的功能。
kafka.admin.DeleteTopicCommand
其中删除topic的具体实现代码如下：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20

import org.I0Itec.zkclient.ZkClient import kafka.utils.{Utils, ZKStringSerializer, ZkUtils} .......     val topic = options.valueOf(topicOpt)     val zkConnect = options.valueOf(zkConnectOpt)     var zkClient: ZkClient = null     try {       zkClient = new ZkClient(zkConnect, 30000, 30000, ZKStringSerializer)       zkClient.deleteRecursive(ZkUtils.getTopicPath(topic))  //其实最终还是通过删除zk里面对应的路径来实现删除topic的功能       println("deletion succeeded!")     }     catch {       case e: Throwable =>         println("delection failed because of " + e.getMessage)         println(Utils.stackTrace(e))     }     finally {       if (zkClient != null)         zkClient.close()     }

因为这个命令只会删除zk里面的信息，真实的数据还是没有删除，所以需要登录各个broker,把对应的topic的分区数据目录删除，也可能正因为这一点，delete命令才没有集成到kafka.admin.TopicCommand这个类。