consumer

May 8, 2019

Changes in this version

+ type ClusterGroup struct

+ Cluster string

+ Group string

type ErrorQType

+ var DLQConsumerGroupNameSuffix = "-dlq-merger"

type Options

+ ClientID string

+ TLSConfig *tls.Config

type Topic

+ ConsumerGroupSuffix string

Aug 8, 2018

Changes in this version

+ type ErrorQType string

+ var DLQErrorQType ErrorQType = "DLQ"

+ var RetryQErrorQType ErrorQType = "retryQ"

type Message

+ func (m *Message) NackToDLQ() error

Jul 6, 2018

Changes in this version

type Options

+ FetchDefaultBytes int32

+ ProducerMaxMessageByes int

type PartitionConsumer

+ func NewPartitionConsumerWithoutCommit(topic Topic, sarama SaramaConsumer, pConsumer cluster.PartitionConsumer, ...) PartitionConsumer

Mar 31, 2018

Changes in this version

type DLQMetadata

+ func NoopDLQMetadataDecoder(b []byte) (*DLQMetadata, error)

+ func ProtobufDLQMetadataDecoder(b []byte) (*DLQMetadata, error)

+ type DLQMetadataDecoder func([]byte) (*DLQMetadata, error)

Mar 9, 2018

Changes in this version

type Options

+ OtherConsumerTopics []Topic

Mar 5, 2018

Changes in this version

+ func NewSaramaClient(brokers []string, config *sarama.Config) (sarama.Client, error)

+ func NewSaramaProducer(client sarama.Client) (sarama.AsyncProducer, error)

+ type ClusterConsumer struct

+ func NewClusterConsumer(cluster string, saramaConsumer SaramaConsumer, ...) *ClusterConsumer

+ func (c *ClusterConsumer) Closed() <-chan struct{}

+ func (c *ClusterConsumer) ResetOffset(topic string, partition int32, offsetRange kafka.OffsetRange) error

+ func (c *ClusterConsumer) Start() error

+ func (c *ClusterConsumer) Stop()

+ type Constructors struct

+ NewSaramaClient func([]string, *sarama.Config) (sarama.Client, error)

+ NewSaramaConsumer func([]string, string, []string, *cluster.Config) (SaramaConsumer, error)

+ NewSaramaProducer func(sarama.Client) (sarama.AsyncProducer, error)

+ type DLQ interface

+ Add func(m kafka.Message) error

+ Start func() error

+ Stop func()

+ func NewBufferedDLQ(topic kafka.Topic, producer sarama.AsyncProducer, scope tally.Scope, ...) DLQ

+ func NewNoopDLQ() DLQ

+ func NewRetryDLQMultiplexer(retryTopic, dlqTopic DLQ, threshold int64) DLQ

+ type DLQMetadata struct

+ Data []byte

+ Offset int64

+ Partition int32

+ RetryCount int64

+ TimestampNs int64

+ Topic string

+ func (*DLQMetadata) Descriptor() ([]byte, []int)

+ func (*DLQMetadata) ProtoMessage()

+ func (m *DLQMetadata) GetData() []byte

+ func (m *DLQMetadata) GetOffset() int64

+ func (m *DLQMetadata) GetPartition() int32

+ func (m *DLQMetadata) GetRetryCount() int64

+ func (m *DLQMetadata) GetTimestampNs() int64

+ func (m *DLQMetadata) GetTopic() string

+ func (m *DLQMetadata) Reset()

+ func (m *DLQMetadata) String() string

+ type Message struct

+ func (m *Message) Ack() error

+ func (m *Message) Key() (key []byte)

+ func (m *Message) MarshalLogObject(e zapcore.ObjectEncoder) error

+ func (m *Message) Nack() error

+ func (m *Message) Offset() int64

+ func (m *Message) Partition() int32

+ func (m *Message) RetryCount() int64

+ func (m *Message) Timestamp() time.Time

+ func (m *Message) Topic() string

+ func (m *Message) Value() []byte

+ type MultiClusterConsumer struct

+ func NewMultiClusterConsumer(groupName string, topics kafka.ConsumerTopicList, ...) *MultiClusterConsumer

+ func (c *MultiClusterConsumer) Closed() <-chan struct{}

+ func (c *MultiClusterConsumer) MergeDLQ(topic kafka.ConsumerTopic, offsetRanges map[int32]kafka.OffsetRange) error

+ func (c *MultiClusterConsumer) Messages() <-chan kafka.Message

+ func (c *MultiClusterConsumer) Name() string

+ func (c *MultiClusterConsumer) ResetOffset(cluster, topic string, partition int32, offsetRange kafka.OffsetRange) error

+ func (c *MultiClusterConsumer) Start() error

+ func (c *MultiClusterConsumer) Stop()

+ func (c *MultiClusterConsumer) Topics() kafka.ConsumerTopicList

+ type Options struct

+ Concurrency int

+ ConsumerMode cluster.ConsumerMode

+ MaxProcessingTime time.Duration

+ OffsetCommitInterval time.Duration

+ OffsetPolicy int64

+ PartitionRcvBufferSize int

+ RangeConsumerTopics []Topic

+ RcvBufferSize int

+ RebalanceDwellTime time.Duration

+ func DefaultOptions() *Options

+ type PartitionConsumer interface

+ Drain func(time.Duration)

+ ResetOffset func(kafka.OffsetRange) error

+ Start func() error

+ Stop func()

+ func NewPartitionConsumer(topic Topic, sarama SaramaConsumer, pConsumer cluster.PartitionConsumer, ...) PartitionConsumer

+ func NewRangePartitionConsumer(topic Topic, sarama SaramaConsumer, pConsumer cluster.PartitionConsumer, ...) PartitionConsumer

+ type PartitionConsumerFactory func(topic Topic, sarama SaramaConsumer, pConsumer cluster.PartitionConsumer, ...) PartitionConsumer

+ type SaramaConsumer interface

+ Close func() error

+ CommitOffsets func() error

+ Errors func() <-chan error

+ HighWaterMarks func() map[string]map[int32]int64

+ MarkOffset func(msg *sarama.ConsumerMessage, metadata string)

+ MarkPartitionOffset func(topic string, partition int32, offset int64, metadata string)

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

+ Notifications func() <-chan *cluster.Notification

+ Partitions func() <-chan cluster.PartitionConsumer

+ ResetPartitionOffset func(topic string, partition int32, offset int64, metadata string)

+ func NewSaramaConsumer(brokers []string, groupID string, topics []string, config *cluster.Config) (SaramaConsumer, error)

+ type Topic struct

+ func (t Topic) MarshalLogObject(e zapcore.ObjectEncoder) error

+ type TopicConsumer struct

+ func NewTopicConsumer(topic Topic, msgC chan kafka.Message, consumer SaramaConsumer, dlq DLQ, ...) *TopicConsumer

+ func (c *TopicConsumer) ResetOffset(partition int32, offsetRange kafka.OffsetRange) error

+ func (c *TopicConsumer) Start() error

+ func (c *TopicConsumer) Stop()

+ type TopicType int

+ const TopicTypeDLQ

+ const TopicTypeDefaultQ

+ const TopicTypeRetryQ

+ func (t TopicType) String() string