[#186] - delete topics from CLI

README.md

@@ -867,6 +867,26 @@ Option     Description
 ------     -----------
 --options  topic options. Example: flush.ms=60000,retention.ms=6000000
 
+Generic Options
+Option  Description
+------  -----------
+--api   Api url. Example: http://master:7000
+
+topic-expr examples:
+  t0        - topic t0
+  t0,t1     - topics t0, t1
+  *         - any topic
+  t*        - topics starting with 't'
+```
+
+Deleting Topic
+--------------
+```
+#./kafka-mesos.sh help topic delete
+delete topic
+Usage: topic delete <topic-expr>
+
+
 Generic Options
 Option  Description
 ------  -----------
@@ -1003,6 +1023,12 @@ Updating topic
 {"topic" : {"name" : "t", "partitions" : {"0" : "0, 1"}, "options" : {"flush.ms" : "1000"}}}
 ```
 
+Deleting topic
+```
+# curl "http://localhost:7000/api/topic/delete?topic=t"
+{"topic" : {"name" : "t", "partitions" : {"0" : "0, 1"}, "options" : {}}}
+```
+
 Project Goals
 ==============
 

src/scala/ly/stealth/mesos/kafka/Cli.scala

@@ -910,6 +910,7 @@ object Cli {
       cmd match {
         case "list" => handleList(arg, args)
         case "add" | "update" => handleAddUpdate(arg, args, cmd == "add")
+        case "delete" => handleDelete(arg, args)
         case "rebalance" => handleRebalance(arg, args)
         case "partitions" => handlePartitions(arg)
         case _ => throw new Error("unsupported topic command " + cmd)
@@ -928,6 +929,8 @@ object Cli {
           handleList(null, null, help = true)
         case "add" | "update" =>
           handleAddUpdate(null, null, cmd == "add", help = true)
+        case "delete" =>
+          handleDelete(null, null, help = true)
         case "rebalance" =>
           handleRebalance(null, null, help = true)
         case "partitions" =>
@@ -979,6 +982,37 @@ object Cli {
       }
     }
 
+    def handleDelete(name: String, args: Array[String], help: Boolean = false): Unit = {
+      if (help) {
+        printLine(s"delete topic\nUsage: topic delete <topic-expr>\n")
+        printLine()
+        handleGenericOptions(null, help = true)
+        printLine()
+        Expr.printTopicExprExamples(out)
+        return
+      }
+
+      val params = new util.LinkedHashMap[String, String]
+      params.put("topic", name)
+
+      var json: Map[String, Object] = null
+      try { json = sendRequest(s"/topic/delete", params) }
+      catch { case e: IOException => throw new Error("Failed to delete " + name + ": " + e) }
+
+      printLine("response: '" + json+"'")
+      val topicNodes = json("topics").asInstanceOf[List[Map[String, Object]]]
+
+      val title = s"topic${if (topicNodes.size > 1) "s" else ""} deleted:"
+      printLine(title)
+
+      for (topicNode <- topicNodes) {
+        val topic = new Topic()
+        topic.fromJson(topicNode)
+        printTopic(topic, 1)
+        printLine()
+      }
+    }
+
     def handleAddUpdate(name: String, args: Array[String], add: Boolean, help: Boolean = false): Unit = {
       val cmd = if (add) "add" else "update"
 
@@ -1144,6 +1178,7 @@ object Cli {
       printLine("list       - list topics", 1)
       printLine("add        - add topic", 1)
       printLine("update     - update topic", 1)
+      printLine("delete     - delete topic", 1)
       printLine("rebalance  - rebalance topics", 1)
       printLine("partitions - list partition details for a topic", 1)
     }

src/scala/ly/stealth/mesos/kafka/HttpServer.scala

@@ -521,6 +521,7 @@ object HttpServer {
 
       if (uri == "list") handleListTopics(request, response)
       else if (uri == "add" || uri == "update") handleAddUpdateTopic(request, response)
+      else if (uri == "delete") handleDeleteTopic(request, response)
       else if (uri == "rebalance") handleTopicRebalance(request, response)
       else response.sendError(404, "uri not found")
     }
@@ -540,6 +541,31 @@ object HttpServer {
       response.getWriter.println("" + new JSONObject(Map("topics" -> new JSONArray(topicNodes.toList))))
     }
 
+    def handleDeleteTopic(request: HttpServletRequest, response: HttpServletResponse): Unit = {
+      val topics: Topics = Scheduler.cluster.topics
+      val errors = new util.ArrayList[String]()
+
+      val topicExpr: String = request.getParameter("topic")
+      if (topicExpr == null || topicExpr.isEmpty) errors.add("topic required")
+      val topicNames: util.List[String] = Expr.expandTopics(topicExpr)
+
+      for (name <- topicNames) {
+        val topic = topics.getTopic(name)
+        if (topic == null) errors.add(s"Topic $name not found")
+      }
+
+      if (!errors.isEmpty) { response.sendError(400, errors.mkString("; ")); return }
+
+      val topicNodes= new ListBuffer[JSONObject]
+
+      for (name <- topicNames) {
+        topics.deleteTopic(topics.getTopic(name))
+        topicNodes.add(topics.getTopic(name).toJson)
+      }
+
+      response.getWriter.println(JSONObject(Map("topics" -> new JSONArray(topicNodes.toList))))
+    }
+
     def handleAddUpdateTopic(request: HttpServletRequest, response: HttpServletResponse): Unit = {
       val topics: Topics = Scheduler.cluster.topics
       val add: Boolean = request.getRequestURI.endsWith("add")

src/scala/ly/stealth/mesos/kafka/Scheduler.scala

@@ -33,7 +33,7 @@ import scala.collection.mutable
 
 object Scheduler extends org.apache.mesos.Scheduler {
   private val logger: Logger = Logger.getLogger(this.getClass)
-  val version: Version = new Version("0.9.5.1")
+  val version: Version = new Version("0.9.5.2")
 
   val cluster: Cluster = new Cluster()
   private var driver: SchedulerDriver = null

src/scala/ly/stealth/mesos/kafka/Topics.scala

@@ -135,6 +135,12 @@ class Topics {
     finally { zkClient.close() }
   }
 
+  def deleteTopic(topic: Topic): Unit = {
+    val zkClient = newZkClient
+    try { AdminUtils.deleteTopic(zkClient, topic.name) }
+    finally { zkClient.close() }
+  }
+
   def validateOptions(options: util.Map[String, String]): String = {
     val config: Properties = new Properties()
     for ((k, v) <- options) config.setProperty(k, v)