instasarama

Jun 28, 2022

Changes in this version

+ func ProducerMessageWithSpanFromContext(ctx context.Context, pm *sarama.ProducerMessage) *sarama.ProducerMessage

Jun 20, 2022

Changes in this version

+ func NewConsumerGroup(addrs []string, groupID string, config *sarama.Config, sensor *instana.Sensor) (sarama.ConsumerGroup, error)

+ func NewConsumerGroupFromClient(groupID string, client sarama.Client, sensor *instana.Sensor) (sarama.ConsumerGroup, error)

May 25, 2022

Changes in this version

+ const FieldLS

+ const FieldS

+ const FieldT

+ const KafkaHeaderEnvVarKey

+ const Version

Apr 20, 2020

Changes in this version

+ const FieldC

+ const FieldL

+ func NewAsyncProducer(addrs []string, conf *sarama.Config, sensor *instana.Sensor) (sarama.AsyncProducer, error)

+ func NewAsyncProducerFromClient(client sarama.Client, sensor *instana.Sensor) (sarama.AsyncProducer, error)

+ func NewConsumer(addrs []string, config *sarama.Config, sensor *instana.Sensor) (sarama.Consumer, error)

+ func NewConsumerFromClient(client sarama.Client, sensor *instana.Sensor) (sarama.Consumer, error)

+ func NewSyncProducer(addrs []string, config *sarama.Config, sensor *instana.Sensor) (sarama.SyncProducer, error)

+ func NewSyncProducerFromClient(client sarama.Client, sensor *instana.Sensor) (sarama.SyncProducer, error)

+ func PackTraceContextHeader(traceID, spanID string) []byte

+ func PackTraceLevelHeader(val string) []byte

+ func ProducerMessageWithSpan(pm *sarama.ProducerMessage, sp ot.Span) *sarama.ProducerMessage

+ func SpanContextFromConsumerMessage(cm *sarama.ConsumerMessage, sensor *instana.Sensor) (ot.SpanContext, bool)

+ func UnpackTraceContextHeader(val []byte) (string, string, error)

+ func UnpackTraceLevelHeader(val []byte) (string, error)

+ type AsyncProducer struct

+ func WrapAsyncProducer(p sarama.AsyncProducer, conf *sarama.Config, sensor *instana.Sensor) *AsyncProducer

+ func (p *AsyncProducer) Errors() <-chan *sarama.ProducerError

+ func (p *AsyncProducer) Input() chan<- *sarama.ProducerMessage

+ func (p *AsyncProducer) Successes() <-chan *sarama.ProducerMessage

+ type Consumer struct

+ func WrapConsumer(c sarama.Consumer, sensor *instana.Sensor) *Consumer

+ func (c *Consumer) ConsumePartition(topic string, partition int32, offset int64) (sarama.PartitionConsumer, error)

+ type ConsumerGroupHandler struct

+ func WrapConsumerGroupHandler(h sarama.ConsumerGroupHandler, sensor *instana.Sensor) *ConsumerGroupHandler

+ func (h *ConsumerGroupHandler) Cleanup(sess sarama.ConsumerGroupSession) error

+ func (h *ConsumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

+ func (h *ConsumerGroupHandler) Setup(sess sarama.ConsumerGroupSession) error

+ type ConsumerMessageCarrier struct

+ Message *sarama.ConsumerMessage

+ func (c ConsumerMessageCarrier) ForeachKey(handler func(key, val string) error) error

+ func (c ConsumerMessageCarrier) RemoveAll()

+ func (c ConsumerMessageCarrier) Set(key, val string)

+ type PartitionConsumer struct

+ func WrapPartitionConsumer(c sarama.PartitionConsumer, sensor *instana.Sensor) *PartitionConsumer

+ func (pc *PartitionConsumer) Messages() <-chan *sarama.ConsumerMessage

+ type ProducerMessageCarrier struct

+ Message *sarama.ProducerMessage

+ func (c ProducerMessageCarrier) ForeachKey(handler func(key, val string) error) error

+ func (c ProducerMessageCarrier) RemoveAll()

+ func (c ProducerMessageCarrier) Set(key, val string)

+ type SyncProducer struct

+ func WrapSyncProducer(sp sarama.SyncProducer, config *sarama.Config, sensor *instana.Sensor) *SyncProducer

+ func (p *SyncProducer) SendMessage(msg *sarama.ProducerMessage) (int32, int64, error)

+ func (p *SyncProducer) SendMessages(msgs []*sarama.ProducerMessage) error