client

Documentation

Index

• func NewReaderClient(kafkaCfg kafka.Config, metrics *kprom.Metrics, logger log.Logger, ...) (*kgo.Client, error)
• func NewReaderClientMetrics(component string, reg prometheus.Registerer) *kprom.Metrics
• func NewWriterClient(kafkaCfg kafka.Config, maxInflightProduceRequests int, logger log.Logger, ...) (*kgo.Client, error)
• type Producer
  • func NewProducer(client *kgo.Client, maxBufferedBytes int64, reg prometheus.Registerer) *Producer
  • func (c *Producer) Close()
  • func (c *Producer) ProduceSync(ctx context.Context, records []*kgo.Record) kgo.ProduceResults

Functions

func NewReaderClient

func NewReaderClient(kafkaCfg kafka.Config, metrics *kprom.Metrics, logger log.Logger, opts ...kgo.Opt) (*kgo.Client, error)

NewReaderClient returns the kgo.Client that should be used by the Reader.

func NewReaderClientMetrics

func NewReaderClientMetrics(component string, reg prometheus.Registerer) *kprom.Metrics

func NewWriterClient

func NewWriterClient(kafkaCfg kafka.Config, maxInflightProduceRequests int, logger log.Logger, reg prometheus.Registerer) (*kgo.Client, error)

NewWriterClient returns the kgo.Client that should be used by the Writer.

The input prometheus.Registerer must be wrapped with a prefix (the names of metrics registered don't have a prefix).

Types

type Producer

type Producer struct {
	*kgo.Client
	// contains filtered or unexported fields
}

Producer is a kgo.Client wrapper exposing some higher level features and metrics useful for producers.

func NewProducer

func NewProducer(client *kgo.Client, maxBufferedBytes int64, reg prometheus.Registerer) *Producer

NewProducer returns a new KafkaProducer.

The input prometheus.Registerer must be wrapped with a prefix (the names of metrics registered don't have a prefix).

func (*Producer) Close

func (c *Producer) Close()

func (*Producer) ProduceSync

func (c *Producer) ProduceSync(ctx context.Context, records []*kgo.Record) kgo.ProduceResults

ProduceSync produces records to Kafka and returns once all records have been successfully committed, or an error occurred.

This function honors the configure max buffered bytes and refuse to produce a record, returnin kgo.ErrMaxBuffered, if the configured limit is reached.