检索过去的kafka消费者的历史

Question

有没有办法获取作为给定消费者组成员连接到代理的消费者（IP）列表？

我知道 kafka-consumer-group.sh 命令会为所有当前连接/活动的消费者执行此操作。但我感兴趣是否有一个临时消费者，例如 5 分钟前，它确实消费了主题中的一些消息。

我还尝试增加授权者的日志记录以进行调试。但结果是在几个小时内增加了数百万条日志。

谢谢，乔臣

Answer 1

尝试使用 AdminClient 和 listConsumerGroups 来获取消费者列表

/*List the consumer groups available in the cluster with the default options.*/
public ListConsumerGroupsResult listConsumerGroups() {
        return listConsumerGroups(new ListConsumerGroupsOptions());
    }

下面的示例获取消费者列表

    import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsResult;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    iomport org.apache.kafka.clients.admin.AdminClient;

    AdminClient client = AdminClient.create(props);
    //.all(): Returns a future that yields either an exception, or the full set of consumer group listings.
    List<String> listConsumer=client.listConsumerGroups().all().get().stream().
    map(f -> f.groupId()).collect(Collectors.toList());