Java, Как получить количество сообщений в теме в apache kafka

Я использую apache kafka для обмена сообщениями. Я реализовал производителя и потребителя на Java. Как мы можем получить количество сообщений в теме?

Ответ 1

Единственный способ, который приходит на ум для этого с потребительской точки зрения, - это фактически потреблять сообщения и подсчитывать их тогда.

Брокер Kafka предоставляет счетчики JMX для количества сообщений, полученных с момента запуска, но вы не можете знать, сколько из них уже было очищено.

В большинстве распространенных сценариев сообщения в Кафке лучше всего воспринимаются как бесконечный поток и получают дискретное значение того, сколько из них в настоящее время хранится на диске, не имеет значения. Более того, все становится более сложным при работе с кластером брокеров, у которых есть подмножество сообщений в теме.

Ответ 2

Это не java, но может быть полезно

./bin/kafka-run-class.sh kafka.tools.GetOffsetShell 
  --broker-list <broker>:  <port> 
  --topic <topic-name> --time -1 --offsets 1 
  | awk -F  ":" '{sum += $3} END {print sum}'

Ответ 3

Я использую это для сравнения моего POC. Элемент, который вы хотите использовать ConsumerOffsetChecker. Вы можете запустить его с помощью bash script, как показано ниже.

bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker  --topic test --zookeeper localhost:2181 --group testgroup

И ниже приведен результат: Как вы можете видеть в красном поле, 999 - это номер сообщения, в настоящее время в теме.

Обновление: ConsumerOffsetChecker устарел с 0.10.0, вы можете начать использовать ConsumerGroupCommand.

Ответ 4

Используйте https://prestodb.io/docs/current/connector/kafka-tutorial.html

Супер-SQL-механизм, предоставляемый Facebook, который подключается к нескольким источникам данных (Cassandra, Kafka, JMX, Redis...).

PrestoDB работает как сервер с дополнительными рабочими (имеется автономный режим без дополнительных работников), тогда вы используете небольшой исполняемый JAR (называемый presto CLI) для создания запросов.

Как только вы правильно настроили сервер Presto, вы можете использовать традиционный SQL:

SELECT count(*) FROM TOPIC_NAME;

Ответ 5

Иногда интерес заключается в том, чтобы узнать количество сообщений в каждом разделе, например, при тестировании пользовательского разделителя. Последующие шаги были протестированы для работы с Kafka 0.10.2.1-2 из Confluent 3.2. Учитывая тему Кафки, kt и следующую командную строку:

$ kafka-run-class kafka.tools.GetOffsetShell \
  --broker-list host01:9092,host02:9092,host02:9092 --topic kt

Это выводит пример вывода, показывающий количество сообщений в трех разделах:

kt:2:6138
kt:1:6123
kt:0:6137

Количество строк может быть более или менее в зависимости от количества разделов для темы.

Ответ 6

Поскольку ConsumerOffsetChecker больше не поддерживается, вы можете использовать эту команду для проверки всех сообщений в теме:

bin/kafka-run-class.sh kafka.admin.ConsumerGroupCommand \
    --group my-group \
    --bootstrap-server localhost:9092 \
    --describe

Где LAG - количество сообщений в тематическом разделе:

Также вы можете попробовать использовать kafkacat. Это проект с открытым исходным кодом, который может помочь вам прочитать сообщения из темы и раздела и распечатать их на стандартный вывод. Вот пример, который читает последние 10 сообщений из темы sample-kafka-topic, затем завершается:

kafkacat -b localhost:9092 -t sample-kafka-topic -p 0 -o -10 -e

Ответ 7

Команда Apache Kafka для получения не обработанных сообщений по всем разделам темы:

kafka-run-class kafka.tools.ConsumerOffsetChecker 
    --topic test --zookeeper localhost:2181 
    --group test_group

Печать

Group      Topic        Pid Offset          logSize         Lag             Owner
test_group test         0   11051           11053           2               none
test_group test         1   10810           10812           2               none
test_group test         2   11027           11028           1               none

Столбец 6 - это не обработанные сообщения. Добавьте их так:

kafka-run-class kafka.tools.ConsumerOffsetChecker 
    --topic test --zookeeper localhost:2181 
    --group test_group 2>/dev/null | awk 'NR>1 {sum += $6} 
    END {print sum}'

awk читает строки, пропускает строку заголовка и добавляет шестой столбец, а в конце печатает сумму.

Печать

Ответ 8

