kafka

Dec 5, 2024

Changes in this version

type ProducerConfig

+ AllowAutoTopicCreation bool

Sep 23, 2024

Changes in this version

type CommonConfig

+ TopicLogFieldFunc TopicLogFieldFunc

+ type TopicLogFieldFunc func(topic string) zap.Field

Aug 19, 2024

Changes in this version

type ConsumerConfig

+ BrokerMaxReadBytes int32

Aug 2, 2024

Changes in this version

type CommonConfig

+ MetadataMaxAge time.Duration

type ConsumerConfig

+ PreferLagFn kgo.PreferLagFn

type ProducerConfig

+ RecordPartitioner kgo.Partitioner

May 7, 2024

Changes in this version

type ProducerConfig

+ ProduceCallback func(*kgo.Record, error)

Apr 26, 2024

Changes in this version

type ConsumerConfig

+ FetchMinBytes int32

Apr 15, 2024

Changes in this version

+ type BatchWriteListener func(topic string, bytesWritten int)

+ func (l BatchWriteListener) OnProduceBatchWritten(_ kgo.BrokerMetadata, topic string, _ int32, m kgo.ProduceBatchMetrics)

type ProducerConfig

+ BatchListener BatchWriteListener

Mar 26, 2024

Changes in this version

type CommonConfig

+ TopicAttributeFunc TopicAttributeFunc

+ type TopicAttributeFunc func(topic string) attribute.KeyValue

Mar 6, 2024

Changes in this version

+ var ErrCommitFailed = errors.New("kafka: failed to commit offsets")

+ type CommonConfig struct

+ Brokers []string

+ ClientID string

+ ConfigFile string

+ Dialer func(ctx context.Context, network, address string) (net.Conn, error)

+ DisableTelemetry bool

+ Logger *zap.Logger

+ MeterProvider metric.MeterProvider

+ Namespace string

+ SASL SASLMechanism

+ TLS *tls.Config

+ TracerProvider trace.TracerProvider

+ Version string

+ type CompressionCodec = kgo.CompressionCodec

+ func GzipCompression() CompressionCodec

+ func Lz4Compression() CompressionCodec

+ func NoCompression() CompressionCodec

+ func SnappyCompression() CompressionCodec

+ func ZstdCompression() CompressionCodec

+ type Consumer struct

+ func NewConsumer(cfg ConsumerConfig) (*Consumer, error)

+ func (c *Consumer) Close() error

+ func (c *Consumer) Healthy(ctx context.Context) error

+ func (c *Consumer) Run(ctx context.Context) error

+ type ConsumerConfig struct

+ ConsumeRegex bool

+ Delivery apmqueue.DeliveryType

+ GroupID string

+ MaxConcurrentFetches int

+ MaxPollBytes int32

+ MaxPollPartitionBytes int32

+ MaxPollRecords int

+ MaxPollWait time.Duration

+ Processor apmqueue.Processor

+ ShutdownGracePeriod time.Duration

+ Topics []apmqueue.Topic

+ type Manager struct

+ func NewManager(cfg ManagerConfig) (*Manager, error)

+ func (m *Manager) Close() error

+ func (m *Manager) DeleteTopics(ctx context.Context, topics ...apmqueue.Topic) error

+ func (m *Manager) Healthy(ctx context.Context) error

+ func (m *Manager) MonitorConsumerLag(topicConsumers []apmqueue.TopicConsumer) (metric.Registration, error)

+ func (m *Manager) NewTopicCreator(cfg TopicCreatorConfig) (*TopicCreator, error)

+ type ManagerConfig struct

+ type Producer struct

+ func NewProducer(cfg ProducerConfig) (*Producer, error)

+ func (p *Producer) Close() error

+ func (p *Producer) Healthy(ctx context.Context) error

+ func (p *Producer) Produce(ctx context.Context, rs ...apmqueue.Record) error

+ type ProducerConfig struct

+ CompressionCodec []CompressionCodec

+ ManualFlushing bool

+ MaxBufferedRecords int

+ ProducerBatchMaxBytes int32

+ Sync bool

+ type SASLMechanism = sasl.Mechanism

+ type TopicCreator struct

+ func (c *TopicCreator) CreateTopics(ctx context.Context, topics ...apmqueue.Topic) error

+ type TopicCreatorConfig struct

+ MeterProvider metric.MeterProvider

+ PartitionCount int

+ TopicConfigs map[string]string

+ func (cfg TopicCreatorConfig) Validate() error