Add v3 burrow API compatibility, drop v2 API compatibility

README.md

@@ -5,7 +5,7 @@ and report it to InfluxDB.
 
 **Works with:**
 * InfluxDB 0.9
-* Burrow latest (Commit SHA: [7930a61](https://github.com/linkedin/Burrow/commit/7930a61a3e72df5df8a59ccdf3158585b785762f))
+* Burrow latest (Commit SHA: [334be91](https://github.com/linkedin/Burrow/commit/334be9125dcb3d0b5cddbee3f69cb956419d5a9e))
 
 Burrower will traverse the Burrow API and retrieve lag metrics for all clusters and consumer groups that are available.
 

project/Dependencies.scala

@@ -6,7 +6,7 @@ object Dependencies {
   )
 
   object V {
-    val playJson = "2.5.2"
+    val playJson = "2.5.19"
   }
 
   object Libraries {

project/build.properties

@@ -0,0 +1 @@
+sbt.version=0.13.15

src/main/scala/com/github/splee/burrower/OffsetMonitor.scala

@@ -4,6 +4,7 @@ import com.github.splee.burrower.lag.{BurrowConsumerStatus, BurrowPartitionLag,
 import com.github.splee.burrower.write.{ConsoleWriter, InfluxWriter, Writer}
 import com.typesafe.config.{Config, ConfigFactory}
 import com.typesafe.scalalogging.LazyLogging
+import scala.util.matching.Regex
 import scalaj.http._
 import play.api.libs.json._
 
@@ -62,8 +63,8 @@ class OffsetMonitor (
   writer: Writer
 ) extends Runnable with LazyLogging {
 
-  val burrowBaseUrl = f"http://$burrowHost:$burrowPort/v2/kafka"
-
+  val burrowBaseUrl = f"http://$burrowHost:$burrowPort/v3/kafka"
+  logger.info(f"burrowBaseURL = '$burrowBaseUrl'")
   def run(): Unit = {
     while (true) {
       val lag = getLag
@@ -98,8 +99,10 @@ class OffsetMonitor (
       logger.error(f"Error retrieving consumers: '$errorMsg'")
       return None
     }
-
-    Option((respJson \ "consumers").as[List[String]])
+    val exclude_groups_re = """^(kmf-consumer-group|console-consumer)""".r
+    val consList = (respJson \ "consumers").as[List[String]]
+    val consListFiltered = consList.filter{ s => ! exclude_groups_re.findFirstIn(s).isDefined }
+    Option(consListFiltered)
   }
 
   def getLag(cluster: String, consumer: String): Option[List[Lag]] = {
@@ -110,16 +113,20 @@ class OffsetMonitor (
       logger.error(f"Error retrieving lag: '$errorMsg")
       return None
     }
-    val statusResponse = (respJson \ "status").as[BurrowConsumerStatus]
-    Option(statusResponse.partitions.map((item: BurrowPartitionLag) => {
+    val status = respJson \ "status"
+    logger.debug(f"respJSONStatus '$status'")
+    val statusResponse = status.as[BurrowConsumerStatus]
+    Option(statusResponse.partitions.filter((item: BurrowPartitionLag) => {
+      !item.end.isEmpty
+    }).map((item: BurrowPartitionLag) => {
       Lag(
         cluster,
         consumer,
         item.topic,
         item.partition,
-        item.end.offset,
-        item.end.timestamp,
-        item.end.lag
+        item.end.get.offset,
+        item.end.get.timestamp,
+        item.end.get.lag
       )
     }))
   }

src/main/scala/com/github/splee/burrower/lag/BurrowConsumerStatus.scala

@@ -10,10 +10,9 @@ case class BurrowConsumerStatus(
   cluster: String,
   group: String,
   status: String,
-  complete: Boolean,
+  complete: Float,
   partitions: List[BurrowPartitionLag],
   partition_count: Int,
   maxlag: Option[BurrowPartitionLag],
   totallag: Int
 )
-

src/main/scala/com/github/splee/burrower/lag/BurrowPartitionLag.scala

@@ -10,7 +10,6 @@ case class BurrowPartitionLag(
   topic: String,
   partition: Int,
   status: String,
-  start: BurrowLagValue,
-  end: BurrowLagValue
+  start: Option[BurrowLagValue],
+  end: Option[BurrowLagValue]
 )
-