Kafka消费者：消费者与消费者的抵消滞后

Question

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); }}); 
}

是否有一种方法以编程方式访问和打印使用者滞后偏移，或者说使用者读取的最后一条记录的偏移与某个生产者写入该使用者分区的最后一条记录的偏移之间的位置差。

要知道我的最终目标是将这个值发送到prometheus进行监视，我应该在上面添加哪些语句来得到滞后偏移值？

Answer 1

如果您的目标是在prometheus中拥有数据，则应该使用使用者发出的Records-Lag度量。

在文档中的Consumer Fetch Metrics部分的最后一个表中，您可以看到Consumer在每个主题分区中发出滞后（和领先）。

默认情况下，度量是通过JMX发出的，因此您可以使用Prometheus的JMX导出器来完成这项工作。