gonsumer

type Client interface {
	// Fetch is responsible for fetching messages for given topic, partition and offset from Kafka broker.
	// Leader change handling happens inside Fetch and is hidden from user so he should not handle such cases.
	// Returns a fetch response and error if it occurred.
	Fetch(topic string, partition int32, offset int64) (*client.FetchResponse, error)

	// GetAvailableOffset issues an offset request to a specified topic and partition with a given offset time.
	// Returns an offet for given topic, partition and offset time and an error if it occurs.
	GetAvailableOffset(topic string, partition int32, offsetTime int64) (int64, error)

	// GetOffset gets the latest committed offset for a given group, topic and partition from Kafka.
	// Returns an offset and an error if it occurs.
	GetOffset(group string, topic string, partition int32) (int64, error)

	// CommitOffset commits the given offset for a given group, topic and partition to Kafka.
	// Returns an error if commit was unsuccessful.
	CommitOffset(group string, topic string, partition int32, offset int64) error
}

type Consumer interface {
	// Add adds a topic/partition to consume for this consumer and starts consuming it immediately.
	// Returns an error if PartitionConsumer for this topic/partition already exists.
	Add(topic string, partition int32) error

	// Remove stops consuming a topic/partition by this consumer once it is done with the current batch.
	// This means the PartitionConsumer will stop accepting new batches but will have a chance to finish its current work.
	// Returns an error if PartitionConsumer for this topic/partition does not exist.
	Remove(topic string, partition int32) error

	// Assignment returns a map of topic/partitions being consumed at the moment by this consumer.
	// The keys are topic names and values are slices of partitions.
	Assignment() map[string][]int32

	// Offset returns the current consuming offset for a given topic/partition.
	// Please note that this value does not correspond to the latest committed offset but the latest fetched offset.
	// This call will return an error if the PartitionConsumer for given topic/partition does not exist.
	Offset(topic string, partition int32) (int64, error)

	// Commit commits the given offset for a given topic/partition to Kafka.
	// Returns an error if the commit was unsuccessful.
	Commit(topic string, partition int32, offset int64) error

	// SetOffset overrides the current fetch offset value for given topic/partition.
	// This does not commit offset but allows you to move back and forth throughout the partition.
	// Returns an error if the PartitionConsumer for this topic/partition does not exist.
	SetOffset(topic string, partition int32, offset int64) error

	// Lag returns the difference between the latest available offset in the partition and the
	// latest fetched offset by this consumer. This allows you to see how much behind the consumer is.
	// Returns lag value for a given topic/partition and an error if the PartitionConsumer for given
	// topic/partition does not exist.
	Lag(topic string, partition int32) (int64, error)

	// Stop stops consuming all topics and partitions with this consumer.
	Stop()

	// AwaitTermination blocks until Stop() is called.
	AwaitTermination()

	// ConsumerMetrics returns a metrics structure for this consumer. An error is returned if metrics are disabled.
	ConsumerMetrics() (ConsumerMetrics, error)

	// PartitionConsumerMetrics returns a metrics structure for a given topic and partition. An error is returned
	// if metrics are disabled or PartitionConsumer for given topic and partition does not exist
	PartitionConsumerMetrics(topic string, partition int32) (PartitionConsumerMetrics, error)

	// AllMetrics returns metrics registries for this consumer and all its PartitionConsumers. An error is returned
	// if metrics are disabled.
	AllMetrics() (*Metrics, error)

	// Join blocks until consumer has at least one topic/partition to consume, e.g. until len(Assignment()) > 0.
	Join()
}

type ConsumerConfig struct {
	// Group is a string that uniquely identifies a set of consumers within the same consumer group.
	Group string

	// ConsumerID is a string that uniquely identifies a consumer within a consumer group.
	// Defaults to a random UUID.
	ConsumerID string

	// KeyDecoder is a function that turns plain bytes into a decoded message key.
	KeyDecoder Decoder

	// ValueDecoder is a function that turns plain bytes into a decoded message value.
	ValueDecoder Decoder

	// AutoOffsetReset defines what to do when there is no committed offset or committed offset is out of range.
	// kafka-client.EarliestTime - automatically reset the offset to the smallest offset.
	// kafka-client.LatestTime - automatically reset the offset to the largest offset.
	// Defaults to kafka-client.EarliestTime.
	AutoOffsetReset int64

	// AutoCommitEnable determines whether the consumer will automatically commit offsets after each batch
	// is finished (e.g. the call to strategy function returns). Turned off by default.
	AutoCommitEnable bool

	// EnableMetrics determines whether the consumer will collect all kinds of metrics to better understand what's
	// going on under the hood. Turned off by default as it may significantly affect performance.
	EnableMetrics bool

	// Backoff between attempts to initialize consumer offset.
	InitOffsetBackoff time.Duration
}

