Version: v1.42.1 Latest Latest

This package is not in the latest version of its module.

Go to latest
Published: Sep 26, 2022 License: Apache-2.0, BSD-3-Clause, Apache-2.0 Imports: 9 Imported by: 0





This section is empty.


This section is empty.


func ExtractSpanContext

func ExtractSpanContext(msg kafka.Message) (ddtrace.SpanContext, error)

ExtractSpanContext retrieves the SpanContext from a kafka.Message


type Option

type Option func(cfg *config)

An Option customizes the config.

func WithAnalytics

func WithAnalytics(on bool) Option

WithAnalytics enables Trace Analytics for all started spans.

func WithAnalyticsRate

func WithAnalyticsRate(rate float64) Option

WithAnalyticsRate sets the sampling rate for Trace Analytics events correlated to started spans.

func WithServiceName

func WithServiceName(serviceName string) Option

WithServiceName sets the config service name to serviceName.

type Reader

type Reader struct {
	// contains filtered or unexported fields

A Reader wraps a kafka.Reader.

package main

import (

	kafkatrace "gopkg.in/DataDog/dd-trace-go.v1/contrib/segmentio/kafka.go.v0"

	kafka "github.com/segmentio/kafka-go"

func main() {
	r := kafkatrace.NewReader(kafka.ReaderConfig{
		Brokers:        []string{"localhost:9092"},
		Topic:          "some-topic",
		GroupID:        "group-id",
		SessionTimeout: 30 * time.Second,
	msg, err := r.ReadMessage(context.Background())
	if err != nil {
		log.Fatal("Failed to read message", err)

	// create a child span using span id and trace id in message header
	spanContext, err := kafkatrace.ExtractSpanContext(msg)
	if err != nil {
		log.Fatal("Failed to extract span context from carrier", err)
	operationName := "child-span"
	s := tracer.StartSpan(operationName, tracer.ChildOf(spanContext))
	defer s.Finish()

func NewReader

func NewReader(conf kafka.ReaderConfig, opts ...Option) *Reader

NewReader calls kafka.NewReader and wraps the resulting Consumer.

func WrapReader

func WrapReader(c *kafka.Reader, opts ...Option) *Reader

WrapReader wraps a kafka.Reader so that any consumed events are traced.

func (*Reader) Close

func (r *Reader) Close() error

Close calls the underlying Reader.Close and if polling is enabled, finishes any remaining span.

func (*Reader) FetchMessage added in v1.40.0

func (r *Reader) FetchMessage(ctx context.Context) (kafka.Message, error)

FetchMessage reads and returns the next message from the reader. Message will be traced.

func (*Reader) ReadMessage

func (r *Reader) ReadMessage(ctx context.Context) (kafka.Message, error)

ReadMessage polls the consumer for a message. Message will be traced.

type Writer

type Writer struct {
	// contains filtered or unexported fields

Writer wraps a kafka.Writer with tracing config data

package main

import (

	kafkatrace "gopkg.in/DataDog/dd-trace-go.v1/contrib/segmentio/kafka.go.v0"

	kafka "github.com/segmentio/kafka-go"

func main() {
	w := kafkatrace.NewWriter(kafka.WriterConfig{
		Brokers: []string{"localhost:9092"},
		Topic:   "some-topic",

	// use slice as it passes the value by reference if you want message headers updated in kafkatrace
	msgs := []kafka.Message{
			Key:   []byte("key1"),
			Value: []byte("value1"),
	if err := w.WriteMessages(context.Background(), msgs...); err != nil {
		log.Fatal("Failed to write message", err)

func NewWriter

func NewWriter(conf kafka.WriterConfig, opts ...Option) *Writer

NewWriter calls kafka.NewWriter and wraps the resulting Producer.

func WrapWriter

func WrapWriter(w *kafka.Writer, opts ...Option) *Writer

WrapWriter wraps a kafka.Writer so requests are traced.

func (*Writer) WriteMessages

func (w *Writer) WriteMessages(ctx context.Context, msgs ...kafka.Message) error

WriteMessages calls kafka.go.v0.Writer.WriteMessages and traces the requests.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL