kafka

Documentation

Index

• Variables
• func GetKafkaDialer(options ...DialerOption) *kafka.Dialer
• func NewGroupReader(brokers []string, topic, groupID string, options ...ReaderOption) (r *kafka.Reader, e error)
• func NewPartitionReader(brokers []string, topic string, partition int, options ...ReaderOption) (r *kafka.Reader, e error)
• func NewPartitionWriter(brokers []string, topic string, partition int, options ...WriterOption) (w *kafka.Writer, e error)
• func NewWriter(brokers []string, topic string, options ...WriterOption) (w *kafka.Writer, e error)
• func SetLog(l logger.Logger)
• type DialerOption
  • func DialerWithCaCertTls(file string) DialerOption
  • func DialerWithUserAndPasswd(user, passwd string) DialerOption
• type PartitionBalancer
  • func (p *PartitionBalancer) Balance(_ kafka.Message, _ ...int) (partition int)
• type ReaderOption
  • func ReaderWithCommitInterval(commitInterval time.Duration) ReaderOption
  • func ReaderWithDialer(dialer *kafka.Dialer) ReaderOption
  • func ReaderWithMaxBytes(maxBytes int) ReaderOption
  • func ReaderWithMinBytes(minBytes int) ReaderOption
  • func ReaderWithOffset(offset int64) ReaderOption
• type WriterOption
  • func WriterWithAsync() WriterOption
  • func WriterWithBatchBytes(batchBytes int) WriterOption
  • func WriterWithBatchSize(batchSize int) WriterOption
  • func WriterWithBatchTimeout(batchTimeout time.Duration) WriterOption
  • func WriterWithDialer(dialer *kafka.Dialer) WriterOption
  • func WriterWithLogger(debugLogger, errorLogger kafka.Logger) WriterOption

Variables

var (
	KLogger      = &debugLogger{log}
	KErrorLogger = &errorLogger{log}
)

Functions

func GetKafkaDialer

func GetKafkaDialer(options ...DialerOption) *kafka.Dialer

func NewGroupReader

func NewGroupReader(brokers []string, topic, groupID string, options ...ReaderOption) (r *kafka.Reader, e error)

func NewPartitionReader

func NewPartitionReader(brokers []string, topic string, partition int, options ...ReaderOption) (r *kafka.Reader, e error)

func NewPartitionWriter

func NewPartitionWriter(brokers []string, topic string, partition int, options ...WriterOption) (w *kafka.Writer, e error)

func NewWriter

func NewWriter(brokers []string, topic string, options ...WriterOption) (w *kafka.Writer, e error)

func SetLog

func SetLog(l logger.Logger)

Types

type DialerOption

type DialerOption func(*kafka.Dialer)

func DialerWithCaCertTls

func DialerWithCaCertTls(file string) DialerOption

func DialerWithUserAndPasswd

func DialerWithUserAndPasswd(user, passwd string) DialerOption

type PartitionBalancer

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

partitionBalancer

func (*PartitionBalancer) Balance

func (p *PartitionBalancer) Balance(_ kafka.Message, _ ...int) (partition int)

type ReaderOption

type ReaderOption func(*kafka.ReaderConfig)

func ReaderWithCommitInterval

func ReaderWithCommitInterval(commitInterval time.Duration) ReaderOption

ReaderWithCommitInterval be careful commitInterval may cause msg repeated consume when server error exit.

func ReaderWithDialer

func ReaderWithDialer(dialer *kafka.Dialer) ReaderOption

func ReaderWithMaxBytes

func ReaderWithMaxBytes(maxBytes int) ReaderOption

func ReaderWithMinBytes

func ReaderWithMinBytes(minBytes int) ReaderOption

func ReaderWithOffset

func ReaderWithOffset(offset int64) ReaderOption

type WriterOption

type WriterOption func(*kafka.WriterConfig)

func WriterWithAsync

func WriterWithAsync() WriterOption

func WriterWithBatchBytes

func WriterWithBatchBytes(batchBytes int) WriterOption

func WriterWithBatchSize

func WriterWithBatchSize(batchSize int) WriterOption

func WriterWithBatchTimeout

func WriterWithBatchTimeout(batchTimeout time.Duration) WriterOption

func WriterWithDialer

func WriterWithDialer(dialer *kafka.Dialer) WriterOption

func WriterWithLogger

func WriterWithLogger(debugLogger, errorLogger kafka.Logger) WriterOption