kafka

package

v1.3.0 Latest Latest Go to latest Published: Jul 14, 2023 License: MIT Imports: 8 Imported by: 2

Details

This section is empty.

View Source

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

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(*kafka.Dialer)

func DialerWithCaCertTls(file string) DialerOption

func DialerWithUserAndPasswd(user, passwd string) DialerOption

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

partitionBalancer

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

type ReaderOption func(*kafka.ReaderConfig)

func ReaderWithCommitInterval(commitInterval time.Duration) ReaderOption

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

func ReaderWithDialer(dialer *kafka.Dialer) ReaderOption

func ReaderWithMaxBytes(maxBytes int) ReaderOption

func ReaderWithMinBytes(minBytes int) ReaderOption

func ReaderWithOffset(offset int64) ReaderOption

type WriterOption func(*kafka.WriterConfig)

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