Added a new API endpoint under '/topic/$TOPIC/consumers' that shows t…

…he active

consumers information for the current topic.

Changelog:
  - Changed endpoint resource from '/topiclist/$TOPIC' to '/topic/$TOPIC'
    so changed app.js and templates to match this change
  - Created a new case class 'TopicAndConsumersDetails' which is the
    return for the new method created
  - Added new method: 'getTopicAndConsumersDetail()' which returns the
    topic details plus the consumer details for each consumer of the topic
  - Added the name into the 'KafkaInfo' case class to provide a little bit
    of context
  - Added the new endpoint to the 'OffsetGetterWeb' intent() method with
    path: '/topic/$TOPIC/consumers'

src/main/resources/offsetapp/scripts/app.js

@@ -26,7 +26,7 @@ var app = angular.module('offsetapp',
                                                               templateUrl: "views/topiclist.html",
                                                               controller: "TopicListCtrl"
                                                           })
-                                                         .when("/topicdetail/:group", {
+                                                         .when("/topic/:group", {
                                                               templateUrl: "views/topic-detail.html",
                                                               controller: "TopicDetailCtrl"
                                                          });;
@@ -65,7 +65,7 @@ angular.module("offsetapp.services", ["ngResource"])
 				return $resource("./group/:group").get({group:group}, groupPartitions(cb));
 			},
 			topicDetail: function(group, cb) {
-            	return $resource("./topicdetails/:group").get({group:group}, groupPartitions(cb));
+            	return $resource("./topic/:group").get({group:group}, groupPartitions(cb));
             },
             loadClusterViz: function(group, cb) {
                 cb(loadViz("#dataviz-container", "./clusterlist"))

src/main/resources/offsetapp/views/topiclist.html

@@ -7,5 +7,5 @@ <h1>Please select the topic you would like to view consumers on</h1>
 </div>
 
 <ul class="list-group">
-    <li ng-repeat="g in topics" class="list-group-item"><a href="./#/topicdetail/{{g}}">{{g}}</a></li>
+    <li ng-repeat="g in topics" class="list-group-item"><a href="./#/topic/{{g}}">{{g}}</a></li>
 </ul>

src/main/scala/com/quantifind/kafka/OffsetGetter.scala

@@ -23,6 +23,8 @@ case class Node(name: String, children: Seq[Node] = Seq())
 
 case class TopicDetails(consumers: Seq[ConsumerDetail])
 
+case class TopicAndConsumersDetails(consumers: Seq[KafkaInfo])
+
 case class ConsumerDetail(name: String)
 
 class OffsetGetter(zkClient: ZkClient) extends Logging {
@@ -122,6 +124,7 @@ class OffsetGetter(zkClient: ZkClient) extends Logging {
     val off = offsetInfo(group, topics)
     val brok = brokerInfo()
     KafkaInfo(
+      name = group,
       brokers = brok.toSeq,
       offsets = off
     )
@@ -155,6 +158,23 @@ class OffsetGetter(zkClient: ZkClient) extends Logging {
     }
   }
 
+  /**
+   * returns details for a given topic such as the active consumers pulling off of it
+   * and for each of the active consumers it will return the consumer data
+   *
+   * @param topic
+   * @return
+   */
+  def getTopicAndConsumersDetail(topic: String): TopicAndConsumersDetails = {
+    val topicMap = getActiveTopicMap
+
+    if (topicMap.contains(topic)) {
+      TopicAndConsumersDetails(topicMap(topic).map(getInfo(_, Seq(topic))))
+    } else {
+      TopicAndConsumersDetails(Seq())
+    }
+  }
+
   def getTopics: Seq[String] = {
     try {
       ZkUtils.getChildren(zkClient, ZkUtils.BrokerTopicsPath).sortWith(_ < _)
@@ -227,7 +247,7 @@ class OffsetGetter(zkClient: ZkClient) extends Logging {
 
 object OffsetGetter {
 
-  case class KafkaInfo(brokers: Seq[BrokerInfo], offsets: Seq[OffsetInfo])
+  case class KafkaInfo(name: String, brokers: Seq[BrokerInfo], offsets: Seq[OffsetInfo])
 
   case class BrokerInfo(id: Int, host: String, port: Int)
 

src/main/scala/com/quantifind/kafka/offsetapp/OffsetGetterWeb.scala

@@ -109,6 +109,10 @@ object OffsetGetterWeb extends UnfilteredWebApp[OWArgs] with Logging {
     _.getTopicDetail(topic)
   }
 
+  def getTopicAndConsumersDetail(topic: String, args: OWArgs) = withOG(args) {
+    _.getTopicAndConsumersDetail(topic)
+  }
+
   def getClusterViz(args: OWArgs) = withOG(args) {
     _.getClusterViz
   }
@@ -152,8 +156,10 @@ object OffsetGetterWeb extends UnfilteredWebApp[OWArgs] with Logging {
         JsonContent ~> ResponseString(write(getTopics(args)))
       case GET(Path(Seg("clusterlist" :: Nil))) =>
         JsonContent ~> ResponseString(write(getClusterViz(args)))
-      case GET(Path(Seg("topicdetails" :: group :: Nil))) =>
-        JsonContent ~> ResponseString(write(getTopicDetail(group, args)))
+      case GET(Path(Seg("topic" :: topic :: Nil))) =>
+        JsonContent ~> ResponseString(write(getTopicDetail(topic, args)))
+      case GET(Path(Seg("topic" :: topic :: "consumers" :: Nil))) =>
+        JsonContent ~> ResponseString(write(getTopicAndConsumersDetail(topic, args)))
       case GET(Path(Seg("activetopics" :: Nil))) =>
         JsonContent ~> ResponseString(write(getActiveTopics(args)))
     }