type ConsumerMetrics interface {
	// NumOwnedTopicPartitions is a counter which value is the number of currently owned topic partitions by
	// enclosing Consumer.
	NumOwnedTopicPartitions(func(metrics.Counter))

	// Registry provides access to metrics registry for enclosing Consumer.
	Registry() metrics.Registry

	// Stop unregisters all metrics from the registry.
	Stop()
}

type Decoder func(raw []byte) (interface{}, error)

type FetchData struct {
	// Messages is a slice of actually fetched messages from Kafka.
	Messages []*MessageAndMetadata

	// HighwaterMarkOffset is an offset in the end of topic/partition this FetchData comes from.
	HighwaterMarkOffset int64

	// Error is an error that occurs both on fetch level or topic/partition level.
	Error error
}

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

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

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

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

type MessageAndMetadata struct {
	// Key is a raw message key.
	Key []byte

	// Value is a raw message value.
	Value []byte

	// Topic is a Kafka topic this message comes from.
	Topic string

	// Partition is a Kafka partition this message comes from.
	Partition int32

	// Offset is an offset for this message.
	Offset int64

	// DecodedKey is a message key processed by KeyDecoder.
	DecodedKey interface{}

	// DecodedValue is a message value processed by ValueDecoder.
	DecodedValue interface{}
}

type Metrics struct {
	// Consumer is all metrics for enclosing Consumer instance.
	Consumer ConsumerMetrics

	// PartitionConsumers is a map of topic/partitions to PartitionConsumer metrics.
	PartitionConsumers map[string]map[int32]PartitionConsumerMetrics
}

type PartitionConsumer interface {
	// Start starts consuming given topic/partition.
	Start()

	// Stop stops consuming given topic/partition.
	Stop()

	// Offset returns the last fetched offset for this partition consumer.
	Offset() int64

	// Commit commits the given offset to Kafka. Returns an error on unsuccessful commit.
	Commit(offset int64) error

	// SetOffset overrides the current fetch offset value for given topic/partition.
	// This does not commit offset but allows you to move back and forth throughout the partition.
	SetOffset(offset int64)

	// Lag returns the difference between the latest available offset in the partition and the
	// latest fetched offset by this consumer. This allows you to see how much behind the consumer is.
	Lag() int64

	// Metrics returns a metrics structure for this partition consumer. An error is returned if metrics are disabled.
	Metrics() (PartitionConsumerMetrics, error)
}

type PartitionConsumerMetrics interface {
	// BatchDuration is a timer that measures time to process a single batch of data from Kafka broker by enclosing PartitionConsumer.
	BatchDuration(func(metrics.Timer))

	// FetchDuration is a timer that measures time to fetch from Kafka broker by enclosing PartitionConsumer.
	FetchDuration(func(metrics.Timer))

	// NumFetches is a counter with a total number of fetches done by enclosing PartitionConsumer.
	NumFetches(func(metrics.Counter))

	// NumFailedFetches is a counter with a number of failed fetches done by enclosing PartitionConsumer.
	NumFailedFetches(func(metrics.Counter))

	// NumEmptyFetches is a counter with a number of fetches that returned 0 messages done by enclosing PartitionConsumer.
	NumEmptyFetches(func(metrics.Counter))

	// NumFetchedMessages is a counter with a total number of fetched messages by enclosing PartitionConsumer.
	NumFetchedMessages(func(metrics.Counter))

	// NumOffsetCommits is a counter with a total number of offset commits done by enclosing PartitionConsumer.
	NumOffsetCommits(func(metrics.Counter))

	// NumFailedOffsetCommits is a counter with a number of failed offset commits done by enclosing PartitionConsumer.
	NumFailedOffsetCommits(func(metrics.Counter))

	// Lag is a gauge with a current lag value for enclosing PartitionConsumer.
	Lag(func(metrics.Gauge))

	// Registry provides access to metrics registry for enclosing PartitionConsumer.
	Registry() metrics.Registry

	// Stop unregisters all metrics from the registry.
	Stop()
}

type Strategy func(data *FetchData, consumer *KafkaPartitionConsumer)