From a40362c95ca5534040d8c29a23b40168a9d70015 Mon Sep 17 00:00:00 2001
From: Guillaume Herail <guillaume@herail.net>
Date: Wed, 16 Jan 2019 18:03:51 +0100
Subject: [PATCH] fix(KafkaConsumerPartitionLag): Use proper current lag

Before this fix, the latest committed lag was used for the metric, it
now uses the current lag on the partition.

Fixes #26
---
 burrow_exporter/client.go   | 11 ++++++-----
 burrow_exporter/exporter.go |  2 +-
 2 files changed, 7 insertions(+), 6 deletions(-)

diff --git a/burrow_exporter/client.go b/burrow_exporter/client.go
index daaf84f..f4dc656 100644
--- a/burrow_exporter/client.go
+++ b/burrow_exporter/client.go
@@ -70,11 +70,12 @@ type ConsumerGroupStatus struct {
 }
 
 type Partition struct {
-	Topic     string `json:"topic"`
-	Partition int32  `json:"partition"`
-	Status    string `json:"status"`
-	Start     Offset `json:"start"`
-	End       Offset `json:"end"`
+	Topic      string `json:"topic"`
+	Partition  int32  `json:"partition"`
+	Status     string `json:"status"`
+	Start      Offset `json:"start"`
+	End        Offset `json:"end"`
+	CurrentLag int64  `json:"current_lag"`
 }
 
 type ConsumerGroupStatusResp struct {
diff --git a/burrow_exporter/exporter.go b/burrow_exporter/exporter.go
index 8b48b35..153ccbb 100644
--- a/burrow_exporter/exporter.go
+++ b/burrow_exporter/exporter.go
@@ -45,7 +45,7 @@ func (be *BurrowExporter) processGroup(cluster, group string) {
 				"group":     status.Status.Group,
 				"topic":     partition.Topic,
 				"partition": strconv.Itoa(int(partition.Partition)),
-			}).Set(float64(partition.End.Lag))
+			}).Set(float64(partition.CurrentLag))
 		}
 
 		if !be.skipPartitionCurrentOffset {