Сценарии - Получение данных о смещении

Page / - Изменено Alexandr Fokin в 2023/05/10 11:40

Получение данных о смещении

Изменено Alexandr Fokin

в 2023/05/10 11:40

Оглавление

Задание: Получение информации о текущем состоянии параметров: Общее кол-во сообщений в topic Общее кол-во прочитанных(commited) сообщений в topic (по определенной consumerGroup) Определение общего значения lag в topic(Разность между общей длиной и кол-во прочитанных сообщений) (по определенной consumerGroup) Опционально по каждому partition. Данные по topic можно определить по сумме совокупности данных по каждом partition, принадлежащему данному topic.
Get Consumer lag using LibrdkafkaHandle.OutQueueLength https://github.com/confluentinc/confluent-kafka-dotnet/issues/1161 Aug 2019 - Kafka Consumer Lag programmatically https://stackoverflow.com/questions/57302244/aug-2019-kafka-consumer-lag-programmatically
Решение: Данные по общему кол-ву сообщений и прочитанным сообщениям можно получить через Consumer. Причем для указанного partition. Данные по topic формируются через сумму данных всех partitions, входящих в него.
//номер смещения последнего сообщения в очереди consumer .QueryWatermarkOffsets(TopicPartitionInfo(topicName, partitionId),ConnectionParamsEntity.ActionTimeout) .High //Получить активное смещение (последнее прочитанное сообщение) consumer .Committed( new TopicPartition[] { TopicPartitionInfo(topicName, partitionId) }, ConnectionParamsEntity.ActionTimeout ) .FirstOrDefault() .Offset .Value;