kafkatools

Jun 28, 2017

Changes in this version

+ func FetchTopicOffsets(client sarama.Client, offset int64, topic string) (topicOffsets map[int32]TopicPartitionOffset)

+ func GenerateOffsetRequests(client sarama.Client, time int64, topics ...string) (requests map[*sarama.Broker]*sarama.OffsetRequest)

+ func GetBrokerGroupOffsets(broker *sarama.Broker, groupOffsetChannel chan GroupOffset)

+ func GetBrokerTopicOffsets(broker *sarama.Broker, request *sarama.OffsetRequest, ...)

+ func GetOffsetFetchRequest(desc *sarama.GroupDescription) *sarama.OffsetFetchRequest

+ func GetSaramaClient(brokers ...string) sarama.Client

+ func GetSaramaConsumer(brokers string, consumerGroup string, topics []string) *cluster.Consumer

+ type GroupOffset struct

+ Group string

+ GroupTopicOffsets GroupTopicOffsetSlice

+ type GroupOffsetSlice []GroupOffset

+ func FetchOffsets(client sarama.Client, offset int64) (groupOffsets GroupOffsetSlice, ...)

+ func (a GroupOffsetSlice) Len() int

+ func (a GroupOffsetSlice) Less(i, j int) bool

+ func (a GroupOffsetSlice) Swap(i, j int)

+ type GroupTopicOffset struct

+ Topic string

+ TopicPartitionOffsets TopicPartitionOffsetSlice

+ type GroupTopicOffsetSlice []GroupTopicOffset

+ func (a GroupTopicOffsetSlice) Len() int

+ func (a GroupTopicOffsetSlice) Less(i, j int) bool

+ func (a GroupTopicOffsetSlice) Swap(i, j int)

+ type MemberAssignment struct

+ Assignments []TopicAssignment

+ Version int

+ func ParseMemberAssignment(byteArr []byte) (assignments MemberAssignment)

+ type TopicAssignment struct

+ Partitions []int32

+ Topic string

+ type TopicPartitionOffset struct

+ Offset int64

+ Partition int32

+ Topic string

+ type TopicPartitionOffsetSlice []TopicPartitionOffset

+ func (a TopicPartitionOffsetSlice) Len() int

+ func (a TopicPartitionOffsetSlice) Less(i, j int) bool

+ func (a TopicPartitionOffsetSlice) Swap(i, j int)