offset

package

v0.7.1 Latest Latest Go to latest Published: Apr 1, 2026 License: Apache-2.0 Imports: 4 Imported by: 0

Details

This section is empty.

This section is empty.

func ComputeTotalLag(src, dst map[int32]int64) int64

ComputeTotalLag computes the total offset lag across all partitions. For partitions missing from dst, the full source offset counts as lag.

func SortedPartitionIDs(src, dst map[int32]int64) []int32

SortedPartitionIDs returns the union of partition IDs from two offset maps, sorted ascending.

type Provider interface {
	Get(topic string) (map[int32]int64, error)
}

Provider abstracts offset retrieval for testability.

type Service struct {
	// contains filtered or unexported fields
}

Service provides offset operations against a Kafka cluster.

func NewOffsetService(client sarama.Client) *Service

NewOffsetService creates a Service backed by the given Kafka client.

func (t *Service) Close() error

Close closes the underlying Kafka client.

func (t *Service) Exists(topic string) (bool, error)

Exists checks whether a topic exists on the cluster by refreshing metadata.

func (t *Service) Get(topic string) (map[int32]int64, error)

Get fetches the log end offset (LEO) for every partition of a topic. Requests are batched by leader broker for efficiency.