我正在尝试使用.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();
}
查找每个分区/使用者组的使用者滞后和最新偏移量。
1条答案
按热度按时间nuypyhwy1#
好吧,我终于知道怎么做了。似乎需要将主题分区分配设置为使用者组。就我而言,主题是
test3
我有三个分区。这必须在代码中设置。我想一个更通用的实现是将与这个组相关联的列表传递给这个方法。下面是我如何得到消费者滞后: