Merge with upstream and fixing build

README.md

@@ -6,7 +6,7 @@ A tool for managing [Apache Kafka](http://kafka.apache.org).
 It supports the following :
 
  - Manage multiple clusters
- - Easy inspection of cluster state (topics, brokers, replica distribution, partition distribution)
+ - Easy inspection of cluster state (topics, consumers, offsets, brokers, replica distribution, partition distribution)
  - Run preferred replica election
  - Generate partition assignments with option to select brokers to use
  - Run reassignment of partition (based on generated assignments)
@@ -18,6 +18,7 @@ It supports the following :
  - Add partitions to existing topic
  - Update config for existing topic
  - Optionally enable JMX polling for broker level and topic level metrics.
+ - Optionally filter out consumers that do not have ids/ owners/ & offsets/ directories in zookeeper.
 
 Cluster Management
 
@@ -37,9 +38,21 @@ Topic View
 
 ***
 
+Consumer List View
+
+![consumer](/img/consumer-list.png)
+
+***
+
+Consumed Topic View
+
+![consumer](/img/consumed-topic.png)
+
+***
+
 Broker List
 
-![topic](/img/broker-list.png)
+![broker](/img/broker-list.png)
 
 ***
 
@@ -52,9 +65,9 @@ Broker View
 Requirements
 ------------
 
-1. [Kafka 0.8.1.1 or 0.8.2.0](http://kafka.apache.org/downloads.html)
+1. [Kafka 0.8.1.1 or 0.8.2.1](http://kafka.apache.org/downloads.html)
 2. [sbt 0.13.x](http://www.scala-sbt.org/download.html)
-3. Java 7+
+3. Java 8+
 
 Configuration
 -------------

app/controllers/Cluster.scala

@@ -70,7 +70,9 @@ object Cluster extends Controller {
       "zkHosts" -> nonEmptyText.verifying(validateZkHosts),
       "zkMaxRetry" -> ignored(100 : Int),
       "jmxEnabled" -> boolean,
-      "logkafkaEnabled" -> boolean
+      "filterConsumers" -> boolean,
+      "logkafkaEnabled" -> boolean,
+      "activeOffsetCacheEnabled" -> boolean
     )(ClusterConfig.apply)(ClusterConfig.customUnapply)
   )
 
@@ -93,7 +95,9 @@ object Cluster extends Controller {
       "zkHosts" -> nonEmptyText.verifying(validateZkHosts),
       "zkMaxRetry" -> ignored(100 : Int),
       "jmxEnabled" -> boolean,
-      "logkafkaEnabled" -> boolean
+      "filterConsumers" -> boolean,
+      "logkafkaEnabled" -> boolean,
+      "activeOffsetCacheEnabled" -> boolean
     )(ClusterOperation.apply)(ClusterOperation.customUnapply)
   )
 
@@ -154,7 +158,9 @@ object Cluster extends Controller {
             cc.curatorConfig.zkConnect,
             cc.curatorConfig.zkMaxRetry,
             cc.jmxEnabled,
-            cc.logkafkaEnabled))
+            cc.filterConsumers,
+            cc.logkafkaEnabled,
+            cc.activeOffsetCacheEnabled))
         }))
       }
     }
