Home  >  Article  >  Backend Development  >  How to Retrieve Consumer Group Offsets in Golang Kafka 10 Using `sarama-cluster`?

How to Retrieve Consumer Group Offsets in Golang Kafka 10 Using `sarama-cluster`?

Patricia Arquette
Patricia ArquetteOriginal
2024-10-26 02:22:27520browse

How to Retrieve Consumer Group Offsets in Golang Kafka 10 Using `sarama-cluster`?

How to Retrieve Consumer Group Offsets in Golang Kafka 10

Previously, external libraries such as kazoo-go were utilized to retrieve consumer group message offsets stored in Zookeeper. However, with the introduction of consumer group capability in the Golang Kafka library (sarama) in version 10, it's possible to access these offsets directly.

Using sarama-cluster

To obtain consumer group offsets using the sarama-cluster library:

<code class="go">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>

This code creates a new ConsumerGroup instance and retrieves the current offset by consuming a message from the specified topic using an empty handler (consumeClaim). The initial offset of the claim is then recorded in the gcInfo struct.

The above is the detailed content of How to Retrieve Consumer Group Offsets in Golang Kafka 10 Using `sarama-cluster`?. For more information, please follow other related articles on the PHP Chinese website!

Statement:
The content of this article is voluntarily contributed by netizens, and the copyright belongs to the original author. This site does not assume corresponding legal responsibility. If you find any content suspected of plagiarism or infringement, please contact admin@php.cn