fix: use offset instead of OldStyleOffsets

command/healer/cmd/apicontrollers/group.go

@@ -1,7 +1,6 @@
 package apicontrollers
 
 import (
-	"fmt"
 	"net/http"
 
 	"github.com/childe/healer"
@@ -85,16 +84,9 @@ func getOffset(brokers *healer.Brokers, topic, client string) (map[int32]int64,
 
 	rst := make(map[int32]int64)
 	for _, offsetsResponse := range offsetsResponses {
-		for topic, partitionOffsets := range offsetsResponse.TopicPartitionOffsets {
+		for _, partitionOffsets := range offsetsResponse.TopicPartitionOffsets {
 			for _, partitionOffset := range partitionOffsets {
-				if len(partitionOffset.OldStyleOffsets) == 0 {
-					rst[partitionOffset.Partition] = -1
-					continue
-				}
-				if len(partitionOffset.OldStyleOffsets) != 1 {
-					return nil, fmt.Errorf("%s[%d] offsets length mismatch: %v", topic, partitionOffset.Partition, partitionOffset.OldStyleOffsets)
-				}
-				rst[partitionOffset.Partition] = partitionOffset.OldStyleOffsets[0]
+				rst[partitionOffset.Partition] = partitionOffset.GetOffset()
 			}
 		}
 	}

command/healer/cmd/get-offsets.go

@@ -61,16 +61,7 @@ var getOffsetsCmd = &cobra.Command{
 		sort.Slice(allPartitions, func(i, j int) bool { return allPartitions[i].Partition < allPartitions[j].Partition })
 		for _, p := range allPartitions {
 			fmt.Printf("%s:%d:", topic, p.Partition)
-			if len(p.OldStyleOffsets) > 0 {
-				for i, offset := range p.OldStyleOffsets {
-					if i != 0 {
-						fmt.Print(",")
-					}
-					fmt.Printf("%d", offset)
-				}
-			} else {
-				fmt.Printf("%d %d", p.Timestamp, p.Offset)
-			}
+			fmt.Printf("%d %d", p.Timestamp, p.GetOffset())
 			fmt.Println()
 		}
 

command/healer/cmd/get-pending.go

@@ -55,16 +55,9 @@ func getOffset(topic, client string) (map[int32]int64, error) {
 
 	rst := make(map[int32]int64)
 	for _, offsetsResponse := range offsetsResponses {
-		for topic, partitionOffsets := range offsetsResponse.TopicPartitionOffsets {
+		for _, partitionOffsets := range offsetsResponse.TopicPartitionOffsets {
 			for _, partitionOffset := range partitionOffsets {
-				if len(partitionOffset.OldStyleOffsets) == 0 {
-					rst[partitionOffset.Partition] = -1
-					continue
-				}
-				if len(partitionOffset.OldStyleOffsets) != 1 {
-					return nil, fmt.Errorf("%s[%d] offsets length mismatch: %v", topic, partitionOffset.Partition, partitionOffset.OldStyleOffsets)
-				}
-				rst[partitionOffset.Partition] = partitionOffset.OldStyleOffsets[0]
+				rst[partitionOffset.Partition] = partitionOffset.GetOffset()
 			}
 		}
 	}

command/healer/cmd/reset-offset.go

@@ -50,15 +50,10 @@ var resetOffsetCmd = &cobra.Command{
 			if err := offsetsResponse.Error(); err != nil {
 				return fmt.Errorf("request offsets error: %w. topic: %s, timestmap: %d", err, topic, timestamp)
 			}
-			for topic, partitionOffsets := range offsetsResponse.TopicPartitionOffsets {
+			for _, partitionOffsets := range offsetsResponse.TopicPartitionOffsets {
 				for _, partitionOffset := range partitionOffsets {
 					partition := partitionOffset.Partition
-					if len(partitionOffset.OldStyleOffsets) == 0 {
-						return fmt.Errorf("offsets of %s[%d] is blank", topic, partition)
-					}
-					klog.Infof("%s:%d:%v", topic, partition, partitionOffset.OldStyleOffsets)
-					offset := int64(partitionOffset.OldStyleOffsets[0])
-					offsets[partition] = offset
+					offsets[partition] = partitionOffset.GetOffset()
 				}
 			}
 		}

offset_response.go

@@ -12,6 +12,18 @@ type PartitionOffset struct {
 	Timestamp       int64
 	Offset          int64
 }
+
+// get the offset of the given partition from OldStyleOffsets or Offset
+func (p *PartitionOffset) GetOffset() int64 {
+	if len(p.OldStyleOffsets) == 0 {
+		return p.Offset
+	}
+	if len(p.OldStyleOffsets) == 1 {
+		return p.OldStyleOffsets[0]
+	}
+	panic("more than one old style offset found in offset response")
+}
+
 type OffsetsResponse struct {
 	CorrelationID         uint32
 	ThrottleTimeMs        int32

simple_consumer.go

@@ -218,7 +218,7 @@ func (c *SimpleConsumer) getOffset(fromBeginning bool) (int64, error) {
 	if err := offsetsResponse.Error(); err != nil {
 		return -1, err
 	}
-	return int64(offsetsResponse.TopicPartitionOffsets[c.topic][0].OldStyleOffsets[0]), nil
+	return int64(offsetsResponse.TopicPartitionOffsets[c.topic][0].GetOffset()), nil
 }
 
 func (c *SimpleConsumer) getCommitedOffet() error {