如何使用kafka.net客户端api列出每个分区和使用者组的kafka使用者延迟和最新偏移量？

d6kp6zgx 于 2021-06-05 发布在 Kafka

关注(0)|答案(1)|浏览(420)

我正在尝试使用.NETConfluent.kafka1.4.0-rc1（对于net472）获取消费者延迟。我可以使用以下脚本获得所需的结果：

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

结果：

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                  HOST            CLIENT-ID
Grp1            test3           1          15              15              0               rdkafka-ca76855f-7b66-4bf1-82bc-73e9a1c1cf71 /10.186.129.93  rdkafka
Grp1            test3           2          13              13              0               rdkafka-d64379dc-881a-4f6f-a793-51e832cc2f5a /10.186.129.93  rdkafka
Grp1            test3           0          9               9               0               rdkafka-a25bdb80-3b70-4e42-963e-d41ad9e2a99a /10.186.129.93  rdkafka
Grp1            test            0          68              68              0               -                                            -               -

我无法使用.net客户端代码获得类似的报告。这是我一直在尝试的代码-但是没有得到任何结果 consumer.Assignment 属性的集合为空。

private string WriteConsumerGroupLags(string bootstrapServers, string consumerGroupName) {
    // kafka-console-consumer.bat --zookeeper MW45670117:2380 --topic powertelemetry --consumer-property group.id=test123 --consumer-property enable.auto.commit=true

    StringBuilder sb = new StringBuilder();
    sb.AppendLine("\n");
    sb.AppendLine("Consumer Group Lag Report");
    sb.AppendLine("-------------------------");
    ConsumerConfig config = new ConsumerConfig {
        BootstrapServers = bootstrapServers,
        GroupId = consumerGroupName,
        AutoOffsetReset = AutoOffsetReset.Earliest
    };

    using (var consumer = new ConsumerBuilder<Ignore, string>(config).Build()) {
        foreach (TopicPartition tp in consumer.Assignment) {
            string topic = tp.Topic;
            int partitionID = tp.Partition.Value;
            // gets the current position (offset) for the specific topic/partition
            Offset offset = consumer.Position(new TopicPartition(topic, new Partition(partitionID)));
            sb.AppendLine($"Offset value is: {offset.Value}");

            // returns current commited offset for the current assignment
            List<TopicPartitionOffset> tpos = consumer.Committed(TimeSpan.FromSeconds(4));
            foreach (TopicPartitionOffset tpo in tpos) {
                sb.AppendLine($"Commited offset for partition {tpo.TopicPartition} is {tpo.Offset}");
            }
        }
    }

    return sb.ToString();
}

查找每个分区/使用者组的使用者滞后和最新偏移量。

apache-kafka .net confluent-platform

来源：https://stackoverflow.com/questions/60506580/how-can-i-list-the-kafka-consumer-lag-and-latest-offset-per-partition-and-consum

1条答案

按热度按时间

nuypyhwy1#

好吧，我终于知道怎么做了。似乎需要将主题分区分配设置为使用者组。就我而言，主题是 test3 我有三个分区。这必须在代码中设置。我想一个更通用的实现是将与这个组相关联的列表传递给这个方法。下面是我如何得到消费者滞后：

private string WriteConsumerGroupLags(string bootstrapServers, string consumerGroupName) {
    StringBuilder sb = new StringBuilder();
    sb.AppendLine("\n");
    sb.AppendLine("Consumer Group Lag Report");
    sb.AppendLine("-------------------------");
    ConsumerConfig config = new ConsumerConfig {
        BootstrapServers = bootstrapServers,
        GroupId = consumerGroupName,
        AutoOffsetReset = AutoOffsetReset.Earliest
    };

    using (var consumer = new ConsumerBuilder<Ignore, string>(config).Build()) {
        List<TopicPartition> topic_partitions = new List<TopicPartition>() {
            { new TopicPartition("test3", new Partition(0)) },
            { new TopicPartition("test3", new Partition(1) ) },
            { new TopicPartition("test3", new Partition(2) ) } };

        consumer.Assign(topic_partitions);

        List<TopicPartitionOffset> tpos = consumer.Committed(TimeSpan.FromSeconds(40));
        foreach (TopicPartitionOffset tpo in tpos) {
            WatermarkOffsets w = consumer.QueryWatermarkOffsets(tpo.TopicPartition, TimeSpan.FromSeconds(40));
            long commited = tpo.Offset.Value;
            long log_end_offset = w.High.Value;
            long lag = log_end_offset - commited;
            sb.AppendLine($"Commited offset for Topic {tpo.TopicPartition.Topic} Partition {tpo.TopicPartition.Partition} is {commited} out of Wantermark end offset {log_end_offset} Lag is: {lag}");
        }

        sb.AppendLine();
    }

    return sb.ToString();
}

赞(0）回复(0）举报 2021-06-06

我来回答

如何使用kafka.net客户端api列出每个分区和使用者组的kafka使用者延迟和最新偏移量？

1条答案

相关问题

热门标签

最新问答