@@ -166,7 +172,7 @@ object Cluster extends Controller {
       clusterConfigForm.bindFromRequest.fold(
         formWithErrors => Future.successful(BadRequest(views.html.cluster.addCluster(formWithErrors))),
         clusterConfig => {
-          kafkaManager.addCluster(clusterConfig.name, clusterConfig.version.toString, clusterConfig.curatorConfig.zkConnect, clusterConfig.jmxEnabled, clusterConfig.logkafkaEnabled).map { errorOrSuccess =>
+          kafkaManager.addCluster(clusterConfig.name, clusterConfig.version.toString, clusterConfig.curatorConfig.zkConnect, clusterConfig.jmxEnabled, clusterConfig.filterConsumers, clusterConfig.logkafkaEnabled, clusterConfig.activeOffsetCacheEnabled).map { errorOrSuccess =>
             Ok(views.html.common.resultOfCommand(
               views.html.navigation.defaultMenu(),
               models.navigation.BreadCrumbs.withView("Add Cluster"),
@@ -225,7 +231,9 @@ object Cluster extends Controller {
               clusterOperation.clusterConfig.version.toString,
               clusterOperation.clusterConfig.curatorConfig.zkConnect,
               clusterOperation.clusterConfig.jmxEnabled,
-              clusterOperation.clusterConfig.logkafkaEnabled
+              clusterOperation.clusterConfig.filterConsumers,
+              clusterOperation.clusterConfig.logkafkaEnabled,
+              clusterOperation.clusterConfig.activeOffsetCacheEnabled
             ).map { errorOrSuccess =>
               Ok(views.html.common.resultOfCommand(
                 views.html.navigation.defaultMenu(),

app/controllers/Consumer.scala

@@ -0,0 +1,37 @@
+/**
+ * Copyright 2015 Yahoo Inc. Licensed under the Apache License, Version 2.0
+ * See accompanying LICENSE file.
+ */
+
+package controllers
+
+import features.ApplicationFeatures
+import play.api.mvc._
+
+/**
+ * @author cvcal
+ */
+object Consumer extends Controller{
+  import play.api.libs.concurrent.Execution.Implicits.defaultContext
+
+  private[this] val kafkaManager = KafkaManagerContext.getKafkaManager
+  private[this] implicit val af: ApplicationFeatures = ApplicationFeatures.features
+
+  def consumers(cluster: String) = Action.async {
+    kafkaManager.getConsumerListExtended(cluster).map { errorOrConsumerList =>
+      Ok(views.html.consumer.consumerList(cluster, errorOrConsumerList))
+    }
+  }
+
+  def consumer(cluster: String, consumerGroup: String) = Action.async {
+    kafkaManager.getConsumerIdentity(cluster,consumerGroup).map { errorOrConsumerIdentity =>
+      Ok(views.html.consumer.consumerView(cluster,consumerGroup,errorOrConsumerIdentity))
+    }
+  }
+
+  def consumerAndTopic(cluster: String, consumerGroup: String, topic: String) = Action.async {
+    kafkaManager.getConsumedTopicState(cluster,consumerGroup,topic).map { errorOrConsumedTopicState =>
+      Ok(views.html.consumer.consumedTopicView(cluster,consumerGroup,topic,errorOrConsumedTopicState))
+    }
+  }
+}

app/controllers/Logkafka.scala

@@ -233,4 +233,40 @@ object Logkafka extends Controller{
       )
     }
   }
+
+  def handleEnableConfig(clusterName: String, hostname: String, log_path: String) = Action.async { implicit request =>
+    clusterFeatureGate(clusterName, KMLogKafkaFeature) { clusterContext =>
+      implicit val clusterFeatures = clusterContext.clusterFeatures
+      val props = new Properties();
+      props.put("valid", true.toString);
+      kafkaManager.updateLogkafkaConfig(clusterName, hostname, log_path, props, false).map { errorOrSuccess =>
+        Ok(views.html.common.resultOfCommand(
+          views.html.navigation.clusterMenu(clusterName, "Logkafka", "Logkafka View", Menus.clusterMenus(clusterName)),
+          models.navigation.BreadCrumbs.withNamedViewAndClusterAndLogkafka("Logkafka View", clusterName, hostname, log_path, "Update Config"),
+          errorOrSuccess,
+          "Enable Config",
+          FollowLink("Go to logkafka view.", routes.Logkafka.logkafka(clusterName, hostname, log_path).toString()),
+          FollowLink("Try again.", routes.Logkafka.updateConfig(clusterName, hostname, log_path).toString())
+        ))
+      }
+    }
+  }
+
+  def handleDisableConfig(clusterName: String, hostname: String, log_path: String) = Action.async { implicit request =>
+    clusterFeatureGate(clusterName, KMLogKafkaFeature) { clusterContext =>
+      implicit val clusterFeatures = clusterContext.clusterFeatures
+      val props = new Properties();
+      props.put("valid", false.toString);
+      kafkaManager.updateLogkafkaConfig(clusterName, hostname, log_path, props, false).map { errorOrSuccess =>
+        Ok(views.html.common.resultOfCommand(
+          views.html.navigation.clusterMenu(clusterName, "Logkafka", "Logkafka View", Menus.clusterMenus(clusterName)),
+          models.navigation.BreadCrumbs.withNamedViewAndClusterAndLogkafka("Logkafka View", clusterName, hostname, log_path, "Update Config"),
+          errorOrSuccess,
+          "Disable Config",
+          FollowLink("Go to logkafka view.", routes.Logkafka.logkafka(clusterName, hostname, log_path).toString()),
+          FollowLink("Try again.", routes.Logkafka.updateConfig(clusterName, hostname, log_path).toString())
+        ))
+      }
+    }
+  }
 }

app/controllers/ReassignPartitions.scala

@@ -17,6 +17,8 @@ import play.api.data.Forms._
 import play.api.data.validation.{Valid, Invalid, Constraint}
 import play.api.mvc._
 
+import scala.collection.mutable
+
 import scala.concurrent.Future
 import scalaz.{\/, \/-, -\/}
 
@@ -204,7 +206,8 @@ object ReassignPartitions extends Controller{
             bl <- blOrError
           } yield {
             Ok(views.html.topic.manualAssignments(
-              c, t, manualReassignmentForm.fill(List(flattenTopicIdentity(ti))), bl, bv, manualReassignmentForm.errors
+              //c, t, manualReassignmentForm.fill(List(flattenTopicIdentity(ti))), bl, bv, manualReassignmentForm.errors
+              c, t, List(flattenTopicIdentity(ti)), bl, bv, manualReassignmentForm.errors
             ))
           }
           errorOrResult.fold(err => {
@@ -227,18 +230,19 @@ object ReassignPartitions extends Controller{
         }, { topics: TopicListExtended =>
           kafkaManager.getBrokerList(c).flatMap { errOrCV =>
             errOrCV.fold(
-            { err: ApiError =>
-              Future.successful(Ok(views.html.topic.confirmMultipleAssignments(c, -\/(err))))
-            }, { brokers: BrokerListExtended => {
-              brokersViews.flatMap { errorOrBVs =>
-                errorOrBVs.fold(
-                { err: ApiError => Future.successful(Ok(views.html.topic.confirmMultipleAssignments(c, -\/(err))))}, { bVs: Seq[BVView] => Future {
-                  Ok(views.html.topic.manualMultipleAssignments(
-                    c, manualReassignmentForm.fill(flattenedTopicListExtended(topics)), brokers, bVs, manualReassignmentForm.errors
-                  ))
-                }
-                }
-                )
+            {err: ApiError =>
+              Future.successful( Ok(views.html.topic.confirmMultipleAssignments( c, -\/(err) )))
+            },
+            { brokers: BrokerListExtended => {
+                brokersViews.flatMap { errorOrBVs =>
+                  errorOrBVs.fold (
+                  {err: ApiError => Future.successful( Ok(views.html.topic.confirmMultipleAssignments( c, -\/(err) )))},
+                  {bVs => Future {
+                    Ok(views.html.topic.manualMultipleAssignments(
+                      c, flattenedTopicListExtended(topics), brokers , bVs, manualReassignmentForm.errors
+                    ))
+                  }}
+                  )
               }
             }
             }
@@ -283,7 +287,7 @@ object ReassignPartitions extends Controller{
         errors => kafkaManager.getClusterList.flatMap { errorOrClusterList =>
           responseScreen(
             "Manual Reassign Partitions Failure",
-            -\/(IndexedSeq(ApiError("There is something really wrong with your submitted data!")))
+            -\/(IndexedSeq(ApiError("There is something really wrong with your submitted data!\n\n" + errors.toString)))
           )
         },
         assignment => {
@@ -397,7 +401,7 @@ object ReassignPartitions extends Controller{
         ),
         cc =>
           reassignPartitionsForm.bindFromRequest.fold(
-            formWithErrors => Future.successful(BadRequest(views.html.topic.topicView(c, t, -\/(ApiError("Unknown operation!"))))),
+            formWithErrors => Future.successful(BadRequest(views.html.topic.topicView(c, t, -\/(ApiError("Unknown operation!")), None))),
             op => op match {
               case RunAssignment =>
                 implicit val clusterFeatures = cc.clusterFeatures

app/controllers/Topic.scala

@@ -139,8 +139,11 @@ object Topic extends Controller{
   }
 
   def topic(c: String, t: String) = Action.async {
-    kafkaManager.getTopicIdentity(c,t).map { errorOrTopicIdentity =>
-      Ok(views.html.topic.topicView(c,t,errorOrTopicIdentity))
+    val futureErrorOrTopicIdentity = kafkaManager.getTopicIdentity(c,t)
+    val futureErrorOrConsumerList = kafkaManager.getConsumersForTopic(c,t)
+
+    futureErrorOrTopicIdentity.zip(futureErrorOrConsumerList).map {case (errorOrTopicIdentity,errorOrConsumerList) =>
+      Ok(views.html.topic.topicView(c,t,errorOrTopicIdentity,errorOrConsumerList))
     }
   }
 
@@ -197,7 +200,8 @@ object Topic extends Controller{
           BadRequest(views.html.topic.topicView(
             clusterName,
             topic,
-            -\/(ApiError(formWithErrors.error("topic").map(_.toString).getOrElse("Unknown error deleting topic!")))))
+            -\/(ApiError(formWithErrors.error("topic").map(_.toString).getOrElse("Unknown error deleting topic!"))),
+            None))
         ),
         deleteTopic => {
           kafkaManager.deleteTopic(clusterName, deleteTopic.topic).map { errorOrSuccess =>

app/features/ApplicationFeature.scala

@@ -50,16 +50,27 @@ case class ApplicationFeatures(features: Set[ApplicationFeature])
 
 object ApplicationFeatures {
   import play.api.Play.current
+  private lazy val log = LoggerFactory.getLogger(classOf[ApplicationFeatures])
+
+  lazy val default : List[String] = List(
+    KMClusterManagerFeature,
+    KMTopicManagerFeature,
+    KMPreferredReplicaElectionFeature, 
+    KMReassignPartitionsFeature).map(_.getClass.getSimpleName)
 
   lazy val features = {
     getApplicationFeatures(play.api.Play.configuration.underlying)
   }
 
   def getApplicationFeatures(config: Config) : ApplicationFeatures = {
     import scala.collection.JavaConverters._
-    val configFeatures: List[String] = config.getStringList("application.features").asScala.toList
+    val configFeatures: Option[List[String]] = Try(config.getStringList("application.features").asScala.toList).toOption
+
+    if(configFeatures.isEmpty) {
+      log.warn(s"application.features not found in conf file, using default values $default")
+    }
 
-    val f = configFeatures.map(ApplicationFeature.from).flatten
+    val f = configFeatures.getOrElse(default).map(ApplicationFeature.from).flatten
     ApplicationFeatures(f.toSet)
   }
 }