как проверить, подключен ли определенный потребитель к Kafka 0.9.0.x с помощью java?

Как получить список подключенных потребителей в kafka? Поскольку потребители подключены к брокеру, есть ли какая-либо Java-утилита, такая как ZkClient/ZkUtils, для получения списка подключенных потребителей в Kafka 0.9.0.x? Как мы используем, чтобы получить список брокеров, используя следующую утилиту:

        ZkClient zkClient = new ZkClient(endpoint.getZookeeperConnect(), 60000);

        if(zkClient!=null){
            List<String> brokerIds = zkClient.getChildren(ZkUtils.BrokerIdsPath());
            if(CollectionUtils.isNotEmpty(brokerIds) &&  brokerIds.contains(brokerId)){
                logger.debug("Broker:{{}} is connected to Zookeeper.",brokerId);
                flag = true;    
            }
            else{
                logger.error("ERROR:Broker:{{}} is not connected to Zookeeper.",brokerId);
            }
            zkClient.close();
        }

Я использую Kafka 0.9.0.x с java-библиотекой ниже от maven:

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

ОБНОВЛЕНО:

я открыл «kafka-console-consumer.bat» и запустил его один раз, а затем пересек приглашение cmd. Затем перешел к "zookeeper-shell.bat" и ls /consumers показал [console-consumer-6008], но мой запрограммированный потребители не показаны. используя zkClient.getChildren(ZkUtils.ConsumersPath()), теперь я могу просматривать только упомянутого потребителя.


java apache-kafka apache-zookeeper
person usman    schedule 17.10.2016    source источник

Ответы (3)


arrow_upward
5
arrow_downward

Не уверен, какая именно информация вам нужна, но я сделал пример программы, которая дает ту же информацию, что и kafka-consumer-groups.sh --describe.

Чтобы использовать этот код, добавьте эти зависимости в ваш файл pom.

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

Затем:

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import kafka.admin.AdminClient;
import kafka.coordinator.GroupOverview;

Properties props = new Properties();
props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092");
AdminClient adminClient = AdminClient.create(props);

List<GroupOverview> groups =  scala.collection.JavaConversions.seqAsJavaList(
        adminClient.listAllConsumerGroupsFlattened());
for (GroupOverview group : groups) {
    String groupId = group.groupId();

    Properties consProps = new Properties();
    consProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092");
    consProps.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
    consProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
    consProps.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
    consProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    consProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    KafkaConsumer consumer = new KafkaConsumer(consProps);

    List<AdminClient.ConsumerSummary> groupSummaries = scala.collection.JavaConversions.seqAsJavaList(
            adminClient.describeConsumerGroup(groupId));

    System.out.println("GROUP, TOPIC, PARTITION, CURRENT OFFSET, LOG END OFFSET, LAG, OWNER");

    for (AdminClient.ConsumerSummary summary : groupSummaries) {
        String owner = summary.clientId() + "_" + summary.clientHost();
        List<TopicPartition> topicPartitions = scala.collection.JavaConversions.seqAsJavaList(
                summary.assignment());
        for (TopicPartition tp : topicPartitions) {

            // Get current offset
            long currentOffset = consumer.committed(tp).offset();

            // get log end offset
            consumer.assign(Arrays.asList(tp));
            consumer.seekToEnd();
            long logEndOffset = consumer.position(tp);

            long lag = logEndOffset - currentOffset;

            System.out.println(groupId + ", " + tp.topic() + ", " + tp.partition() + ", " +
                    currentOffset + ", " + logEndOffset + ", " + lag + ", " + owner);
        }
    }
}
person Luciano Afranllie    schedule 17.10.2016
comment
Спасибо, именно то, что мне нужно, чтобы получить список только работающих потребителей. это было достигнуто с помощью метода AdminClient + listAllConsumerGroupsFlattened(). В kafak все еще довольно скрыто. - person usman; 18.10.2016

arrow_upward
1
arrow_downward

Для нового потребителя 0.9.x и списка всех активных групп потребителей:

  1. найти всех брокеров и отправить запрос «ListGroups» каждому из брокеров и получить всю информацию о группе;

Для получения дополнительной информации см. $KAFKA_HOME/bin/kafka-consumer-groups.sh(kafka.admin.ConsumerGroupCommand.KafkaConsumerGroupService.list())

Для 0.9.x новый потребитель и опишите подробную информацию об определенной группе потребителей:

  1. найти координатора группы потребителей и отправить ему запрос «DescribeGroups» и получить информацию обо всех членах группы и информацию о назначении разделов;
  2. вызовите KafkaConsumer.committed(раздел TopicPartition), чтобы получить последнее зафиксированное смещение для данного раздела.

Для получения дополнительной информации см. $KAFKA_HOME/bin/kafka-consumer-groups.sh(kafka.admin.ConsumerGroupCommand.KafkaConsumerGroupService.describe())

Обратите внимание, что старый потребитель и новый потребитель имеют совершенно разные реализации по этому поводу. (Обе логики реализованы в kafka.admin.ConsumerGroupCommand.

person Shawn Guo    schedule 17.10.2016
comment
это редактирование было ошибкой, которую пришлось добавить в свой вопрос, извините за то, что не знаю, как его отбросить. - person usman; 18.10.2016
comment
Кстати, я использую код Java в Windows, в Kafka Windows нет «kafka-consumer-groups.bat», что делать сейчас. - person usman; 18.10.2016

arrow_upward
0
arrow_downward

Это почти то же самое, но вы должны проверить ZkUtils.ConsumersPath (= /consumers).

Структура потребителей в Zookeeper следующая: /consumers/[groupId]/ids/[consumerId], поэтому при навигации вы можете получить группы и потребителей для каждой группы.

person gasparms    schedule 17.10.2016
comment
ZkUtils.ConsumersPath (/consumers) всегда возвращал [ ]. Я думаю, что информация о группах потребителей теперь сохраняется на kafka. Я уже прошел эту часть, чтобы проверить список потребителей. - person usman; 17.10.2016
comment
В 0.9.x и 0.10.x по-прежнему сохраняются группы потребителей и потребители. Вы можете проверить это в коде. ZkUtils.getConsumers получает дочерние элементы ConsumersPath. github.com/ apache/kafka/blob/trunk/core/src/main/scala/kafka/ - person gasparms; 17.10.2016
comment
zkClient.getChildren(ZkUtils.ConsumersPath()) возвращает пустой [ ]. - person usman; 17.10.2016
comment
Попробуйте с оболочкой, откройте производителя консоли и потребителя консоли. Затем создайте сообщение. Наконец, проверьте с помощью средства zookeeper shell ls /consumers = [console-consumer-44669]. Если оно работает, все в порядке, и ошибка будет на ваших незарегистрированных потребителях. Я только что попробовал и увидел потребителя в узле zk/consumers. - person gasparms; 17.10.2016
comment
я открыл «kafka-console-consumer.bat» и запустил его один раз, а затем пересек приглашение cmd. Затем перешел к zookeeper-shell.bat, и ls /consumers показал [console-consumer-6008], но мои запрограммированные потребители не отображаются. используя 'zkClient.getChildren(ZkUtils.ConsumersPath())', теперь я могу просматривать только упомянутого потребителя. - person usman; 18.10.2016