feat(metrics): track consumer records lag

fetch_response.go

@@ -64,8 +64,10 @@ type FetchResponseBlock struct {
 func (b *FetchResponseBlock) decode(pd packetDecoder, version int16) (err error) {
 	metricRegistry := pd.metricRegistry()
 	var sizeMetric metrics.Histogram
+	var lagMetric metrics.Histogram
 	if metricRegistry != nil {
 		sizeMetric = getOrRegisterHistogram("consumer-fetch-response-size", metricRegistry)
+		lagMetric = getOrRegisterHistogram("consumer-records-lag", metricRegistry)
 	}
 
 	tmp, err := pd.getInt16()
@@ -180,6 +182,16 @@ func (b *FetchResponseBlock) decode(pd packetDecoder, version int16) (err error)
 		}
 	}
 
+	//Follows the Java implementation for calculating the records lag:
+	//https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java
+	if len(b.RecordsSet) > 0 {
+		logEndOffset := *b.LastRecordsBatchOffset - 1
+		partitionLag := logEndOffset - b.RecordsSet[len(b.RecordsSet)-1].RecordBatch.LastOffset()
+		if lagMetric != nil {
+			lagMetric.Update(partitionLag)
+		}
+	}
+
 	return nil
 }
 

sarama.go

@@ -72,6 +72,7 @@ Consumer related metrics:
 	| consumer-fetch-rate-for-broker-<broker>   | meter      | Fetch requests/second sent to a given broker                                         |
 	| consumer-fetch-rate-for-topic-<topic>     | meter      | Fetch requests/second sent for a given topic                                         |
 	| consumer-fetch-response-size              | histogram  | Distribution of the fetch response size in bytes                                     |
+	| consumer-records-lag                      | histogram  | The latest lag of the partition                                                      |
 	| consumer-group-join-total-<GroupID>       | counter    | Total count of consumer group join attempts                                          |
 	| consumer-group-join-failed-<GroupID>      | counter    | Total count of consumer group join failures                                          |
 	| consumer-group-sync-total-<GroupID>       | counter    | Total count of consumer group sync attempts                                          |