kafka

Documentation

Index

• func GetBroker(zk string) (brokers []string, err error)
• func GetKafkaController(zkUrl string) (controller string, err error)
• func GetProducer(broker []string, topic string) (writer *kafka.Writer, err error)
• func InitTopic(zkUrl string, topics ...string) (err error)
• func NewConsumer(ctx context.Context, zkUrl string, groupId string, topic string, ...) error

Functions

func GetBroker

func GetBroker(zk string) (brokers []string, err error)

func GetKafkaController

func GetKafkaController(zkUrl string) (controller string, err error)

func GetProducer

func GetProducer(broker []string, topic string) (writer *kafka.Writer, err error)

func InitTopic

func InitTopic(zkUrl string, topics ...string) (err error)

func NewConsumer

func NewConsumer(ctx context.Context, zkUrl string, groupId string, topic string, listener func(delivery []byte) error) error