kafkautil

Documentation

Index

• func IsTemporaryError(err error) bool
• func IsTransientNetworkError(err error) bool
• func NewConsumer(config ConsumerConfig, authConfig AuthConfig, dialer *kafka.Dialer) *kafka.Reader
• func NewDialer(authConfig AuthConfig) (*kafka.Dialer, error)
• func NewProducer(config ProducerConfig, authConfig AuthConfig, dialer *kafka.Dialer) *kafka.Writer
• type AuthConfig
• type ConsumerConfig
• type ProducerConfig

Functions

func IsTemporaryError

func IsTemporaryError(err error) bool

func IsTransientNetworkError

func IsTransientNetworkError(err error) bool

func NewConsumer

func NewConsumer(config ConsumerConfig, authConfig AuthConfig, dialer *kafka.Dialer) *kafka.Reader

func NewDialer

func NewDialer(authConfig AuthConfig) (*kafka.Dialer, error)

NewDialer configures a connection dialer using the supplied AuthConfig

func NewProducer

func NewProducer(config ProducerConfig, authConfig AuthConfig, dialer *kafka.Dialer) *kafka.Writer

Types

type AuthConfig

type AuthConfig struct {
	Brokers          []string          `required:"true"`
	Mechanism        string            `required:"true"`
	MechanismOptions map[string]string `envconfig:"MECHANISM_OPTIONS"`
	Tls              bool              `required:"true"`
}

AuthConfig allows one to configure auth with a plain SASL authnetication mechanism to the Kafka brokers.

type ConsumerConfig

type ConsumerConfig struct {
	GroupID string `envconfig:"group_id",required:"true"`
	Topic   string `required:"true"`
}

ConsumerConfig allows one to configure a Kafka consumer using environment variables.

type ProducerConfig

type ProducerConfig struct {
	Topic string `required:"true"`
}

ProducerConfig allows one to configure a Kafka producer using environment variables.