group

package

Go to main page

Versions in this module

v1

v1.1.0

Mar 15, 2026

Changes in this version

type GroupCoordinator

+ func (gc *GroupCoordinator) PerformAssignment(groupID string, partitions []TopicPartition) (map[string][]TopicPartition, error)

+ type MemberSubscription struct

+ MemberID string

+ Topics []string

type OffsetAndMetadata

+ LeaderEpoch int64

+ type PartitionAssignor interface

+ Assign func(members []MemberSubscription, partitions []TopicPartition) map[string][]TopicPartition

+ Name func() string

+ func GetAssignor(name string) PartitionAssignor

+ type RangeAssignor struct

+ func (r *RangeAssignor) Assign(members []MemberSubscription, partitions []TopicPartition) map[string][]TopicPartition

+ func (r *RangeAssignor) Name() string

+ type RoundRobinAssignor struct

+ func (rr *RoundRobinAssignor) Assign(members []MemberSubscription, partitions []TopicPartition) map[string][]TopicPartition

+ func (rr *RoundRobinAssignor) Name() string

+ type StickyAssignor struct

+ func (s *StickyAssignor) Assign(members []MemberSubscription, partitions []TopicPartition) map[string][]TopicPartition

+ func (s *StickyAssignor) AssignWithPrevious(members []MemberSubscription, partitions []TopicPartition, ...) map[string][]TopicPartition

+ func (s *StickyAssignor) Name() string

+ type TopicPartition struct

+ Partition int32

+ Topic string

v1.0.1

Nov 22, 2025

Changes in this version

+ const ErrorCodeGroupAuthorizationFailed

+ const ErrorCodeGroupIDNotFound

+ const ErrorCodeIllegalGeneration

+ const ErrorCodeInvalidSessionTimeout

+ const ErrorCodeNone

+ const ErrorCodeRebalanceInProgress

+ const ErrorCodeUnknown

+ const ErrorCodeUnknownMemberID

+ type CoordinatorConfig struct

+ DefaultRebalanceTimeoutMs int32

+ DefaultSessionTimeoutMs int32

+ HeartbeatCheckIntervalMs int32

+ MaxSessionTimeoutMs int32

+ MinSessionTimeoutMs int32

+ OffsetRetentionMs int64

+ func DefaultCoordinatorConfig() CoordinatorConfig

+ type GroupCoordinator struct

+ func NewGroupCoordinator(storage OffsetStorage, config CoordinatorConfig) *GroupCoordinator

+ func (gc *GroupCoordinator) GetCommittedOffset(groupID string, topic string, partition int32) (int64, error)

+ func (gc *GroupCoordinator) GetGroup(groupID string) *GroupMetadata

+ func (gc *GroupCoordinator) HandleHeartbeat(req *HeartbeatRequest) (*HeartbeatResponse, error)

+ func (gc *GroupCoordinator) HandleJoinGroup(req *JoinGroupRequest) (*JoinGroupResponse, error)

+ func (gc *GroupCoordinator) HandleLeaveGroup(req *LeaveGroupRequest) (*LeaveGroupResponse, error)

+ func (gc *GroupCoordinator) HandleOffsetCommit(req *OffsetCommitRequest) (*OffsetCommitResponse, error)

+ func (gc *GroupCoordinator) HandleOffsetFetch(req *OffsetFetchRequest) (*OffsetFetchResponse, error)

+ func (gc *GroupCoordinator) HandleSyncGroup(req *SyncGroupRequest) (*SyncGroupResponse, error)

+ func (gc *GroupCoordinator) ListGroups() []*GroupMetadata

+ func (gc *GroupCoordinator) Stop() error

+ type GroupError struct

+ Code int16

+ func (e *GroupError) Error() string

+ type GroupMetadata struct

+ GenerationID int32

+ GroupID string

+ LeaderID string

+ Members map[string]*MemberMetadata

+ ProtocolName string

+ ProtocolType string

+ RebalanceTimeoutMs int32

+ SessionTimeoutMs int32

+ State GroupState

+ StateTimestamp time.Time

+ type GroupOffsets struct

+ GroupID string

+ Offsets map[string]map[int32]*OffsetAndMetadata

+ type GroupState string

+ const GroupStateCompletingRebalance

+ const GroupStateDead

+ const GroupStateEmpty

+ const GroupStatePreparingRebalance

+ const GroupStateStable

+ type HeartbeatRequest struct

+ GenerationID int32

+ GroupID string

+ MemberID string

+ type HeartbeatResponse struct

+ ErrorCode int16

+ type JoinGroupMember struct

+ MemberID string

+ Metadata []byte

+ type JoinGroupRequest struct

+ ClientID string

+ GroupID string

+ MemberID string

+ ProtocolType string

+ Protocols []ProtocolMetadata

+ RebalanceTimeoutMs int32

+ SessionTimeoutMs int32

