kafka

Aug 28, 2020

Changes in this version

+ func ConvertHeaderToMap(header sarama.RecordHeader) map[string]string

+ type AsyncProducer struct

+ BrokerList []string

+ Client sarama.Client

+ Config *sarama.Config

+ KafkaVersion sarama.KafkaVersion

+ Producer sarama.AsyncProducer

+ func NewAsyncProducer(kafkaVersion string, brokerList []string) (p *AsyncProducer, err error)

+ func (p *AsyncProducer) BuildProducerMessage(topicName string, key string, message string, headers []sarama.RecordHeader) *sarama.ProducerMessage

+ func (p *AsyncProducer) BuildProducerMessageHeader(key string, value string) sarama.RecordHeader

+ func (p *AsyncProducer) Close() error

+ func (p *AsyncProducer) Produce(topicName string, message interface{}) (err error)

+ type ConsumerGroup struct

+ BrokerList []string

+ Client sarama.Client

+ Config *sarama.Config

+ Group sarama.ConsumerGroup

+ GroupName string

+ KafkaVersion sarama.KafkaVersion

+ func NewConsumerGroup(kafkaVersion string, brokerList []string, groupName string, initOffset int64) (cg *ConsumerGroup, err error)

+ func (cg *ConsumerGroup) Close() error

+ func (cg *ConsumerGroup) Consume(ctx context.Context, topicName string, handler sarama.ConsumerGroupHandler) (err error)

+ type DefaultConsumerGroupHandler struct

+ func (DefaultConsumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error

+ func (h DefaultConsumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

+ func (h DefaultConsumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error