在Kafka 0.9中，有没有一种方法可以列出一个消费组中所有消费者的偏移量？

Question

我正在使用Kafka 0.9新的消费者API。

我让Kafka为消费者负责补偿。我让消费者在多台机器上阅读相同的主题。

我试图找出以下内容：

• 在消费者组中注册的消费者
• 每个消费者的偏移量

我以为消费者-群体-消费者关系会存储在ZooKeeper中。我在ZooKeeper中看到了消费者节点，它没有孩子。

通过查看代码，我可以知道的偏移量被写入了kafka，但我不知道它们被写入了哪个主题？

Answer 1

如果kafka处理的偏移量没有存储在zookeeper中，它存储在kafka-logs文件夹中的主题调用“__consumer_offsets-#”中。

如果您想了解有关使用者组check bin/Kafka-consumer-groups . sh的更多信息，可以在执行poll()时通过检查KafkaRecords中的offset字段来找出每个使用者的偏移量

希望它有帮助！

Answer 2

正如@hba所提到的，编码/解码的详细信息可以在kafka.coordinator中找到。底部附近的GroupMetadataManager。查找readMessageKey和以下两个方法。基本上，您需要的是一系列呼叫，如

import org.apache.kafka.common.protocol.types.Type;
...
ByteBuffer bb = ByteBuffer.wrap(consumerRecord.key())
short version = bb.getShort();
String group = (String)Type.String.read(bb);
String topic = (String)Type.String.read(bb);
int partition = (int)Type.INT32.read(bb);

好处是org.apache.kafka.common.protocol.types.Type是Javaapi的一部分，独立于Big main Jar。难看的是，上面的代码片段并不完整。每个consumerRecord.key（）和consumerRecord.value（）都有两个版本，其中一个必须模仿上述方法的解码。没什么大不了的，只是有点乏味。

如果您的项目可以依赖于卡拉罐，完整的Kafka罐和Kafka所需的一个或两个额外的罐子，那么您也可以使用组元数据管理器.readMessageKey（bb）和其他两种方法来读取键和值。至少在0.9.0.1中是它的公开版本。

Answer 3

@nautilus指出的__consumer_offsets topic中似乎至少存储了2种类型的键值对。

1. 组元数据信息
2. 偏移提交

据我所知，Kafka使用自己的模式和序列化。您可以通过查看< code > Kafka . coordinator . groupmetadatamanager 找到有关这些结构的更多信息:

• GroupMetadataManager.OFFSET_COMMIT_KEY_SCHEMA
• GroupMetadataManager.OFFSET_COMMIT_VALUE_SCHEMA_V0
• GroupMetadataManager.GROUP_METADATA_KEY_SCHEMA
• GroupMetadataManager.GROUP_METADATA_VALUE_SCHEMA_V0