get Kafka unconsumed messages count in Java

user_2881308 0
Viewed 1928

We would like to know the messages count which are not yet consumed and acknowledge by Kafka Consumer for a given Kafka topic.

Is there any way we can get the count of Unconsumed Kafka messages from a given Topic (Topic is having 10 partitions) in Java ??

spring-kafka kafka-consumer-api apache-kafka java
2 Answers

kafka consumers have a per-partition "records-lag" metric (introduced in kip-92). you can sum those metrics for all partitions of your topic, and get a figure for unconsumed messages

user_2620220

There is another alternate approach

Kafka Consumer have API to get endpoint for each partition of a topic

List partitions = new ArrayList<>();
        for (PartitionInfo p : parts) {
            partitions.add(new TopicPartition(topic, p.partition()));
        }
Map<TopicPartition, Long> offsets = consumer.endOffsets(partitions);

For each topic partition, u can get the latest committed offset.You can easily get the unconsumed lagging by using these 2 numbers

lag=(end of offset-latest committed)

for (TopicPartition tp : offsets.keySet()) {
            OffsetAndMetadata commitOffset = consumer.committed(new TopicPartition(tp.topic(), tp.partition()));
            Long lag = commitOffset == null ? offsets.get(tp) : offsets.get(tp) - commitOffset.offset();            
        }
user_25318430
Related