Как получить смещение, хранящееся в zookeeper

Question

Требуется для генерации метрик задержки потребителя для нашего приложения (Kafka 0.10), в котором смещения хранятся в zookeeper.Ищете Java API или другие способы получения смещений, хранящихся в zookeeper.

Answer 1

Спасибо Нишу за ваше предложение.Поскольку мы ищем смещения, хранящиеся в zookeeper, нам нужно использовать bin / kafka-consumer-groups.sh --zookeeper --describe --group

Соответствующий API Scala доступен в базовой реализации

ConsumerGroupCommand.scala

Answer 2

Если вы пытаетесь из командной строки, вы можете использовать скрипт kafka-consumer-groups.sh следующим образом.

bin/kafka-consumer-groups.sh --bootstrap-server <broker> --describe --group <group-name>

Он вернет данные следующим образом:

    TOPIC       PARTITION  CURRENT-OFFSET  LOG-END-OFFSET      LAG        CONSUMER-ID                                       HOST                           CLIENT-ID
testtopic-1         0          13164704        13164773        69         test-consumer-bf8d1655-63a6-4061-b680-0f11cdf182e5/100.96.67.0                     test-consumer              
testtopic-1         1          13161581        13161650        69         test-consumer-agent-bf8d1655-63a6-4061-b680-0f11cdf182e5/100.96.67.0               test-consumer   
testtopic-1         2          12229509        12229578        69         test-consumer-bf8d1655-63a6-4061-b680-0f11cdf182e5/100.96.67.0                     test-consumer
testtopic-1         3          86              86              0          test-consumer-agent-bf8d1655-63a6-4061-b680-0f11cdf182e5/100.96.67.0               test-consumer

Если вы используете Java, вы можете использовать AdminClient.describeConsumerGroups метод ниже:

http://kafka.apache.org/20/javadoc/org/apache/kafka/clients/admin/AdminClient.html#describeConsumerGroups-java.util.Collection-