consumergroup

Jun 4, 2021

Changes in this version

type ConsumerGroup

+ func DefaultConsumerGroup(name string, topics []string, zookeeper []string, config *Config) (cg *ConsumerGroup, err error)

Apr 8, 2021

Changes in this version

+ var AlreadyClosing = errors.New("The consumer group is already shutting down.")

+ var UncleanClose = errors.New("Not all offsets were committed before shutdown was completed")

+ type Config struct

+ Offsets struct{ ... }

+ Zookeeper *kazoo.Config

+ func NewConfig() *Config

+ func (cgc *Config) Validate() error

+ type ConsumerGroup struct

+ func JoinConsumerGroup(name string, topics []string, zookeeper []string, config *Config) (cg *ConsumerGroup, err error)

+ func (cg *ConsumerGroup) Close() error

+ func (cg *ConsumerGroup) Closed() bool

+ func (cg *ConsumerGroup) CommitUpto(message *sarama.ConsumerMessage) error

+ func (cg *ConsumerGroup) Errors() <-chan error

+ func (cg *ConsumerGroup) FlushOffsets() error

+ func (cg *ConsumerGroup) InstanceRegistered() (bool, error)

+ func (cg *ConsumerGroup) Logf(format string, args ...interface{})

+ func (cg *ConsumerGroup) Messages() <-chan *sarama.ConsumerMessage

+ type OffsetManager interface

+ Close func() error

+ FinalizePartition func(topic string, partition int32, lastOffset int64, timeout time.Duration) error

+ Flush func() error

+ InitializePartition func(topic string, partition int32) (int64, error)

+ MarkAsProcessed func(topic string, partition int32, offset int64) bool

+ func NewZookeeperOffsetManager(cg *ConsumerGroup, config *OffsetManagerConfig) OffsetManager

+ type OffsetManagerConfig struct

+ CommitInterval time.Duration

+ VerboseLogging bool

+ func NewOffsetManagerConfig() *OffsetManagerConfig