connection

Documentation

Overview

Package connection handles connecting to brokers via JMX and Kafka protocol

Index

• type Broker
  • func GetBrokerFromZookeeper(zkConn zookeeper.Connection, id, preferredListener string) (*Broker, error)
  • func GetBrokerListFromZookeeper(zkConn zookeeper.Connection, preferredListener string) ([]*Broker, error)
  • func NewBroker(brokerArgs *args.BrokerHost) (*Broker, error)
  • func (b *Broker) Entity(i *integration.Integration) (*integration.Entity, error)
• type Client
  • func NewSaramaClientFromBrokerList(brokers []*Broker) (Client, error)
• type SaramaBroker

Types

type Broker

type Broker struct {
	Config      *sarama.Config
	JMXPort     int
	JMXUser     string
	JMXPassword string
	Host        string
	ID          string
	SaramaBroker
}

Broker is a struct containing all the information to collect from both Kafka and from JMX

func GetBrokerFromZookeeper

func GetBrokerFromZookeeper(zkConn zookeeper.Connection, id, preferredListener string) (*Broker, error)

GetBrokerFromZookeeper gets a broker with given ID from zookeeper

func GetBrokerListFromZookeeper

func GetBrokerListFromZookeeper(zkConn zookeeper.Connection, preferredListener string) ([]*Broker, error)

GetBrokerListFromZookeeper gets a list of brokers from zookeeper

func NewBroker

func NewBroker(brokerArgs *args.BrokerHost) (*Broker, error)

NewBroker creates a new broker

func (*Broker) Entity

func (b *Broker) Entity(i *integration.Integration) (*integration.Entity, error)

Entity gets the entity object for the broker

type Client

type Client interface {
	Config() *sarama.Config
	Controller() (*sarama.Broker, error)
	Brokers() []*sarama.Broker
	Topics() ([]string, error)
	Partitions(topic string) ([]int32, error)
	WritablePartitions(topic string) ([]int32, error)
	Leader(topic string, partitionID int32) (*sarama.Broker, error)
	Replicas(topic string, partitionID int32) ([]int32, error)
	InSyncReplicas(topic string, partitionID int32) ([]int32, error)
	OfflineReplicas(topic string, partitionID int32) ([]int32, error)
	RefreshMetadata(topics ...string) error
	GetOffset(topic string, partitionID int32, time int64) (int64, error)
	Coordinator(consumerGroup string) (*sarama.Broker, error)
	RefreshCoordinator(consumerGroup string) error
	InitProducerID() (*sarama.InitProducerIDResponse, error)
	Close() error
	Closed() bool
}

Client is a wrapper around sarama.Client so that we can generate mocks See sarama.Client for documentation

func NewSaramaClientFromBrokerList

func NewSaramaClientFromBrokerList(brokers []*Broker) (Client, error)

NewSaramaClientFromBrokerList creates a new Client from a list of brokers

type SaramaBroker

type SaramaBroker interface {
	AddOffsetsToTxn(request *sarama.AddOffsetsToTxnRequest) (*sarama.AddOffsetsToTxnResponse, error)
	AddPartitionsToTxn(request *sarama.AddPartitionsToTxnRequest) (*sarama.AddPartitionsToTxnResponse, error)
	Addr() string
	AlterConfigs(request *sarama.AlterConfigsRequest) (*sarama.AlterConfigsResponse, error)
	ApiVersions(request *sarama.ApiVersionsRequest) (*sarama.ApiVersionsResponse, error)
	Close() error
	CommitOffset(request *sarama.OffsetCommitRequest) (*sarama.OffsetCommitResponse, error)
	Connected() (bool, error)
	CreateAcls(request *sarama.CreateAclsRequest) (*sarama.CreateAclsResponse, error)
	CreatePartitions(request *sarama.CreatePartitionsRequest) (*sarama.CreatePartitionsResponse, error)
	CreateTopics(request *sarama.CreateTopicsRequest) (*sarama.CreateTopicsResponse, error)
	DeleteAcls(request *sarama.DeleteAclsRequest) (*sarama.DeleteAclsResponse, error)
	DeleteGroups(request *sarama.DeleteGroupsRequest) (*sarama.DeleteGroupsResponse, error)
	DeleteRecords(request *sarama.DeleteRecordsRequest) (*sarama.DeleteRecordsResponse, error)
	DeleteTopics(request *sarama.DeleteTopicsRequest) (*sarama.DeleteTopicsResponse, error)
	DescribeAcls(request *sarama.DescribeAclsRequest) (*sarama.DescribeAclsResponse, error)
	DescribeConfigs(request *sarama.DescribeConfigsRequest) (*sarama.DescribeConfigsResponse, error)
	DescribeGroups(request *sarama.DescribeGroupsRequest) (*sarama.DescribeGroupsResponse, error)
	EndTxn(request *sarama.EndTxnRequest) (*sarama.EndTxnResponse, error)
	Fetch(request *sarama.FetchRequest) (*sarama.FetchResponse, error)
	FetchOffset(request *sarama.OffsetFetchRequest) (*sarama.OffsetFetchResponse, error)
	FindCoordinator(request *sarama.FindCoordinatorRequest) (*sarama.FindCoordinatorResponse, error)
	GetAvailableOffsets(request *sarama.OffsetRequest) (*sarama.OffsetResponse, error)
	GetConsumerMetadata(request *sarama.ConsumerMetadataRequest) (*sarama.ConsumerMetadataResponse, error)
	GetMetadata(request *sarama.MetadataRequest) (*sarama.MetadataResponse, error)
	Heartbeat(request *sarama.HeartbeatRequest) (*sarama.HeartbeatResponse, error)
	ID() int32
	InitProducerID(request *sarama.InitProducerIDRequest) (*sarama.InitProducerIDResponse, error)
	JoinGroup(request *sarama.JoinGroupRequest) (*sarama.JoinGroupResponse, error)
	LeaveGroup(request *sarama.LeaveGroupRequest) (*sarama.LeaveGroupResponse, error)
	ListGroups(request *sarama.ListGroupsRequest) (*sarama.ListGroupsResponse, error)
	Open(conf *sarama.Config) error
	Produce(request *sarama.ProduceRequest) (*sarama.ProduceResponse, error)
	Rack() string
	SyncGroup(request *sarama.SyncGroupRequest) (*sarama.SyncGroupResponse, error)
	TxnOffsetCommit(request *sarama.TxnOffsetCommitRequest) (*sarama.TxnOffsetCommitResponse, error)
}

SaramaBroker is an interface over sarama.Broker for mocking