kafkaclient

Documentation

Overview

Package kafkaclient provides implementations of Kafka producers using different client libraries.

Index

• type FranzProducerMetrics
  • func NewFranzProducerMetrics(tb *metadata.TelemetryBuilder) *FranzProducerMetrics
  • func (fpm *FranzProducerMetrics) OnBrokerConnect(meta kgo.BrokerMetadata, _ time.Duration, _ net.Conn, err error)
  • func (fpm *FranzProducerMetrics) OnBrokerDisconnect(meta kgo.BrokerMetadata, _ net.Conn)
  • func (fpm *FranzProducerMetrics) OnBrokerE2E(meta kgo.BrokerMetadata, key int16, e2e kgo.BrokerE2E)
  • func (fpm *FranzProducerMetrics) OnBrokerThrottle(meta kgo.BrokerMetadata, throttleInterval time.Duration, _ bool)
  • func (fpm *FranzProducerMetrics) OnProduceBatchWritten(meta kgo.BrokerMetadata, topic string, partition int32, ...)
  • func (fpm *FranzProducerMetrics) OnProduceRecordUnbuffered(r *kgo.Record, err error)
• type FranzSyncProducer
  • func NewFranzSyncProducer(client *kgo.Client, metadataKeys []string, recordHeaders []RecordHeader, ...) *FranzSyncProducer
  • func (p *FranzSyncProducer) Close(ctx context.Context) error
  • func (p *FranzSyncProducer) ExportData(ctx context.Context, records []*kgo.Record) error
• type MessageTooLargeError
  • func (e *MessageTooLargeError) Error() string
  • func (e *MessageTooLargeError) Unwrap() error
• type RecordHeader
• type StatusReporter
  • func NewStatusReporter(host component.Host) *StatusReporter
  • func (s *StatusReporter) OnBrokerConnect(_ kgo.BrokerMetadata, _ time.Duration, _ net.Conn, err error)
  • func (s *StatusReporter) OnBrokerDisconnect(_ kgo.BrokerMetadata, _ net.Conn)

Types

type FranzProducerMetrics

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

FranzProducerMetrics implements the relevant franz-go hook interfaces to record the metrics defined in the metadata telemetry.

func NewFranzProducerMetrics

func NewFranzProducerMetrics(tb *metadata.TelemetryBuilder) *FranzProducerMetrics

NewFranzProducerMetrics creates an instance of FranzProducerMetrics from metadata TelemetryBuilder.

func (*FranzProducerMetrics) OnBrokerConnect

func (fpm *FranzProducerMetrics) OnBrokerConnect(meta kgo.BrokerMetadata, _ time.Duration, _ net.Conn, err error)

func (*FranzProducerMetrics) OnBrokerDisconnect

func (fpm *FranzProducerMetrics) OnBrokerDisconnect(meta kgo.BrokerMetadata, _ net.Conn)

func (*FranzProducerMetrics) OnBrokerE2E

func (fpm *FranzProducerMetrics) OnBrokerE2E(meta kgo.BrokerMetadata, key int16, e2e kgo.BrokerE2E)

func (*FranzProducerMetrics) OnBrokerThrottle

func (fpm *FranzProducerMetrics) OnBrokerThrottle(meta kgo.BrokerMetadata, throttleInterval time.Duration, _ bool)

func (*FranzProducerMetrics) OnProduceBatchWritten

func (fpm *FranzProducerMetrics) OnProduceBatchWritten(meta kgo.BrokerMetadata, topic string, partition int32, m kgo.ProduceBatchMetrics)

OnProduceBatchWritten is called when a batch has been produced. https://pkg.go.dev/github.com/twmb/franz-go/pkg/kgo#HookProduceBatchWritten

func (*FranzProducerMetrics) OnProduceRecordUnbuffered

func (fpm *FranzProducerMetrics) OnProduceRecordUnbuffered(r *kgo.Record, err error)

OnProduceRecordUnbuffered records the number of produced messages that were not produced due to errors. The successfully produced records is recorded by `OnProduceBatchWritten`. https://pkg.go.dev/github.com/twmb/franz-go/pkg/kgo#HookProduceRecordUnbuffered

type FranzSyncProducer

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

FranzSyncProducer is a wrapper around the franz-go client that implements the Producer interface. Allowing us to use the franz-go client while maintaining compatibility with the existing Kafka exporter code.

func NewFranzSyncProducer

func NewFranzSyncProducer(client *kgo.Client,
	metadataKeys []string,
	recordHeaders []RecordHeader,
	maxMessageBytes int,
	clientCancel context.CancelFunc,
) *FranzSyncProducer

NewFranzSyncProducer Franz-go producer from a kgo.Client and a Messenger. clientCancel must cancel the context passed to kgo.WithContext when the client was created; it is called by Close to unblock any in-flight ProduceSync calls.

func (*FranzSyncProducer) Close

func (p *FranzSyncProducer) Close(ctx context.Context) error

Close shuts down the producer, unblocking any in-flight ExportData call.

func (*FranzSyncProducer) ExportData

func (p *FranzSyncProducer) ExportData(ctx context.Context, records []*kgo.Record) error

ExportData sends a batch of records to Kafka. It attaches configured record headers and per-call metadata-derived headers to each record before producing.

type MessageTooLargeError

type MessageTooLargeError struct {
	// RecordBytes is the user-visible size of the record (key + value + headers).
	RecordBytes int
	// MaxMessageBytes is the configured producer max message size.
	MaxMessageBytes int
	Err             error
}

MessageTooLargeError wraps a MessageTooLarge Kafka error with the actual record size that caused the rejection. The size is computed the same way as franz-go's Record.userSize: len(Key) + len(Value) + Σ(len(header.Key) + len(header.Value)).

func (*MessageTooLargeError) Error

func (e *MessageTooLargeError) Error() string

func (*MessageTooLargeError) Unwrap

func (e *MessageTooLargeError) Unwrap() error

type RecordHeader

type RecordHeader struct {
	Name  string              `mapstructure:"name"`
	Value configopaque.String `mapstructure:"value"`
	// contains filtered or unexported fields
}

RecordHeader includes key-value pairs to be added as headers to Kafka records.

type StatusReporter

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

func NewStatusReporter

func NewStatusReporter(host component.Host) *StatusReporter

func (*StatusReporter) OnBrokerConnect

func (s *StatusReporter) OnBrokerConnect(_ kgo.BrokerMetadata, _ time.Duration, _ net.Conn, err error)

func (*StatusReporter) OnBrokerDisconnect

func (s *StatusReporter) OnBrokerDisconnect(_ kgo.BrokerMetadata, _ net.Conn)