offset

Documentation

Index

• type ConcurrentList
• type Manager
  • func NewManager(minOffset int64, markOffset MarkOffset, topic string, partition int32, ...) *Manager
  • func (m *Manager) Close() error
  • func (m *Manager) MarkOffset(offset int64)
  • func (m *Manager) Start()
• type MarkOffset

Types

type ConcurrentList

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

ConcurrentList is a list that maintains kafka offsets with thread-safe Insert and setToHighestContiguous operations

type Manager

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

Manager accepts kafka offsets and commits them using the provided kafka consumer

The Manager is designed to be used in a scenario where the consumption of kafka offsets is decoupled from the processing of offsets asynchronously via goroutines. This breaks the ordering guarantee which could result in the completion of processing of an earlier message after the processing of a later message.

It assumes that Kafka offsets are sequential and monotonically increasing[1], and maintains sorted lists of offsets per partition.

[1] https://kafka.apache.org/0100/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html

func NewManager

func NewManager(
	minOffset int64,
	markOffset MarkOffset,
	topic string,
	partition int32,
	factory metrics.Factory,
) *Manager

NewManager creates a new Manager

func (*Manager) Close

func (m *Manager) Close() error

Close closes the Manager

func (*Manager) MarkOffset

func (m *Manager) MarkOffset(offset int64)

MarkOffset marks the offset of a consumer message

func (*Manager) Start

func (m *Manager) Start()

Start starts the Manager

type MarkOffset

type MarkOffset func(offset int64)

MarkOffset is a func that marks offsets in Kafka