[SOLVED] Monitor the lag for the consumers that are assigned to partitions topic

Monitor the lag for the consumers that are assigned to partitions topic

I'm working with Kafka 0.9.1 new consumer API. The consumer is manually assigned to a partition. For this consumer I would like to see its progress (meaning the lag). Since I added the group id consumer-tutorial as property, I assumed that I can use the command

bin/kafka-consumer-groups.sh --new-consumer --describe --group consumer-tutorial --bootstrap-server localhost:9092

(as explained here http://www.confluent.io/blog/tutorial-getting-started-with-the-new-apache-kafka-0.9-consumer-client)

Unfortunately, my consumer group details is not shown using the above command. Therefore I cannot monitor the progress of my consumer (it's lag). How can I monitor the lag in the above described scenario (manually assigned partition)?

The code is:

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "consumer-tutorial");
props.put("key.deserializer", StringDeserializer.class.getName());
props.put("value.deserializer", StringDeserializer.class.getName());
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

String topic = "my-topic";
TopicPartition topicPartition = new TopicPartition(topic, 0);
consumer.assign(Arrays.asList(topicPartition));
consumer.seekToBeginning(topicPartition);
try {
  while (true) {
    ConsumerRecords<String, String> records = consumer.poll(1000);
    for (ConsumerRecord<String, String> record : records)
      System.out.println(record.offset() + ": " + record.value());
    consumer.commitSynch();
  }
} finally {
  consumer.close();
}

Solution

Just in case you don't want to write code to get this info or run command-like tools/shell scripts ad-hoc, there are N tools that will capture Kafka metrics, including Consumer Lag. Off the top of my head: Burrow and SPM for Kafka do a good job. Here is a bit of background about Kafka offsets, consumer lag, and a few metrics derived from what Kafka exposes via JMX. HTH.