如何使用sharma或confluent-kafka-go库实现Kafka消费者组偏移量的获取与重置操作

Great question! I’ve worked with both confluent-kafka-go (the official Confluent Go client) and what I suspect is a typo for Sarama (Shopify’s widely used Kafka client for Go) to handle consumer group offset management. Let’s break down how to replicate the offset export/import workflow you’re running via kafka-consumer-groups.sh using both libraries.

Using confluent-kafka-go #

1. Export Offsets (Equivalent to the --export flag) #

To fetch and save offsets like your topic-offset.csv, you’ll use the AdminClient to query consumer group offsets and topic partition details. Here’s a practical example:

import (
    "fmt"
    "os"
    "strconv"

    "github.com/confluentinc/confluent-kafka-go/kafka"
)

func exportOffsets() {
    // Initialize admin client
    admin, err := kafka.NewAdminClient(&kafka.ConfigMap{
        "bootstrap.servers": "your-kafka-broker-list",
    })
    if err != nil {
        panic(fmt.Sprintf("Failed to create admin client: %v", err))
    }
    defer admin.Close()

    targetTopic := "your-target-topic"
    sourceGroup := "cg1"

    // Get all partitions for the target topic
    metadata, err := admin.GetMetadata(&targetTopic, false, 5000)
    if err != nil {
        panic(fmt.Sprintf("Failed to get topic metadata: %v", err))
    }
    var partitions []kafka.TopicPartition
    for _, part := range metadata.Topics[targetTopic].Partitions {
        partitions = append(partitions, kafka.TopicPartition{
            Topic:     &targetTopic,
            Partition: part.ID,
        })
    }

    // Fetch current offsets for the source consumer group
    offsetResults, err := admin.ListConsumerGroupOffsets(sourceGroup, partitions)
    if err != nil {
        panic(fmt.Sprintf("Failed to fetch group offsets: %v", err))
    }

    // Export to CSV file
    file, err := os.Create("topic-offset.csv")
    if err != nil {
        panic(fmt.Sprintf("Failed to create CSV file: %v", err))
    }
    defer file.Close()

    // Write CSV header
    fmt.Fprintln(file, "topic,partition,offset")
    for _, tp := range offsetResults {
        fmt.Fprintf(file, "%s,%d,%d\n", *tp.Topic, tp.Partition, tp.Offset)
    }

    fmt.Println("Offsets exported successfully to topic-offset.csv")
}

2. Import Offsets (Equivalent to --execute --from-file) #

To apply the exported offsets to your target consumer group (cg2), read the CSV and use the AlterConsumerGroupOffsets method:

func importOffsets() {
    admin, err := kafka.NewAdminClient(&kafka.ConfigMap{
        "bootstrap.servers": "your-kafka-broker-list",
    })
    if err != nil {
        panic(fmt.Sprintf("Failed to create admin client: %v", err))
    }
    defer admin.Close()

    targetGroup := "cg2"

    // Read the exported CSV file
    file, err := os.Open("topic-offset.csv")
    if err != nil {
        panic(fmt.Sprintf("Failed to open CSV file: %v", err))
    }
    defer file.Close()

    // Parse CSV records
    reader := csv.NewReader(file)
    records, err := reader.ReadAll()
    if err != nil {
        panic(fmt.Sprintf("Failed to parse CSV: %v", err))
    }

    // Build offset map for the admin API
    offsetMap := make(map[kafka.TopicPartition]kafka.Offset)
    for i, record := range records {
        if i == 0 { // Skip header row
            continue
        }
        topic := record[0]
        partition, _ := strconv.Atoi(record[1])
        offsetVal, _ := strconv.ParseInt(record[2], 10, 64)

        tp := kafka.TopicPartition{
            Topic:     &topic,
            Partition: int32(partition),
        }
        offsetMap[tp] = kafka.Offset(offsetVal)
    }

    // Apply offsets to the target consumer group
    _, err = admin.AlterConsumerGroupOffsets(targetGroup, offsetMap)
    if err != nil {
        panic(fmt.Sprintf("Failed to reset offsets: %v", err))
    }

    fmt.Println("Offsets applied to consumer group cg2 successfully!")
}

