Skip to content

Commit

Permalink
Report current_offset metric for incomplete partitions
Browse files Browse the repository at this point in the history
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.
  • Loading branch information
andpol committed Sep 27, 2024
1 parent 8690c58 commit 9ae3118
Show file tree
Hide file tree
Showing 3 changed files with 5 additions and 3 deletions.
2 changes: 1 addition & 1 deletion core/internal/evaluator/caching.go
Original file line number Diff line number Diff line change
Expand Up @@ -281,7 +281,7 @@ 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 := len(partition.Offsets) // defaults to the length, so if all offsets are nil we make an empty slice
for i, offset := range partition.Offsets {
if offset != nil {
firstOffset = i
Expand Down
4 changes: 3 additions & 1 deletion core/internal/httpserver/prometheus.go
Original file line number Diff line number Diff line change
Expand Up @@ -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))
}
}
Expand Down
2 changes: 1 addition & 1 deletion core/internal/httpserver/prometheus_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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`)
Expand Down

0 comments on commit 9ae3118

Please sign in to comment.