Чтобы получить все сообщения, сохраненные для этой темы, вы можете искать пользователя в начале и конце потока для каждого раздела и суммировать результаты

List<TopicPartition> partitions = consumer.partitionsFor(topic).stream()
        .map(p -> new TopicPartition(topic, p.partition()))
        .collect(Collectors.toList());
    consumer.assign(partitions); 
    consumer.seekToEnd(Collections.emptySet());
Map<TopicPartition, Long> endPartitions = partitions.stream()
        .collect(Collectors.toMap(Function.identity(), consumer::position));
    consumer.seekToBeginning(Collections.emptySet());
System.out.println(partitions.stream().mapToLong(p -> endPartitions.get(p) - consumer.position(p)).sum());

Ответ 9

В последних версиях Kafka Manager есть столбец "Суммированные недавние смещения".

Ответ 10

Запустите следующее (при условии, что kafka-console-consumer.sh находится на пути):

kafka-console-consumer.sh  --from-beginning \
--bootstrap-server yourbroker:9092 --property print.key=true  \
--property print.value=false --property print.partition \
--topic yourtopic --timeout-ms 5000 | tail -n 10|grep "Processed a total of"

Ответ 11

Используя Java-клиент Kafka 2.11-1.0.0, вы можете сделать следующее:

    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
    consumer.subscribe(Collections.singletonList("test"));
    while(true) {
        ConsumerRecords<String, String> records = consumer.poll(100);
        for (ConsumerRecord<String, String> record : records) {
            System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());

            // after each message, query the number of messages of the topic
            Set<TopicPartition> partitions = consumer.assignment();
            Map<TopicPartition, Long> offsets = consumer.endOffsets(partitions);
            for(TopicPartition partition : offsets.keySet()) {
                System.out.printf("partition %s is at %d\n", partition.topic(), offsets.get(partition));
            }
        }
    }

Результат выглядит примерно так:

offset = 10, key = null, value = un
partition test is at 13
offset = 11, key = null, value = deux
partition test is at 13
offset = 12, key = null, value = trois
partition test is at 13

Ответ 12

Я не пробовал this сам, но, похоже, имеет смысл.

Вы также можете использовать kafka.tools.ConsumerOffsetChecker (источник).

Ответ 13

Выдержки из документов Кафки

Амортизация в 0.9.0.0

Файл kafka-consumer-offset-checker.sh(kafka.tools.ConsumerOffsetChecker) устарел. В дальнейшем, пожалуйста, используйте kafka-consumer-groups.sh(kafka.admin.ConsumerGroupCommand) для этой функциональности.

Я использую брокер Kafka с включенным SSL как для сервера, так и для клиента. Ниже я использую команду

kafka-consumer-groups.sh --bootstrap-server Broker_IP:Port --list --command-config/tmp/ssl_config kafka-consumer-groups.sh --bootstrap-server Broker_IP:Port --command-config/tmp/ssl_config --describe --group group_name_x

где /tmp/ssl_config, как показано ниже

security.protocol=SSL
ssl.truststore.location=truststore_file_path.jks
ssl.truststore.password=truststore_password
ssl.keystore.location=keystore_file_path.jks
ssl.keystore.password=keystore_password
ssl.key.password=key_password

Ответ 14

Если у вас есть доступ к интерфейсу JMX сервера, начальные и конечные смещения присутствуют в:

kafka.log:type=Log,name=LogStartOffset,topic=TOPICNAME,partition=PARTITIONNUMBER
kafka.log:type=Log,name=LogEndOffset,topic=TOPICNAME,partition=PARTITIONNUMBER

(вам нужно заменить TOPICNAME & PARTITIONNUMBER). Имейте в виду, что вам нужно проверять каждую из реплик данного раздела, или вам нужно выяснить, какой из брокеров является лидером для данного раздела (и это может со временем измениться).

Кроме того, вы можете использовать Кафка Потребительские методы beginningOffsets и endOffsets.

Ответ 15

Самый простой способ, который я нашел, - это использовать Kafdrop REST API /topic/topicName и указать /topic/topicName ключа: "Accept"/value: "application/json", чтобы получить ответ JSON.

Это задокументировано здесь.

Ответ 16

Вы можете использовать kafkatool. Пожалуйста, проверьте эту ссылку → http://www.kafkatool.com/download.html

Kafka Tool - приложение с графическим интерфейсом для управления и использования Apache Kafka кластеры. Он предоставляет интуитивно понятный интерфейс, который позволяет быстро просматривать объекты в кластере Кафки, а также сообщения, хранящиеся в Темы кластера.