kafka

Documentation

Index

• Constants
• type Brokers
• type Config
• type DefaultKafkaClient
  • func NewDefaultKafkaClient() DefaultKafkaClient
  • func (kc DefaultKafkaClient) CreatePartitions(ctx context.Context, uri string, topic Topic, numberOfPartitions int64) error
  • func (kc DefaultKafkaClient) CreateTopic(uri string, topic Topic) error
  • func (kc DefaultKafkaClient) GetTopic(uri string, name string) (*Topic, error)
  • func (kc DefaultKafkaClient) UpdateTopicConfiguration(ctx context.Context, uri string, topic Topic) error
• type KafkaClient
• type Topic

Constants

const (
	DefaultKafkaClientTimeout = 4 * time.Minute
)

const (
	TCP = "tcp"
)

Types

type Brokers

type Brokers map[int64]int64

Holds BrokerIds as keys, and number of assignments for this Broker (currently the partitions that the Broker is a leader for)

type Config

type Config map[string]string

Holds name-value configuration options, passed to underlying kafka libs as-is

type DefaultKafkaClient

type DefaultKafkaClient struct{}

func NewDefaultKafkaClient

func NewDefaultKafkaClient() DefaultKafkaClient

func (DefaultKafkaClient) CreatePartitions

func (kc DefaultKafkaClient) CreatePartitions(ctx context.Context, uri string, topic Topic, numberOfPartitions int64) error

func (DefaultKafkaClient) CreateTopic

func (kc DefaultKafkaClient) CreateTopic(uri string, topic Topic) error

func (DefaultKafkaClient) GetTopic

func (kc DefaultKafkaClient) GetTopic(uri string, name string) (*Topic, error)

func (DefaultKafkaClient) UpdateTopicConfiguration

func (kc DefaultKafkaClient) UpdateTopicConfiguration(ctx context.Context, uri string, topic Topic) error

type KafkaClient

type KafkaClient interface {
	CreateTopic(uri string, topic Topic) error
	CreatePartitions(ctx context.Context, uri string, topic Topic, numberOfPartitions int64) error
	UpdateTopicConfiguration(ctx context.Context, uri string, topic Topic) error
	GetTopic(uri string, name string) (*Topic, error)
}

type Topic

type Topic struct {
	Name              string
	Partitions        int64
	ReplicationFactor int64
	Brokers           Brokers
	Config            Config
}