groups

Documentation

Index

• Constants
• func FormatGroupCoordinators(groupCoordinators []GroupCoordinator) string
• func FormatGroupMembers(members []MemberInfo, full bool) string
• func FormatMemberLags(memberLags []MemberPartitionLag, full bool) string
• func FormatMemberPartitionCounts(members []MemberInfo) string
• func FormatPartitionOffsets(partitionOffsets map[int]int64) string
• func GetEarliestOrLatestOffset(ctx context.Context, connector *admin.Connector, topic string, strategy string, ...) (int64, error)
• func ResetOffsets(ctx context.Context, connector *admin.Connector, topic string, groupID string, ...) error
• type GroupCoordinator
  • func GetGroups(ctx context.Context, connector *admin.Connector) ([]GroupCoordinator, error)
• type GroupDetails
  • func GetGroupDetails(ctx context.Context, connector *admin.Connector, groupID string) (*GroupDetails, error)
  • func (g GroupDetails) PartitionMembers(topic string) map[int]MemberInfo
  • func (g GroupDetails) TopicsMap() map[string]struct{}
• type MemberInfo
  • func (m MemberInfo) Topics() []string
• type MemberPartitionLag
  • func GetMemberLags(ctx context.Context, connector *admin.Connector, topic string, groupID string) ([]MemberPartitionLag, error)
  • func (m MemberPartitionLag) OffsetLag() int64
  • func (m MemberPartitionLag) TimeLag() time.Duration

Constants

const (
	LatestResetOffsetsStrategy   string = "latest"
	EarliestResetOffsetsStrategy string = "earliest"
)

Consumer Group Offset reset strategies

Functions

func FormatGroupCoordinators

func FormatGroupCoordinators(groupCoordinators []GroupCoordinator) string

FormatGroupCoordinators generates a pretty table from the results of a call to GetGroups.

func FormatGroupMembers

func FormatGroupMembers(members []MemberInfo, full bool) string

FormatGroupMembers generates a pretty table from a slice of MemberInfo details.

func FormatMemberLags

func FormatMemberLags(memberLags []MemberPartitionLag, full bool) string

FormatMemberLags generates a pretty table from the results of GetMemberLags.

func FormatMemberPartitionCounts

func FormatMemberPartitionCounts(members []MemberInfo) string

FormatMemberPartitionCounts generates a pretty table with the details on how many partitions group members are consuming from.

func FormatPartitionOffsets

func FormatPartitionOffsets(partitionOffsets map[int]int64) string

FormatPartitionOffsets generates a pretty table that shows the proposed offsets for each partition in a reset.

func GetEarliestOrLatestOffset

func GetEarliestOrLatestOffset(
	ctx context.Context,
	connector *admin.Connector,
	topic string,
	strategy string,
	partition int,
) (int64, error)

GetEarliestorLatestOffset gets earliest/latest offset for a given topic partition for resetting offsets of consumer group

func ResetOffsets

func ResetOffsets(
	ctx context.Context,
	connector *admin.Connector,
	topic string,
	groupID string,
	partitionOffsets map[int]int64,
) error

ResetOffsets updates the offsets for a given topic / group combination.

Types

type GroupCoordinator

type GroupCoordinator struct {
	GroupID     string
	Coordinator int
	Topics      []string
}

GroupCoordinator stores the coordinator broker for a single consumer group.

func GetGroups

func GetGroups(
	ctx context.Context,
	connector *admin.Connector,
) ([]GroupCoordinator, error)

GetGroups fetches and returns information about all consumer groups in the cluster.

type GroupDetails

type GroupDetails struct {
	GroupID string
	State   string
	Members []MemberInfo
}

GroupDetails stores the state and members for a consumer group.

func GetGroupDetails

func GetGroupDetails(
	ctx context.Context,
	connector *admin.Connector,
	groupID string,
) (*GroupDetails, error)

GetGroupDetails returns the details (membership, etc.) for a single consumer group.

func (GroupDetails) PartitionMembers

func (g GroupDetails) PartitionMembers(topic string) map[int]MemberInfo

PartitionMembers returns the members for each partition in the argument topic.

func (GroupDetails) TopicsMap

func (g GroupDetails) TopicsMap() map[string]struct{}

TopicsMap returns a map of all the topics consumed by the current group.

type MemberInfo

type MemberInfo struct {
	MemberID        string
	ClientID        string
	ClientHost      string
	TopicPartitions map[string][]int
}

MemberInfo stores information about a single consumer group member.

func (MemberInfo) Topics

func (m MemberInfo) Topics() []string

Topics returns a slice of all topics that the current MemberInfo is consuming from.

type MemberPartitionLag

type MemberPartitionLag struct {
	Topic        string
	Partition    int
	MemberID     string
	NewestOffset int64
	NewestTime   time.Time
	MemberOffset int64
	MemberTime   time.Time
}

MemberPartitionLag information about the lag for a single topic / partition / group member combination.

func GetMemberLags

func GetMemberLags(
	ctx context.Context,
	connector *admin.Connector,
	topic string,
	groupID string,
) ([]MemberPartitionLag, error)

GetMemberLags returns the lag for each partition being consumed by the argument group in the argument topic.

func (MemberPartitionLag) OffsetLag

func (m MemberPartitionLag) OffsetLag() int64

OffsetLag returns the difference between the latest offset in the partition and the latest one committed by the group member.

func (MemberPartitionLag) TimeLag

func (m MemberPartitionLag) TimeLag() time.Duration

TimeLag returns the time difference between the latest timestamp in the the partition and the timestamp in the latest message committed by the group member.