+ type JoinGroupResponse struct

+ ErrorCode int16

+ GenerationID int32

+ LeaderID string

+ MemberID string

+ Members []JoinGroupMember

+ ProtocolName string

+ type LeaveGroupRequest struct

+ GroupID string

+ MemberID string

+ type LeaveGroupResponse struct

+ ErrorCode int16

+ type MemberAssignment struct

+ Partitions map[string][]int32

+ UserData []byte

+ Version int16

+ type MemberAssignmentData struct

+ Assignment []byte

+ MemberID string

+ type MemberMetadata struct

+ Assignment *MemberAssignment

+ ClientHost string

+ ClientID string

+ JoinTime time.Time

+ LastHeartbeat time.Time

+ MemberID string

+ ProtocolMetadata []byte

+ RebalanceTimeoutMs int32

+ SessionTimeoutMs int32

+ State MemberState

+ Subscription []string

+ type MemberState string

+ const MemberStateAwaitingSync

+ const MemberStateJoining

+ const MemberStateLeaving

+ const MemberStateStable

+ type MemoryOffsetStorage struct

+ func NewMemoryOffsetStorage() *MemoryOffsetStorage

+ func (s *MemoryOffsetStorage) DeleteOffsets(groupID string) error

+ func (s *MemoryOffsetStorage) FetchOffset(groupID string, topic string, partition int32) (*OffsetAndMetadata, error)

+ func (s *MemoryOffsetStorage) FetchOffsets(groupID string) (*GroupOffsets, error)

+ func (s *MemoryOffsetStorage) StoreOffset(groupID string, topic string, partition int32, offset *OffsetAndMetadata) error

+ type MetadataStore interface

+ DeleteGroupOffsets func(groupID string) error

+ FetchGroupOffset func(groupID string, topic string, partition int32) (*OffsetAndMetadata, error)

+ FetchGroupOffsets func(groupID string) (*GroupOffsets, error)

+ StoreGroupOffset func(groupID string, topic string, partition int32, offset *OffsetAndMetadata) error

+ type OffsetAndMetadata struct

+ CommitTime time.Time

+ ExpireTime time.Time

+ Metadata string

+ Offset int64

+ type OffsetCommitData struct

+ Metadata string

+ Offset int64

+ type OffsetCommitLog struct

+ GroupID string

+ MemberID string

+ Metadata string

+ Offset int64

+ Partition int32

+ Timestamp int64

+ Topic string

+ type OffsetCommitRequest struct

+ GenerationID int32

+ GroupID string

+ MemberID string

+ Offsets map[string]map[int32]OffsetCommitData

+ type OffsetCommitResponse struct

+ Errors map[string]map[int32]int16

+ type OffsetFetchData struct

+ ErrorCode int16

+ Metadata string

+ Offset int64

+ type OffsetFetchRequest struct

+ GroupID string

+ Topics map[string][]int32

+ type OffsetFetchResponse struct

+ Offsets map[string]map[int32]OffsetFetchData

+ type OffsetManager struct

+ func NewOffsetManager(storage OffsetStorage) *OffsetManager

+ func (m *OffsetManager) CommitOffset(groupID string, topic string, partition int32, offset int64, metadata string) error

+ func (m *OffsetManager) GetAllOffsets(groupID string) (map[string]map[int32]int64, error)

+ func (m *OffsetManager) GetOffset(groupID string, topic string, partition int32) (int64, error)

+ func (m *OffsetManager) ResetOffsets(groupID string) error

+ func (m *OffsetManager) ValidateOffset(offset int64, minOffset int64, maxOffset int64) error

+ type OffsetStorage interface

+ DeleteOffsets func(groupID string) error

+ FetchOffset func(groupID string, topic string, partition int32) (*OffsetAndMetadata, error)

+ FetchOffsets func(groupID string) (*GroupOffsets, error)

+ StoreOffset func(groupID string, topic string, partition int32, offset *OffsetAndMetadata) error

+ type PersistentOffsetStorage struct

+ func NewPersistentOffsetStorage(store MetadataStore) *PersistentOffsetStorage

+ func (s *PersistentOffsetStorage) DeleteOffsets(groupID string) error

+ func (s *PersistentOffsetStorage) FetchOffset(groupID string, topic string, partition int32) (*OffsetAndMetadata, error)

+ func (s *PersistentOffsetStorage) FetchOffsets(groupID string) (*GroupOffsets, error)

+ func (s *PersistentOffsetStorage) StoreOffset(groupID string, topic string, partition int32, offset *OffsetAndMetadata) error

+ type ProtocolMetadata struct

+ Metadata []byte

+ Name string

+ type SyncGroupRequest struct

+ Assignments []MemberAssignmentData

+ GenerationID int32

+ GroupID string

+ MemberID string

+ type SyncGroupResponse struct

+ Assignment []byte

+ ErrorCode int16

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL