Report current_offset metric for incomplete partitions

For consumers of low volume topic-partitions, it can take a very long
time for enough offset commits to happen before a partition is marked as
"complete".

For the `burrow_kafka_consumer_current_offset` metric to be reported,
only a single committed offset is required.

Author: andpol

core/internal/evaluator/caching.go

@@ -281,13 +281,19 @@ func evaluatePartitionStatus(partition *protocol.ConsumerPartition, minimumCompl
 	}
 
 	// Slice the offsets to remove all nil entries (they'll be at the start)
-	firstOffset := len(partition.Offsets) - 1
+	firstOffset := -1
 	for i, offset := range partition.Offsets {
 		if offset != nil {
 			firstOffset = i
 			break
 		}
 	}
+
+	// If all the offsets are nil, we can't do anything
+	if firstOffset == -1 {
+		return status
+	}
+
 	offsets := partition.Offsets[firstOffset:]
 
 	// Check if we had any nil offsets, and mark the partition as incomplete

core/internal/httpserver/prometheus.go

@@ -137,8 +137,10 @@ func (hc *Coordinator) handlePrometheusMetrics() http.HandlerFunc {
 
 					consumerPartitionLagGauge.With(labels).Set(float64(partition.CurrentLag))
 
-					if partition.Complete == 1.0 {
+					if partition.Complete > 0.0 {
 						consumerPartitionCurrentOffset.With(labels).Set(float64(partition.End.Offset))
+					}
+					if partition.Complete == 1.0 {
 						partitionStatusGauge.With(labels).Set(float64(partition.Status))
 					}
 				}

core/internal/httpserver/prometheus_test.go

@@ -158,7 +158,7 @@ func TestHttpServer_handlePrometheusMetrics(t *testing.T) {
 	assert.Contains(t, promExp, `burrow_kafka_consumer_current_offset{cluster="testcluster",consumer_group="testgroup",partition="0",topic="testtopic"} 22663`)
 	assert.Contains(t, promExp, `burrow_kafka_consumer_current_offset{cluster="testcluster",consumer_group="testgroup",partition="1",topic="testtopic"} 2488`)
 	assert.Contains(t, promExp, `burrow_kafka_consumer_current_offset{cluster="testcluster",consumer_group="testgroup",partition="0",topic="testtopic1"} 99888`)
-	assert.NotContains(t, promExp, `burrow_kafka_consumer_current_offset{cluster="testcluster",consumer_group="testgroup",partition="0",topic="incomplete"} 5335`)
+	assert.Contains(t, promExp, `burrow_kafka_consumer_current_offset{cluster="testcluster",consumer_group="testgroup",partition="0",topic="incomplete"} 5335`)
 	assert.Contains(t, promExp, `burrow_kafka_consumer_current_offset{cluster="testcluster",consumer_group="testgroup",partition="1",topic="incomplete"} 99888`)
 
 	assert.Contains(t, promExp, `burrow_kafka_topic_partition_offset{cluster="testcluster",partition="0",topic="testtopic"} 6556`)