kafka

Jul 1, 2020

Changes in this version

type StorageRequest

+ SizeByBroker map[int32]int64

+ SizeByTopic map[string]int64

type StorageRequestType

+ const StorageAddSizeByBroker

+ const StorageAddSizeByTopic

Apr 22, 2019

Changes in this version

type Cluster

+ func (module *Cluster) IsHealthy() bool

+ type ConsumerGroupMetadata struct

+ Group string

+ Header metadataHeader

+ Members []metadataMember

type StorageRequest

+ GroupMetadata *ConsumerGroupMetadata

+ PartitionCount int

+ TopicConfig *TopicConfiguration

type StorageRequestType

+ const StorageAddGroupMetadata

+ const StorageAddTopicConfiguration

+ const StorageDeleteGroupMetadata

+ const StorageDeleteTopic

+ const StorageRegisterOffsetPartitions

+ type TopicConfiguration struct

+ CleanupPolicy string

+ PartitionCount int

+ TopicName string

Mar 29, 2019

Changes in this version

type StorageRequestType

+ const StorageMarkOffsetPartitionReady

+ const StorageRegisterOffsetPartition

Mar 25, 2019

Changes in this version

+ type Cluster struct

+ func NewCluster(opts *options.Options, partitionWaterMarksCh chan *StorageRequest) *Cluster

+ func (module *Cluster) Start()

+ type ConsumerPartitionOffset struct

+ Group string

+ Offset int64

+ Partition int32

+ Timestamp int64

+ Topic string

+ type OffsetConsumer struct

+ func NewOffsetConsumer(opts *options.Options, storageChannel chan *StorageRequest) *OffsetConsumer

+ func (module *OffsetConsumer) Start()

+ type PartitionWaterMark struct

+ PartitionID int32

+ Timestamp int64

+ TopicName string

+ WaterMark int64

+ type StorageRequest struct

+ ConsumerGroupName string

+ ConsumerOffset *ConsumerPartitionOffset

+ PartitionID int32

+ PartitionWaterMark *PartitionWaterMark

+ RequestType StorageRequestType

+ TopicName string

+ type StorageRequestType int

+ const StorageAddConsumerOffset

+ const StorageAddPartitionHighWaterMark

+ const StorageAddPartitionLowWaterMark

+ const StorageDeleteConsumerGroup

Feb 11, 2019

Changes in this version

+ func CanReadCertAndKey(certPath, keyPath string) (bool, error)

+ type Client struct

+ func NewKafkaClient(opts *options.Options) (*Client, error)

+ func (c *Client) ConsumerGroupTopicLags(topicsByName map[string]*Topic) map[string][]*ConsumerGroupTopicLag

+ func (c *Client) GetPartitionIDs(topicName string) ([]int32, error)

+ func (c *Client) GetPartitionIDsBulk(topicNames []string) map[string][]int32

+ func (c *Client) GetPartitionOffsets(partitionIDsByTopicName map[string][]int32) map[string]*Topic

+ func (c *Client) GetTopicNames() ([]string, error)

+ func (c *Client) IsHealthy() bool

+ type ConsumerGroupTopicLag struct

+ Name string

+ TopicLag int64

+ TopicName string

+ type Partition struct

+ HighWaterMark int64

+ OldestOffset int64

+ PartitionID int32

+ func (p *Partition) MessageCount() int64

+ type Topic struct

+ MessageCount int64

+ Name string

+ Partitions map[int32]*Partition

+ func (t *Topic) HighWaterMarkByPartitionID(partitionID int32) int64