producer

Documentation

Index

• type Batch
  • func (b *Batch) AddMessages(ctx *models.ListenerContext, messages []kafka.Message, eventTime time.Time)
  • func (b *Batch) Close()
  • func (b *Batch) FlushMessages()
  • func (b *Batch) PrepareEndRebalancing()
  • func (b *Batch) PrepareStartRebalancing()
  • func (b *Batch) StartBatchTicker()
• type Metric
• type Producer
  • func NewProducer(kafkaClient gKafka.Client, config *config.Connector, ...) (Producer, error)
  • func (p *Producer) Close() error
  • func (p *Producer) GetMetric() *Metric
  • func (p *Producer) Produce(ctx *models.ListenerContext, eventTime time.Time, messages []kafka.Message)
  • func (p *Producer) StartBatch()

Types

type Batch

type Batch struct {
	Writer *kafka.Writer
	// contains filtered or unexported fields
}

func (*Batch) AddMessages

func (b *Batch) AddMessages(ctx *models.ListenerContext, messages []kafka.Message, eventTime time.Time)

func (*Batch) Close

func (b *Batch) Close()

func (*Batch) FlushMessages

func (b *Batch) FlushMessages()

func (*Batch) PrepareEndRebalancing

func (b *Batch) PrepareEndRebalancing()

func (*Batch) PrepareStartRebalancing

func (b *Batch) PrepareStartRebalancing()

func (*Batch) StartBatchTicker

func (b *Batch) StartBatchTicker()

type Metric

type Metric struct {
	KafkaConnectorLatency int64
	BatchProduceLatency   int64
}

type Producer

type Producer struct {
	ProducerBatch *Batch
}

func NewProducer

func NewProducer(kafkaClient gKafka.Client,
	config *config.Connector,
	dcpCheckpointCommit func(),
) (Producer, error)

func (*Producer) Close

func (p *Producer) Close() error

func (*Producer) GetMetric

func (p *Producer) GetMetric() *Metric

func (*Producer) Produce

func (p *Producer) Produce(
	ctx *models.ListenerContext,
	eventTime time.Time,
	messages []kafka.Message,
)

func (*Producer) StartBatch

func (p *Producer) StartBatch()