kafka

Documentation

Overview

#### references: - https://medium.com/event-driven-utopia/understanding-kafka-topic-partitions-ae40f80552e8 - https://silverback-messaging.net/concepts/broker/kafka/kafka-partitioning.html?tabs=destination-partition-fluent%2Cenricher-fluent%2Cconcurrency-fluent%2Cassignment-fluent - https://www.youtube.com/watch?v=JalUUBKdcA0

Index

• Constants
• type Config
• type Handle
• type Handler
  • func HandlerFromConfig(vp *viper.Viper, field string, ctx context.Context) (handler *Handler, err error)
  • func NewHandler(ctx context.Context, group sarama.ConsumerGroup, topics []string) (handler *Handler)
  • func (handler *Handler) Cleanup(sess sarama.ConsumerGroupSession) (err error)
  • func (handler *Handler) Close() error
  • func (handler *Handler) Consume() (err error)
  • func (handler *Handler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (handler *Handler) Setup(sess sarama.ConsumerGroupSession) (err error)
  • func (handler *Handler) WithHandle(handle Handle) *Handler
  • func (handler *Handler) WithLogger(logger *zap.Logger) *Handler

Constants

const (
	RFC3339ms = "2006-01-02T15:04:05.000Z07:00"
)

Types

type Config

type Config struct {
	Addrs   []string `mapstructure:"addrs"`
	Version string   `mapstructure:"version"`  // 3.4.0
	GroupId string   `mapstructure:"group_id"` // default
	Topic   string   `mapstructure:"topic"`
}

type Handle

type Handle func(msg *sarama.ConsumerMessage)

type Handler

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

func HandlerFromConfig

func HandlerFromConfig(vp *viper.Viper, field string, ctx context.Context) (
	handler *Handler, err error)

func NewHandler

func NewHandler(ctx context.Context, group sarama.ConsumerGroup, topics []string) (
	handler *Handler)

func (*Handler) Cleanup

func (handler *Handler) Cleanup(sess sarama.ConsumerGroupSession) (err error)

func (*Handler) Close

func (handler *Handler) Close() error

func (*Handler) Consume

func (handler *Handler) Consume() (err error)

func (*Handler) ConsumeClaim

func (handler *Handler) ConsumeClaim(sess sarama.ConsumerGroupSession,
	claim sarama.ConsumerGroupClaim) error

func (*Handler) Setup

func (handler *Handler) Setup(sess sarama.ConsumerGroupSession) (err error)

func (*Handler) WithHandle

func (handler *Handler) WithHandle(handle Handle) *Handler

func (*Handler) WithLogger

func (handler *Handler) WithLogger(logger *zap.Logger) *Handler