kafka

Documentation

Index

• func BuildConsumerGroupConfig(conf *Config, initial int64) (*sarama.Config, error)
• func BuildProducerConfig(conf Config) (*sarama.Config, error)
• func CheckKafka(ctx context.Context, conf *Config, topics []string) error
• func NewConsumerGroup(conf *sarama.Config, addr []string, groupID string) (sarama.ConsumerGroup, error)
• func NewProducer(conf *sarama.Config, addr []string) (sarama.SyncProducer, error)
• type Config
• type TLSConfig

Functions

func BuildConsumerGroupConfig

func BuildConsumerGroupConfig(conf *Config, initial int64) (*sarama.Config, error)

func BuildProducerConfig

func BuildProducerConfig(conf Config) (*sarama.Config, error)

func CheckKafka

func CheckKafka(ctx context.Context, conf *Config, topics []string) error

func NewConsumerGroup

func NewConsumerGroup(conf *sarama.Config, addr []string, groupID string) (sarama.ConsumerGroup, error)

func NewProducer

func NewProducer(conf *sarama.Config, addr []string) (sarama.SyncProducer, error)

Types

type Config

type Config struct {
	Username     string     `yaml:"username"`
	Password     string     `yaml:"password"`
	ProducerAck  string     `yaml:"producerAck"`
	CompressType string     `yaml:"compressType"`
	Addr         []string   `yaml:"addr"`
	TLS          *TLSConfig `yaml:"tls"`
}

type TLSConfig

type TLSConfig struct {
	CACrt              string `yaml:"caCrt"`
	ClientCrt          string `yaml:"clientCrt"`
	ClientKey          string `yaml:"clientKey"`
	ClientKeyPwd       string `yaml:"clientKeyPwd"`
	InsecureSkipVerify bool   `yaml:"insecureSkipVerify"`
}