Delete topic from CLI.

Closes mesos#186. Inspired by mesos#245

Author: plaflamme

README.md

@@ -896,6 +896,26 @@ topic-expr examples:
   t*        - topics starting with 't'
 ```
 
+Deleting Topics
+--------------
+Note: needs to be supported and enabled in the broker configuration (i.e.: `delete.topic.enable=true`).
+```
+#./kafka-mesos.sh help topic delete
+Delete topics
+Usage: topic delete [<topic-expr>]
+
+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'
+```
+
 Listing topic partition details
 -------------------------------
 ```

src/scala/iface/0_10/ly/stealth/mesos/kafka/interface/impl/AdminUtils.scala

@@ -41,6 +41,9 @@ class AdminUtils(zkUrl: String) extends AdminUtilsProxy {
     configs: Properties
   ): Unit = KafkaAdminUtils.changeTopicConfig(zkUtils, topic, configs)
 
+  override def deleteTopic(topicToDelete: String): Unit =
+    KafkaAdminUtils.deleteTopic(zkUtils, topicToDelete)
+
   override def fetchEntityConfig(
     entityType: String,
     entity: String

src/scala/iface/0_8/ly/stealth/mesos/kafka/interface/impl/AdminUtils.scala

@@ -41,6 +41,9 @@ class AdminUtils(zkUrl: String) extends AdminUtilsProxy {
     configs: Properties
   ): Unit = KafkaAdminUtils.changeTopicConfig(zkClient, topic, configs)
 
+  override def deleteTopic(topicToDelete: String): Unit =
+    KafkaAdminUtils.deleteTopic(zkClient, topicToDelete)
+
   override def fetchEntityConfig(
     entityType: String,
     entity: String

src/scala/iface/0_9/ly/stealth/mesos/kafka/interface/impl/AdminUtils.scala

@@ -41,6 +41,9 @@ class AdminUtils(zkUrl: String) extends AdminUtilsProxy {
     configs: Properties
   ): Unit = KafkaAdminUtils.changeTopicConfig(zkUtils, topic, configs)
 
+  override def deleteTopic(topicToDelete: String): Unit =
+    KafkaAdminUtils.deleteTopic(zkUtils, topicToDelete)
+
   override def fetchEntityConfig(
     entityType: String,
     entity: String

src/scala/iface/base/ly/stealth/mesos/kafka/interface/AdminUtilsProxy.scala

@@ -35,6 +35,8 @@ trait AdminUtilsProxy {
 
   def changeTopicConfig(topic: String, configs: Properties)
 
+  def deleteTopic(topic: String)
+
   def fetchEntityConfig(entityType: String, entity: String): Properties
 
   def changeClientIdConfig(clientId: String, configs: Properties)

src/scala/main/ly/stealth/mesos/kafka/cli/TopicCli.scala

@@ -53,6 +53,7 @@ trait TopicCli {
       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 "realign" => handleRealign(arg, args)
         case "partitions" => handlePartitions(arg)
@@ -72,6 +73,8 @@ trait TopicCli {
           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 "realign" =>
@@ -136,6 +139,40 @@ trait TopicCli {
       }
     }
 
+    def handleDelete(expr: String, args: Array[String], help: Boolean = false): Unit = {
+      val parser = newParser()
+
+      if (help) {
+        printLine("Delete topics\nUsage: topic delete [<topic-expr>]\n")
+        handleGenericOptions(null, help = true)
+
+        printLine()
+        printTopicExprExamples()
+
+        return
+      }
+
+      val params = new util.LinkedHashMap[String, String]
+      if (expr != null) params.put("topic", expr)
+
+      val json =
+        try {
+          sendRequestObj[ListTopicsResponse]("/topic/delete", params)
+        }
+        catch {
+          case e: IOException => throw new Error("" + e)
+        }
+
+      val topics = json.topics
+      val title = s"topic${if (topics.size > 1) "s" else ""} deleted:"
+      printLine(title)
+
+      topics.foreach { topic =>
+        printTopic(topic, 1)
+        printLine()
+      }
+    }
+
     def handleAddUpdate(
       name: String,
       args: Array[String],
@@ -362,6 +399,7 @@ trait TopicCli {
       printLine("Commands:")
       printLine("list       - list topics", 1)
       printLine("add        - add topic", 1)
+      printLine("delete     - delete topic", 1)
       printLine("update     - update topic", 1)
       printLine("rebalance  - rebalance topics", 1)
       printLine("realign    - realign topics, keeping existing broker assignments where possible", 1)

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

@@ -83,6 +83,12 @@ class Topics {
     AdminUtilsWrapper().assignReplicasToBrokers(brokers_, partitions, replicas, fixedStartIndex, startPartitionId)
   }
 
+  def deleteTopic(name: String): Topic = {
+    val topicToDelete = getTopic(name)
+    if(topicToDelete != null) AdminUtilsWrapper().deleteTopic(topicToDelete.name)
+    topicToDelete
+  }
+
   def addTopic(name: String, assignment: Map[Int, Seq[Int]] = null, options: Map[String, String] = null): Topic = {
     val config = new Properties()
     if (options != null)

src/scala/main/ly/stealth/mesos/kafka/scheduler/http/api/TopicApi.scala

@@ -59,6 +59,27 @@ trait TopicApiComponentImpl extends TopicApiComponent {
       @DefaultValue("*") @BothParam("topic") expr: String
     ) = list(expr)
 
+    @Path("delete")
+    @POST
+    @Produces(Array(MediaType.APPLICATION_JSON))
+    def deletePost(
+      @BothParam("topic") topicExpr: String
+    ): Response = {
+      if (topicExpr == null)
+        return Status.BadRequest("topic required")
+
+      val names = Expr.expandTopics(topicExpr).toSet
+
+      val missing = names.filter(cluster.topics.getTopic(_) == null)
+      if(missing.nonEmpty) {
+        Status.BadRequest(s"topic${if (missing.size > 1) "s" else ""} not found ${missing.mkString(",")}")
+      } else {
+        Response.ok(ListTopicsResponse(
+          names.toSeq.map(cluster.topics.deleteTopic(_))
+        )).build()
+      }
+    }
+
     @Path("{op: (add|update)}")
     @POST
     @Produces(Array(MediaType.APPLICATION_JSON))