Versions in this module Expand all Collapse all v1 v1.0.0 Feb 25, 2025 Changes in this version + func Version() string + type Consumer struct + func NewConsumer(conf *kafka.ConfigMap, opts ...Option) (*Consumer, error) + func WrapConsumer(c *kafka.Consumer, opts ...Option) (*Consumer, error) + func (c *Consumer) Close() error + func (c *Consumer) Poll(timeoutMs int) (event kafka.Event) + func (c *Consumer) ReadMessage(timeout time.Duration) (*kafka.Message, error) + type MessageCarrier struct + func NewMessageCarrier(msg *kafka.Message) *MessageCarrier + func (c MessageCarrier) Get(key string) string + func (c MessageCarrier) Keys() []string + func (c MessageCarrier) Set(key string, value string) + type Option interface + func WithCustomAttributeInjector(fn func(msg *kafka.Message) []attribute.KeyValue) Option + func WithMeterProvider(meterProvider metric.MeterProvider) Option + func WithPropagators(propagators propagation.TextMapPropagator) Option + func WithTracerProvider(provider trace.TracerProvider) Option + type Producer struct + func NewProducer(conf *kafka.ConfigMap, opts ...Option) (*Producer, error) + func (p *Producer) Close() + func (p *Producer) Produce(msg *kafka.Message, deliveryChan chan kafka.Event) error
Go to main page