Using Sarama (likely what you meant by "sharma") #

Sarama offers robust admin tools for offset management too. Here’s how to replicate the workflow:

1. Export Offsets #

import (
    "fmt"
    "os"
    "strconv"

    "github.com/Shopify/sarama"
)

func exportOffsetsSarama() {
    config := sarama.NewConfig()
    config.Version = sarama.V2_8_0_0 // Match your Kafka cluster version
    admin, err := sarama.NewClusterAdmin([]string{"your-kafka-broker-list"}, config)
    if err != nil {
        panic(fmt.Sprintf("Failed to create admin client: %v", err))
    }
    defer admin.Close()

    targetTopic := "your-target-topic"
    sourceGroup := "cg1"

    // Get topic partitions
    partitions, err := admin.ListPartitions(targetTopic)
    if err != nil {
        panic(fmt.Sprintf("Failed to get topic partitions: %v", err))
    }

    // Fetch group offsets
    offsets, err := admin.ListConsumerGroupOffsets(sourceGroup, map[string][]int32{targetTopic: partitions})
    if err != nil {
        panic(fmt.Sprintf("Failed to fetch group offsets: %v", err))
    }

    // Export to CSV
    file, err := os.Create("topic-offset.csv")
    if err != nil {
        panic(fmt.Sprintf("Failed to create CSV file: %v", err))
    }
    defer file.Close()

    fmt.Fprintln(file, "topic,partition,offset")
    for tp, block := range offsets.Blocks {
        fmt.Fprintf(file, "%s,%d,%d\n", tp.Topic, tp.Partition, block.Offset)
    }

    fmt.Println("Offsets exported successfully via Sarama")
}

2. Import Offsets #

func importOffsetsSarama() {
    config := sarama.NewConfig()
    config.Version = sarama.V2_8_0_0
    admin, err := sarama.NewClusterAdmin([]string{"your-kafka-broker-list"}, config)
    if err != nil {
        panic(fmt.Sprintf("Failed to create admin client: %v", err))
    }
    defer admin.Close()

    targetGroup := "cg2"

    // Read CSV file
    file, err := os.Open("topic-offset.csv")
    if err != nil {
        panic(fmt.Sprintf("Failed to open CSV file: %v", err))
    }
    defer file.Close()

    reader := csv.NewReader(file)
    records, err := reader.ReadAll()
    if err != nil {
        panic(fmt.Sprintf("Failed to parse CSV: %v", err))
    }

    // Build offset map for Sarama's admin API
    offsetMap := make(map[string]map[int32]int64)
    for i, record := range records {
        if i == 0 {
            continue
        }
        topic := record[0]
        partition, _ := strconv.Atoi(record[1])
        offsetVal, _ := strconv.ParseInt(record[2], 10, 64)

        if _, ok := offsetMap[topic]; !ok {
            offsetMap[topic] = make(map[int32]int64)
        }
        offsetMap[topic][int32(partition)] = offsetVal
    }

    // Apply offsets to target group
    err = admin.AlterConsumerGroupOffsets(targetGroup, offsetMap)
    if err != nil {
        panic(fmt.Sprintf("Failed to reset offsets: %v", err))
    }

    fmt.Println("Offsets applied to consumer group cg2 via Sarama successfully!")
}

Key Notes #

• Always match your client library version to your Kafka cluster version to avoid compatibility issues.
• To mimic the --to-current flag (setting offsets to the latest partition end, not just the group’s committed offset), replace the group offset fetch with a call to get partition high watermarks (both libraries have methods for this).
• Add dry-run logic (log intended offsets without applying them) before executing offset resets in production to avoid accidental data loss.

内容的提问来源于stack exchange，提问作者nonus25