kafka

Documentation

Index

• Constants
• Variables
• func ConsumerHandle(group *ConsumerGroup) (*kafka.Consumer, error)
• type ConsumerConfig
• type ConsumerGroup
  • func (cg *ConsumerGroup) Consume(ctx context.Context, handler ziggurat.Handler) error
• type ErrorWorkerKilled
  • func (e ErrorWorkerKilled) Error() string
• type MockConsumer
  • func (m *MockConsumer) Close() error
  • func (m *MockConsumer) Commit() ([]kafka.TopicPartition, error)
  • func (m *MockConsumer) Logs() chan kafka.LogEvent
  • func (m *MockConsumer) Poll(i int) kafka.Event
  • func (m *MockConsumer) StoreOffsets(partitions []kafka.TopicPartition) ([]kafka.TopicPartition, error)

Constants

const (
	EventType = "kafka"
)

Variables

var ErrCleanShutdown = errors.New("error: clean shutdown of kafka consumers")

Functions

func ConsumerHandle

func ConsumerHandle(group *ConsumerGroup) (*kafka.Consumer, error)

Types

type ConsumerConfig

type ConsumerConfig struct {
	BootstrapServers      string
	DebugLevel            string
	GroupID               string
	Topics                []string
	AutoCommitInterval    int
	ConsumerCount         int
	PollTimeout           int
	AutoOffsetReset       string
	PartitionAssignment   string
	MaxPollIntervalMS     int
	AllowAutoCreateTopics bool
}

type ConsumerGroup

type ConsumerGroup struct {
	Logger      ziggurat.StructuredLogger
	GroupConfig ConsumerConfig
	// contains filtered or unexported fields
}

func (*ConsumerGroup) Consume

func (cg *ConsumerGroup) Consume(ctx context.Context, handler ziggurat.Handler) error

type ErrorWorkerKilled

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

func (ErrorWorkerKilled) Error

func (e ErrorWorkerKilled) Error() string

type MockConsumer

type MockConsumer struct {
	mock.Mock
}

func (*MockConsumer) Close

func (m *MockConsumer) Close() error

func (*MockConsumer) Commit

func (m *MockConsumer) Commit() ([]kafka.TopicPartition, error)

func (*MockConsumer) Logs

func (m *MockConsumer) Logs() chan kafka.LogEvent

func (*MockConsumer) Poll

func (m *MockConsumer) Poll(i int) kafka.Event

func (*MockConsumer) StoreOffsets

func (m *MockConsumer) StoreOffsets(partitions []kafka.TopicPartition) ([]kafka.TopicPartition, error)