How to get kafka consume lag in java program

问题:

I wrote a java program to consume messsage from kafka. I want to monitor the consume lag, how to get it by java?

BTW, I use:

<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.11</artifactId>
<version>0.10.1.1</version>

Thanks in advance.

回答1:

I personnaly query directly jmx informations from my consumers. I only consume in java so the JMX beans : kafka.consumer:type=consumer-fetch-manager-metrics,client-id=*/records-lag-max are available.

If jolokia is in your classpath you can retrieve the value with a GET on /jolokia/read/kafka.consumer:type=consumer-fetch-manager-metrics,client-id=*/records-lag-max and gather all the results in one place.

There is also Burrow which is very easy to configure, but it's a bit outdated (doesn't work for 0.10 if I remember well).

回答2:

In case if you don't want to include kafka (and scala) dependencies to your project you can use class below. It uses only kafka-clients dependencies.

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.BinaryOperator;
import java.util.stream.Collectors;

public class KafkaConsumerMonitor {

    public static class PartionOffsets {
        private long endOffset;
        private long currentOffset;
        private int partion;
        private String topic;

        public PartionOffsets(long endOffset, long currentOffset, int partion, String topic) {
            this.endOffset = endOffset;
            this.currentOffset = currentOffset;
            this.partion = partion;
            this.topic = topic;
        }

        public long getEndOffset() {
            return endOffset;
        }

        public long getCurrentOffset() {
            return currentOffset;
        }

        public int getPartion() {
            return partion;
        }

        public String getTopic() {
            return topic;
        }
    }

    private final String monitoringConsumerGroupID = "monitoring_consumer_" + UUID.randomUUID().toString();

    public Map<TopicPartition, PartionOffsets> getConsumerGroupOffsets(String host, String topic, String groupId) {
        Map<TopicPartition, Long> logEndOffset = getLogEndOffset(topic, host);


        KafkaConsumer consumer = createNewConsumer(groupId, host);

        BinaryOperator<PartionOffsets> mergeFunction = (a, b) -> {
            throw new IllegalStateException();
        };

        Map<TopicPartition, PartionOffsets> result = logEndOffset.entrySet()
                .stream()
                .collect(Collectors.toMap(
                        entry -> (entry.getKey()),
                        entry -> {
                            OffsetAndMetadata committed = consumer.committed(entry.getKey());
                            return new PartionOffsets(entry.getValue(), committed.offset(), entry.getKey().partition(), topic);
                        }, mergeFunction));


        return result;
    }

    public Map<TopicPartition, Long> getLogEndOffset(String topic, String host) {
        Map<TopicPartition, Long> endOffsets = new ConcurrentHashMap<>();
        KafkaConsumer<?, ?> consumer = createNewConsumer(monitoringConsumerGroupID, host);
        List<PartitionInfo> partitionInfoList = consumer.partitionsFor(topic);
        List<TopicPartition> topicPartitions = partitionInfoList.stream().map(pi -> new TopicPartition(topic, pi.partition())).collect(Collectors.toList());
        consumer.assign(topicPartitions);
        consumer.seekToEnd(topicPartitions);
        topicPartitions.forEach(topicPartition -> endOffsets.put(topicPartition, consumer.position(topicPartition)));
        consumer.close();
        return endOffsets;
    }

    private static KafkaConsumer<?, ?> createNewConsumer(String groupId, String host) {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, host);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return new KafkaConsumer<>(properties);
    }
}

回答3:

I am using Spring for my api. Using the below code, you can get the metrics via java.The code works.

@Component
public class Receiver {

private static final Logger LOGGER =
      LoggerFactory.getLogger(Receiver.class);


@Autowired
private KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

  public void testlag() {
      for (MessageListenerContainer messageListenerContainer : kafkaListenerEndpointRegistry
                .getListenerContainers()) {
          Map<String, Map<MetricName, ? extends Metric>> metrics = messageListenerContainer.metrics();
          metrics.forEach( (clientid, metricMap) ->{
              System.out.println("------------------------For client id : "+clientid);
              metricMap.forEach((metricName,metricValue)->{
                  //if(metricName.name().contains("lag"))
                  System.out.println("------------Metric name: "+metricName.name()+"-----------Metric value: "+metricValue.metricValue());
              });
          });
            }
  }

回答4:

Try to use AdminClient#listGroupOffsets(groupID) to retrieve offsets of all topic partitions associated with the consumer's group. For example:

AdminClient client = AdminClient.createSimplePlaintext("localhost:9092");
Map<TopicPartition, Object> offsets = JavaConversions.asJavaMap(
    client.listGroupOffsets("groupID"));
Long offset = (Long) offsets.get(new TopicPartition("topic", 0));
...

EDIT:
Snippets above show how to get the committed offset for a given partition. Code below shows how to retrieve LEO for a partition.

public long getLogEndOffset(TopicPartition tp) {
    KafkaConsumer consumer = createNewConsumer();
    Collections.singletonList(tp);
    consumer.assign(Collections.singletonList(tp));
    consumer.seekToEnd(Collections.singletonList(tp));
    return consumer.position(tp);
}

private KafkaConsumer<String, String> createNewConsumer() {
    Properties properties = new Properties();
    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    properties.put(ConsumerConfig.GROUP_ID_CONFIG, "g1");
    properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
    properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    return new KafkaConsumer(properties);
}

Invoking getLogEndOffset returns the LEO for the given partition, then subtract the committed offset from it and the result is the lag.