kafka

Documentation

Overview

Package kafka is a Metricbeat module that contains MetricSets.

Kafka is organised as following

- Topic - Partition - Producer - Consumer - Consumer Groups - Broker

Notes - Topics has a list of partitions - Each partition has an offset - Topic can be across brokers - Each broker has a list of partitions

Index

• type Broker
  • func NewBroker(host string, settings BrokerSettings) *Broker
  • func (b *Broker) Addr() string
  • func (b *Broker) AdvertisedAddr() string
  • func (b *Broker) Close() error
  • func (b *Broker) Connect() error
  • func (b *Broker) DescribeGroups(queryGroups []string) (map[string]GroupDescription, error)
  • func (b *Broker) FetchGroupOffsets(group string, partitions map[string][]int32) (*sarama.OffsetFetchResponse, error)
  • func (b *Broker) GetMetadata(topics ...string) (*sarama.MetadataResponse, error)
  • func (b *Broker) GetTopicsMetadata(topics ...string) ([]*sarama.TopicMetadata, error)
  • func (b *Broker) ID() int32
  • func (b *Broker) ListGroups() ([]string, error)
  • func (b *Broker) PartitionOffset(replicaID int32, topic string, partition int32, time int64) (int64, error)
• type BrokerSettings
• type GroupDescription
• type MemberDescription
• type Version
  • func (v *Version) Unpack(s string) error
  • func (v *Version) Validate() error

Types

type Broker

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

Broker provides functionality for communicating with a single kafka broker

func NewBroker

func NewBroker(host string, settings BrokerSettings) *Broker

NewBroker creates a new unconnected kafka Broker connection instance.

func (*Broker) Addr

func (b *Broker) Addr() string

Addr returns the configured broker endpoint.

func (*Broker) AdvertisedAddr

func (b *Broker) AdvertisedAddr() string

AdvertisedAddr returns the advertised broker address in case of matching broker has been found.

func (*Broker) Close

func (b *Broker) Close() error

Close the broker connection

func (*Broker) Connect

func (b *Broker) Connect() error

Connect connects the broker to the configured host

func (*Broker) DescribeGroups

func (b *Broker) DescribeGroups(
	queryGroups []string,
) (map[string]GroupDescription, error)

DescribeGroups fetches group details from broker.

func (*Broker) FetchGroupOffsets

func (b *Broker) FetchGroupOffsets(group string, partitions map[string][]int32) (*sarama.OffsetFetchResponse, error)

FetchGroupOffsets fetches the consume offset of group. The partitions is a MAP mapping from topic name to partitionid array.

func (*Broker) GetMetadata

func (b *Broker) GetMetadata(topics ...string) (*sarama.MetadataResponse, error)

GetMetadata fetches most recent cluster metadata from the broker.

func (*Broker) GetTopicsMetadata

func (b *Broker) GetTopicsMetadata(topics ...string) ([]*sarama.TopicMetadata, error)

GetTopicsMetadata fetches most recent topics/partition metadata from the broker.

func (*Broker) ID

func (b *Broker) ID() int32

ID returns the broker or -1 if the broker id is unknown.

func (*Broker) ListGroups

func (b *Broker) ListGroups() ([]string, error)

ListGroups lists all groups managed by the broker. Other consumer groups might be managed by other brokers.

func (*Broker) PartitionOffset

func (b *Broker) PartitionOffset(
	replicaID int32,
	topic string,
	partition int32,
	time int64,
) (int64, error)

PartitionOffset fetches the available offset from a partition.

type BrokerSettings

type BrokerSettings struct {
	MatchID                  bool
	DialTimeout, ReadTimeout time.Duration
	ClientID                 string
	Retries                  int
	Backoff                  time.Duration
	TLS                      *tls.Config
	Username, Password       string
	Version                  Version
}

BrokerSettings defines common configurations used when connecting to a broker

type GroupDescription

type GroupDescription struct {
	Members map[string]MemberDescription
}

type MemberDescription

type MemberDescription struct {
	Err        error
	ClientID   string
	ClientHost string
	Topics     map[string][]int32
}

type Version

type Version struct {
	String string
}

func (*Version) Unpack

func (v *Version) Unpack(s string) error

func (*Version) Validate

func (v *Version) Validate() error