"how to get latest offset/size of a kafka topic using kafkaadminclient (java) for 2.x version" Code Answer

Question

externally to the kafka consuming application you are correct, your options are to look at partition end offsets vs the latest checkpointed positions of the consumer group (assuming the consumers in question even use kafka to store offsets).

there are tools that will monitor this for you, such as burrow.

however, if you have access to the consuming application itself there is a more accurate way. here's a list of all consumer sensors (exposed either via api or jmx by default) https://kafka.apache.org/documentation/#consumer_fetch_monitoring.

there is a per-partition records-lag metric. its updated every time poll() is called so is more accurate and lower latency than committed offsets. the only complication is you'd need to sum the values of these sensors across all partitions the consumer is assigned.

here's how to get at it via kafkaconsumer.metrics():

private long calctotallag(map<metricname, ? extends metric> metrics) {
   long totallag = 0;
   for (map.entry<metricname, ? extends metric> entry : metrics.entryset()) {
     metricname metricname = entry.getkey();
     metric metric = entry.getvalue();
     map<string, string> tags = metricname.tags();
     if (metricname.name().equals("records-lag") && tags.containskey("partition")) {
        totallag += ((number) metric.metricvalue()).longvalue();
     }
   }

   return totallag;
}

"how to get latest offset/size of a kafka topic using kafkaadminclient (java) for 2.x version" Code Answer

Browse Other Code Languages