startup_kafka

Documentation

Index

• type Consumer
  • func NewConsumer(options ConsumerOptions) *Consumer
  • func (c *Consumer) Close() error
  • func (c *Consumer) Run()
• type ConsumerOptions
• type KafkaOptions
  • func (opts KafkaOptions) NewConsumer(config kafka.ConfigMap) *kafka.Consumer

Types

type Consumer

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

func NewConsumer

func NewConsumer(options ConsumerOptions) *Consumer

func (*Consumer) Close

func (c *Consumer) Close() error

func (*Consumer) Run

func (c *Consumer) Run()

type ConsumerOptions

type ConsumerOptions struct {
	KafkaAddresses     []string
	KafkaConsumerGroup string
	KafkaOffsetReset   string

	SchemaRegistry *avro.SchemaRegistry
}

type KafkaOptions

type KafkaOptions struct {
	KafkaAddresses        []string `` /* 155-byte string literal not displayed */
	KafkaConsumerGroup    string   `` /* 131-byte string literal not displayed */
	KafkaOffsetReset      string   `long:"kafka-offset-reset" default:"smallest" description:"Offset reset for kafka topic" choice:"smallest" choice:"largest"`
	KafkaSecurityProtocol string   `long:"kafka-security-protocol" default:"ssl" description:"Security protocol" choice:"ssl" choice:"plaintext"`
}

KafkaOptions provides simple flags for to create a kafka consumer

func (KafkaOptions) NewConsumer

func (opts KafkaOptions) NewConsumer(config kafka.ConfigMap) *kafka.Consumer