获取分配给 Kafka 分区的使用者或客户端 ID

Question

我们有一个代码来获取kafka主题的消费者的一些细节。下面的代码显示了如何获得分区和相应的偏移量。我们需要的缺失信息是客户组中分区的客户id/客户。我们有办法得到每个主题分区的消费者吗？

    ArrayList<TopicPartition> partitions = new ArrayList<TopicPartition>();
    ArrayList<OffsetAndMetadata> offsets = new ArrayList<OffsetAndMetadata>();

    for (int i=0;i<consumer.partitionsFor(topic).size();i++)
    {
        TopicPartition partitiontemp = new TopicPartition(topic, i);
        partitions.add(partitiontemp);
        OffsetAndMetadata offsettemp = consumer.committed(partitiontemp);
        offsets.add(offsettemp);
    }

    consumer.assign(partitions);
    consumer.seekToEnd(partitions);

    for (int i=0;i<consumer.partitionsFor(topic).size();i++)
    {
        try {

            long cur_offset = offsets.get(partitions.get(i).partition()).offset();
            long log_offset = consumer.position(partitions.get(partitions.get(i).partition()));

            System.out.printf("Topic: %s partitionID: %d current offset: %d log offset: %d uncommitted: %d\n",
                    topic, partitions.get(i).partition(),cur_offset , log_offset , log_offset - cur_offset);


        }catch (Exception ex){
            System.out.printf("Topic: %s partitionID: %d current offset: - log offset: - uncommitted: -\n", topic, partitions.get(i).partition());

        }
    }

Answer 1

您可能是指消费者 group.id 属性，因为偏移量是每个组，也不是每个 client.id。此讨论 stackoverflow.com/questions/55937806/apache-kafka-get-list-of-consumers-on-a-specific-topic/55938325

可以回答你的问题。