如何通过Sarama ClusterAdmin获取日志末尾偏移量（无需消费者）

没问题，你可以通过Sarama ClusterAdmin的GetOffset方法获取指定Topic的末尾偏移量，无需启动消费者实例。

实现步骤 #

1. 先获取目标Topic的所有分区信息：通过ClusterAdmin的DescribeTopics方法拿到Topic的分区列表，确保覆盖所有分区。
2. 遍历每个分区调用GetOffset：传入sarama.OffsetNewest作为时间戳参数，该参数对应Kafka的最新偏移量（末尾偏移量）。

代码示例 #

package main

import (
	"fmt"
	"github.com/Shopify/sarama"
)

func main() {
	// 初始化Kafka配置
	config := sarama.NewConfig()
	config.Version = sarama.V2_8_0_0 // 匹配你的Kafka集群版本

	// 创建ClusterAdmin实例
	admin, err := sarama.NewClusterAdmin([]string{"localhost:9092"}, config)
	if err != nil {
		panic(fmt.Sprintf("Failed to create ClusterAdmin: %v", err))
	}
	defer admin.Close()

	targetTopic := "your-topic-name"
	// 获取目标Topic的详细信息（含分区列表）
	topicsInfo, err := admin.DescribeTopics([]string{targetTopic})
	if err != nil {
		panic(fmt.Sprintf("Failed to describe topic %s: %v", targetTopic, err))
	}

	if len(topicsInfo) == 0 {
		fmt.Printf("Topic %s does not exist\n", targetTopic)
		return
	}

	// 遍历每个分区获取末尾偏移量
	topicDetail := topicsInfo[0]
	for _, partition := range topicDetail.Partitions {
		newestOffset, err := admin.GetOffset(targetTopic, partition.ID, sarama.OffsetNewest)
		if err != nil {
			fmt.Printf("Failed to get newest offset for partition %d: %v\n", partition.ID, err)
			continue
		}
		fmt.Printf("Partition %d: newest offset = %d\n", partition.ID, newestOffset)
	}
}

补充说明 #

• GetOffset方法底层调用Kafka的OffsetFetchRequest API，通过指定OffsetNewest（对应Kafka协议中的-1时间戳）直接获取分区的最新偏移量，完全独立于消费者组逻辑。
• 如果你已经明确知道Topic的分区数量，也可以直接遍历分区ID调用GetOffset，但用DescribeTopics能动态获取当前Topic的所有分区，避免因分区扩容导致的遗漏。

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