kafkacommon

README

kafkacommon

Wrapper for segmentio/kafka-go writer with retries and metrics

Documentation

Index

• type Broker
• type Client
  • func NewClient(config *Config, metrics promcommon.BrokerIncrementObserver) (*Client, error)
  • func (c *Client) Write(ctx context.Context, message kafka.Message) error
• type Config
  • func (c *Config) WithDefaults() (*Config, error)

Types

type Broker

type Broker interface {
	Write(ctx context.Context, messages []kafka.Message) error
}

type Client

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

func NewClient

func NewClient(config *Config, metrics promcommon.BrokerIncrementObserver) (*Client, error)

func (*Client) Write

func (c *Client) Write(ctx context.Context, message kafka.Message) error

type Config

type Config struct {
	Addresses               []string      `env:"ADDRESSES" yaml:"addresses"`
	WriteTopic              string        `env:"WRITE_TOPIC" yaml:"write_topic"`
	WriteBatchSize          int           `env:"WRITE_BATCH_SIZE" yaml:"write_batch_size"`
	WriteBatchTimeout       time.Duration `env:"WRITE_BATCH_TIMEOUT" yaml:"write_batch_timeout"`
	WriteMaxRetryCount      uint64        `env:"WRITE_MAX_RETRY_COUNT" yaml:"write_max_retry_count"`
	WriteRetryJitterPercent uint64        `env:"WRITE_RETRY_JITTER_PERCENT" yaml:"write_retry_jitter_percent"`
	WriteRetryTimeout       time.Duration `env:"WRITE_RETRY_TIMEOUT" yaml:"write_retry_timeout"`
}

func (*Config) WithDefaults

func (c *Config) WithDefaults() (*Config, error)