Kafka consumer : offset lag between consumer and poducer

Question

I am writing a Kafka consumer application where I have one consumer for each partition. The code is shown below

while (true) {  
  ConsumerRecords<String, String> records = consumer.poll(100);   
        for (ConsumerRecord<String, String> record : records) {     
            System.out.printf("topic = %s, partition = %s, offset = %d, customer = %s, country = %s\n",        
            record.topic(), record.partition(), record.offset(), record.key(), record.value());   
 }    
  consumer.commitAsync(new OffsetCommitCallback() {       
        public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception e) {  
           if (e != null)              
              log.error("Commit failed for offsets {}", offsets, e); }}); 
}

Is there a way to programmatically access and print the consumer lag offset, or otherwise said the difference in positions between the offset of the last record read by the consumer and the offset of the last record written into that consumer partition by a certain producer.

What statements should I add to the above to get the lag offset value, knowing that my final goal is to send this value to prometheus for monitoring?

Mickael Maison Mickael Maison · Accepted Answer · 2020-10-26T12:47:54

If your goal is to have the data in prometheus, you should use the records-lag metrics emitted by Consumers.

In the last table in the Consumer Fetch Metrics section in the docs, you can see that Consumers emit lag (and lead) per topic-partition.

By default, metrics are emitted via JMX, so you could use Prometheus' JMX exporter to do the job for example.

Kafka consumer : offset lag between consumer and poducer

2 Answers

Using KafkaConsumer's endOffsets API

Using JMX Metrics