go - 如何在 Golang Kafka 10 中获取 GroupID？

Question

我正在使用 Kafka 10.0 和 https://github.com/Shopify/sarama .我正在尝试获取消费者处理的最新消息的偏移量。

为此，我找到了方法 NewOffsetManagerFromClient(group string, client Client)需要组名。

如何获取消费者组名称？

offsets := make(map[int32]int64)

config := sarama.NewConfig()
config.Consumer.Offsets.CommitInterval = 200 * time.Millisecond
config.Version = sarama.V0_10_0_0

// config.Consumer.Offsets.Initial = sarama.OffsetNewest
cli, _ := sarama.NewClient(kafkaHost, config)
defer cli.Close()

offsetManager, _ := sarama.NewOffsetManagerFromClient(group, cli)
for _, partition := range partitions {
    partitionOffsetManager, _ := offsetManager.ManagePartition(topic, partition)
    offset, _ := partitionOffsetManager.NextOffset()

    offsets[partition] = offset
}
return offsets

我创建了一个消费者

consumer := sarama.NewConsumer(connections, config)

但我不知道如何创建消费者组并获取其组名。

Answer 1

您正在尝试创建自己的偏移量管理器以查找当前偏移量:

offsetManager, _ := sarama.NewOffsetManagerFromClient(group, cli)

类似地，消费您的主题消息的消费者必须使用相同的偏移量管理器，并且他们将使用特定的组 ID。使用该组 ID。