kafka

Documentation

Index

• func CreateEarliestConsumer(bootstrap string, topic string, group string, tracker *progress.Tracker) *kafka.Consumer
• func CreateLatestConsumer(bootstrap string, topic string, group string, tracker *progress.Tracker) *kafka.Consumer
• func GetPartitions(consumer *kafka.Consumer, topic string, tracker *progress.Tracker) map[int32]Partition
• func StopConsumer(consumer *kafka.Consumer, tracker *progress.Tracker)
• type Message
• type Partition
• type Result
  • func Consume(consumer *kafka.Consumer, partitions map[int32]Partition, topic string, ...) Result
  • func Tail(consumer *kafka.Consumer, keyQuery string, valueQuery string, limit int64, ...) Result

Functions

func CreateEarliestConsumer

func CreateEarliestConsumer(bootstrap string, topic string, group string, tracker *progress.Tracker) *kafka.Consumer

CreateEarliestConsumer Creates a new Kafka consumer with the earliest offset

func CreateLatestConsumer

func CreateLatestConsumer(bootstrap string, topic string, group string, tracker *progress.Tracker) *kafka.Consumer

CreateLatestConsumer Creates a new Kafka consumer with the latest offset

func GetPartitions

func GetPartitions(consumer *kafka.Consumer, topic string, tracker *progress.Tracker) map[int32]Partition

GetPartitions retrieves information regarding all partitions for a provided topic

func StopConsumer

func StopConsumer(consumer *kafka.Consumer, tracker *progress.Tracker)

StopConsumer will stop and disconnect a consumer from Kafka

Types

type Message

type Message struct {
	Key       string
	Value     string
	Timestamp time.Time
	Partition int32
	Offset    string
}

Message contains information and data from a Kafka message/event

type Partition

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

Partition information about a topic partition

type Result

type Result struct {
	MatchedMessages int64
	ReadMessages    int64
	Duration        time.Duration
	Messages        list.List
}

Result from a Kafka operation, such as grep or tail

func Consume

func Consume(consumer *kafka.Consumer, partitions map[int32]Partition, topic string, keyQuery string, valueQuery string,
	limit int64, seekTimestamp string, latest bool, tracker *progress.Tracker) Result

Consume messages from a Kafka consumer

func Tail

func Tail(consumer *kafka.Consumer, keyQuery string, valueQuery string, limit int64, tracker *progress.Tracker) Result

Tail messages from a Kafka consumer