adding API to fetch consumers.

Author: Lokesh Balakrishnan

app/controllers/api/KafkaStateCheck.scala

@@ -118,4 +118,15 @@ class KafkaStateCheck (val messagesApi: MessagesApi, val kafkaManagerContext: Ka
     }
     Future.sequence(cosumdTopicSummary).map(_.toMap)
   }
+  
+  def consumersSummaryAction(cluster: String) = Action.async { implicit request =>
+    implicit val formats = org.json4s.DefaultFormats
+    kafkaManager.getConsumerListExtended(cluster).map { errorOrConsumersSummary =>
+      errorOrConsumersSummary.fold(
+        error => BadRequest(Json.obj("msg" -> error.msg)),
+        consumersSummary => Ok(Serialization.writePretty("consumers" -> consumersSummary.list.map{case ((consumer, consumerType), consumerIdentity) => (consumer, consumerType.toString())}.toMap))
+        )
+    }
+  }
+
 }

conf/routes

@@ -60,6 +60,7 @@ GET    /api/status/:c/:t/underReplicatedPartitions          controllers.api.Kafk
 GET    /api/status/:c/:t/unavailablePartitions              controllers.api.KafkaStateCheck.unavailablePartitions(c:String,t:String)
 GET    /api/status/:cluster/:consumer/:topic/topicSummary   controllers.api.KafkaStateCheck.topicSummaryAction(cluster:String, consumer:String, topic:String, consumerType:String)
 GET    /api/status/:cluster/:consumer/groupSummary          controllers.api.KafkaStateCheck.groupSummaryAction(cluster:String, consumer:String, consumerType:String)
+GET    /api/status/:cluster/consumersSummary                controllers.api.KafkaStateCheck.consumersSummaryAction(cluster:String)
 
 # Versioned Assets
 GET    /vassets/*file                                       controllers.Assets.versioned(path="/public", file: Asset)