kafka

Documentation

Index

• type Client
  • func NewClient(conf []kgo.Opt, opts ...Option) (*Client, error)
  • func WrapClient(c KafkaClient, opts ...Option) *Client
  • func (c *Client) Close()
  • func (c *Client) Produce(ctx context.Context, msg *kgo.Record, fn func(record *kgo.Record, err error))
  • func (c *Client) ProduceSync(ctx context.Context, msgs ...*kgo.Record) kgo.ProduceResults
  • func (c *Client) WrapFetchPartition(ctx context.Context, fp kgo.FetchPartition) *FetchPartition
  • func (c *Client) WrapFetchesRecordIter(ctx context.Context, i *kgo.FetchesRecordIter) *FetchesRecordIter
• type FetchPartition
  • func (fp *FetchPartition) ConsumeRecord(rec *kgo.Record)
  • func (fp *FetchPartition) EachRecord(fn func(rec *kgo.Record))
• type FetchesRecordIter
  • func (i *FetchesRecordIter) Done() bool
  • func (i *FetchesRecordIter) Next() *kgo.Record
• type KafkaClient
• type MessageCarrier
  • func NewMessageCarrier(msg *kgo.Record) MessageCarrier
  • func (c MessageCarrier) ForeachKey(handler func(key, val string) error) error
  • func (c MessageCarrier) Set(key, val string)
• type Option
  • func WithAnalytics(on bool) Option
  • func WithAnalyticsRate(rate float64) Option
  • func WithServiceName(serviceName string) Option

Types

type Client

type Client struct {
	KafkaClient
	// contains filtered or unexported fields
}

func NewClient

func NewClient(conf []kgo.Opt, opts ...Option) (*Client, error)

NewClient calls kgo.NewClient and wraps the resulting Client.

func WrapClient

func WrapClient(c KafkaClient, opts ...Option) *Client

WrapClient wraps kgo.Client so that produced and consumed messages are traced.

func (*Client) Close

func (c *Client) Close()

Close calls the underlying *kgo.Close and finishes the remaining span.

func (*Client) Produce

func (c *Client) Produce(ctx context.Context, msg *kgo.Record, fn func(record *kgo.Record, err error))

Produce calls the underlying *kgo.Client.Produce, the request will be traced. This function is used for producing message asynchronously.

func (*Client) ProduceSync

func (c *Client) ProduceSync(ctx context.Context, msgs ...*kgo.Record) kgo.ProduceResults

ProduceSync calls the underlying *kgo.Client.ProduceSync and traces all results.

func (*Client) WrapFetchPartition

func (c *Client) WrapFetchPartition(ctx context.Context, fp kgo.FetchPartition) *FetchPartition

WrapFetchPartition wraps the kgo.FetchPartition and links it to the Client.

func (*Client) WrapFetchesRecordIter

func (c *Client) WrapFetchesRecordIter(ctx context.Context, i *kgo.FetchesRecordIter) *FetchesRecordIter

WrapFetchesRecordIter wraps the kgo.FetchesRecordIter and links it to the Client.

type FetchPartition

type FetchPartition struct {
	kgo.FetchPartition
	// contains filtered or unexported fields
}

func (*FetchPartition) ConsumeRecord

func (fp *FetchPartition) ConsumeRecord(rec *kgo.Record)

ConsumeRecord finishes the span for a particular record.

func (*FetchPartition) EachRecord

func (fp *FetchPartition) EachRecord(fn func(rec *kgo.Record))

EachRecord calls underlying kgo.FetchPartition.EachRecord and traces the message.

type FetchesRecordIter

type FetchesRecordIter struct {
	*kgo.FetchesRecordIter
	// contains filtered or unexported fields
}

func (*FetchesRecordIter) Done

func (i *FetchesRecordIter) Done() bool

Done calls underlying kgo.FetchesRecordIter.Done and finishes any remaining span.

func (*FetchesRecordIter) Next

func (i *FetchesRecordIter) Next() *kgo.Record

Next calls underlying kgo.FetchesRecordIter.Next and traces the message.

type KafkaClient

type KafkaClient interface {
	Produce(ctx context.Context, r *kgo.Record, promise func(*kgo.Record, error))
	ProduceSync(ctx context.Context, rs ...*kgo.Record) kgo.ProduceResults
	PollRecords(ctx context.Context, num int) kgo.Fetches
	Flush(ctx context.Context) error
	Close()
}

type MessageCarrier

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

func NewMessageCarrier

func NewMessageCarrier(msg *kgo.Record) MessageCarrier

NewMessageCarrier creates a new MessageCarrier.

func (MessageCarrier) ForeachKey

func (c MessageCarrier) ForeachKey(handler func(key, val string) error) error

ForeachKey iterates over every header.

func (MessageCarrier) Set

func (c MessageCarrier) Set(key, val string)

Set sets a header.

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.