Getting Consumer Group Offsets in Golang Kafka 10
With the introduction of consumer group support in Golang's Kafka library (Sarama) in Kafka 10, developers now have access to consumer group-related features without relying on external libraries. One common task is retrieving the current message offset being processed by a consumer group. Previously, this required the use of Zookeeper-based solutions, such as Kazoo-go. Now, with Sarama-cluster, this can be achieved through the following code:
<code class="go">package main import ( "context" "log" "strings" "github.com/Shopify/sarama" ) func main() { groupName := "testgrp" topic := "topic_name" offset, err := GetCGOffset(context.Background(), "localhost:9092", groupName, topic) if err != nil { log.Fatal(err) } log.Printf("Consumer group %s offset for topic %s is: %d", groupName, topic, offset) } type gcInfo struct { offset int64 } func (g *gcInfo) Setup(sarama.ConsumerGroupSession) error { return nil } func (g *gcInfo) Cleanup(sarama.ConsumerGroupSession) error { return nil } func (g *gcInfo) ConsumeClaim(_ sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error { g.offset = claim.InitialOffset() return nil } func GetCGOffset(ctx context.Context, brokers, groupName, topic string) (int64, error) { config := sarama.NewConfig() config.Consumer.Offsets.AutoCommit.Enable = false // we're not going to update the consumer group offsets client, err := sarama.NewConsumerGroup(strings.Split(brokers, ","), groupName, config) if err != nil { return 0, err } info := gcInfo{} if err := client.Consume(ctx, []string{topic}, &info); err != nil { return 0, err } return info.offset, nil }</code>
The above is the detailed content of **How to Get Consumer Group Offsets in Golang Kafka 10?**. For more information, please follow other related articles on the PHP Chinese website!