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

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

Patricia Arquette

Release： 2024-10-26 02:22:27

Original

639 people have browsed it

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>

Copy after login

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!