func (tl TopicList) Find(name string) *Topic {
	for _, topic := range tl {
		if topic.Name == name {
			return topic
		}
	}
	return nil
}
Find returns the topic with the given name if it exists in the topic list,
and will return `nil` otherwise.
package kazoo
import (
	"encoding/json"
	"errors"
	"fmt"
	"math/rand"
	"strconv"
	"github.com/samuel/go-zookeeper/zk"
)
var (
	ErrInvalidPartitionCount    = errors.New("Number of partitions must be larger than 0")
	ErrInvalidReplicationFactor = errors.New("Replication factor must be between 1 and the number of brokers")
	ErrInvalidReplicaCount      = errors.New("All partitions must have the same number of replicas")
	ErrReplicaBrokerOverlap     = errors.New("All replicas for a partition must be on separate brokers")
	ErrInvalidBroker            = errors.New("Replica assigned to invalid broker")
	ErrMissingPartitionID       = errors.New("Partition ids must be sequential starting from 0")
	ErrDuplicatePartitionID     = errors.New("Each partition must have a unique ID")
)
type Topic struct {
	Name string
	kz   *Kazoo
}
type TopicList []*Topic
type Partition struct {
	topic    *Topic
	ID       int32
	Replicas []int32
}
type PartitionList []*Partition
func (kz *Kazoo) Topics() (TopicList, error) {
	root := fmt.Sprintf("%s/brokers/topics", kz.conf.Chroot)
	children, _, err := kz.conn.Children(root)
	if err != nil {
		return nil, err
	}
	result := make(TopicList, 0, len(children))
	for _, name := range children {
		result = append(result, kz.Topic(name))
	}
	return result, nil
}
func (kz *Kazoo) WatchTopics() (TopicList, <-chan zk.Event, error) {
	root := fmt.Sprintf("%s/brokers/topics", kz.conf.Chroot)
	children, _, c, err := kz.conn.ChildrenW(root)
	if err != nil {
		return nil, nil, err
	}
	result := make(TopicList, 0, len(children))
	for _, name := range children {
		result = append(result, kz.Topic(name))
	}
	return result, c, nil
}
func (kz *Kazoo) Topic(topic string) *Topic {
	return &Topic{Name: topic, kz: kz}
}
func (t *Topic) Exists() (bool, error) {
	return t.kz.exists(t.metadataPath())
}
func (t *Topic) Partitions() (PartitionList, error) {
	value, _, err := t.kz.conn.Get(t.metadataPath())
	if err != nil {
		return nil, err
	}
	return t.parsePartitions(value)
}
func (t *Topic) WatchPartitions() (PartitionList, <-chan zk.Event, error) {
	value, _, c, err := t.kz.conn.GetW(t.metadataPath())
	if err != nil {
		return nil, nil, err
	}
	list, err := t.parsePartitions(value)
	return list, c, err
}
func (t *Topic) Watch() (<-chan zk.Event, error) {
	_, _, c, err := t.kz.conn.GetW(t.metadataPath())
	if err != nil {
		return nil, err
	}
	return c, err
}
type topicMetadata struct {
	Version    int                `json:"version"`
	Partitions map[string][]int32 `json:"partitions"`
}
func (t *Topic) metadataPath() string {
	return fmt.Sprintf("%s/brokers/topics/%s", t.kz.conf.Chroot, t.Name)
}
func (t *Topic) parsePartitions(value []byte) (PartitionList, error) {
	var tm topicMetadata
	if err := json.Unmarshal(value, &tm); err != nil {
		return nil, err
	}
	result := make(PartitionList, len(tm.Partitions))
	for partitionNumber, replicas := range tm.Partitions {
		partitionID, err := strconv.ParseInt(partitionNumber, 10, 32)
		if err != nil {
			return nil, err
		}
		replicaIDs := make([]int32, 0, len(replicas))
		for _, r := range replicas {
			replicaIDs = append(replicaIDs, int32(r))
		}
		result[partitionID] = t.Partition(int32(partitionID), replicaIDs)
	}
	return result, nil
}
func (t *Topic) marshalPartitions(partitions PartitionList) ([]byte, error) {
	tm := topicMetadata{Version: 1, Partitions: make(map[string][]int32, len(partitions))}
	for _, part := range partitions {
		tm.Partitions[fmt.Sprintf("%d", part.ID)] = part.Replicas
	}
	return json.Marshal(tm)
}
func (t *Topic) generatePartitionAssignments(brokers []int32, partitionCount int, replicationFactor int) (PartitionList, error) {
	if partitionCount <= 0 {
		return nil, ErrInvalidPartitionCount
	}
	if replicationFactor <= 0 || len(brokers) < replicationFactor {
		return nil, ErrInvalidReplicationFactor
	}
	result := make(PartitionList, partitionCount)
	brokerCount := len(brokers)
	brokerIdx := rand.Intn(brokerCount)
	for p := 0; p < partitionCount; p++ {
		partition := &Partition{topic: t, ID: int32(p), Replicas: make([]int32, replicationFactor)}
		brokerIndices := rand.Perm(len(brokers))[0:replicationFactor]
		for r := 0; r < replicationFactor; r++ {
			partition.Replicas[r] = brokers[brokerIndices[r]]
		}
		result[p] = partition
		brokerIdx = (brokerIdx + 1) % brokerCount
	}
	return result, nil
}
func (t *Topic) validatePartitionAssignments(brokers []int32, assignment PartitionList) error {
	if len(assignment) == 0 {
		return ErrInvalidPartitionCount
	}
	var replicaCount int
	for _, part := range assignment {
		replicaCount = len(part.Replicas)
		break
	}
	if replicaCount == 0 {
		return ErrInvalidReplicationFactor
	}
	maxPartitionID := int32(-1)
	partitionIDmap := make(map[int32]struct{}, len(assignment))
	for _, part := range assignment {
		if part == nil {
			continue
		}
		if maxPartitionID < part.ID {
			maxPartitionID = part.ID
		}
		partitionIDmap[part.ID] = struct{}{}
		if len(part.Replicas) != replicaCount {
			return ErrInvalidReplicaCount
		}
		rset := make(map[int32]struct{}, replicaCount)
		for _, r := range part.Replicas {
			found := false
			for _, b := range brokers {
				if r == b {
					found = true
					break
				}
			}
			if !found {
				return ErrInvalidBroker
			}
			rset[r] = struct{}{}
		}
		if len(rset) != replicaCount {
			return ErrReplicaBrokerOverlap
		}
	}
	if int(maxPartitionID) != len(assignment)-1 {
		return ErrMissingPartitionID
	}
	if len(partitionIDmap) != len(assignment) {
		return ErrDuplicatePartitionID
	}
	return nil
}
func (t *Topic) Partition(id int32, replicas []int32) *Partition {
	return &Partition{ID: id, Replicas: replicas, topic: t}
}
type topicConfig struct {
	Version   int               `json:"version"`
	ConfigMap map[string]string `json:"config"`
}
func (t *Topic) configPath() string {
	return fmt.Sprintf("%s/config/topics/%s", t.kz.conf.Chroot, t.Name)
}
func (t *Topic) parseConfig(data []byte) (map[string]string, error) {
	var cfg topicConfig
	if err := json.Unmarshal(data, &cfg); err != nil {
		return nil, err
	}
	return cfg.ConfigMap, nil
}
func (t *Topic) marshalConfig(data map[string]string) ([]byte, error) {
	cfg := topicConfig{Version: 1, ConfigMap: data}
	if cfg.ConfigMap == nil {
		cfg.ConfigMap = make(map[string]string)
	}
	return json.Marshal(&cfg)
}
func (t *Topic) Config() (map[string]string, error) {
	value, _, err := t.kz.conn.Get(t.configPath())
	if err != nil {
		return nil, err
	}
	return t.parseConfig(value)
}
func (p *Partition) Topic() *Topic {
	return p.topic
}
func (p *Partition) Key() string {
	return fmt.Sprintf("%s/%d", p.topic.Name, p.ID)
}
func (p *Partition) PreferredReplica() int32 {
	if len(p.Replicas) > 0 {
		return p.Replicas[0]
	} else {
		return -1
	}
}
func (p *Partition) Leader() (int32, error) {
	if state, err := p.state(); err != nil {
		return -1, err
	} else {
		return state.Leader, nil
	}
}
func (p *Partition) ISR() ([]int32, error) {
	if state, err := p.state(); err != nil {
		return nil, err
	} else {
		return state.ISR, nil
	}
}
func (p *Partition) UnderReplicated() (bool, error) {
	if state, err := p.state(); err != nil {
		return false, err
	} else {
		return len(state.ISR) < len(p.Replicas), nil
	}
}
func (p *Partition) UsesPreferredReplica() (bool, error) {
	if state, err := p.state(); err != nil {
		return false, err
	} else {
		return len(state.ISR) > 0 && state.ISR[0] == p.Replicas[0], nil
	}
}
type partitionState struct {
	Leader int32   `json:"leader"`
	ISR    []int32 `json:"isr"`
}
func (p *Partition) state() (partitionState, error) {
	var state partitionState
	node := fmt.Sprintf("%s/brokers/topics/%s/partitions/%d/state", p.topic.kz.conf.Chroot, p.topic.Name, p.ID)
	value, _, err := p.topic.kz.conn.Get(node)
	if err != nil {
		return state, err
	}
	if err := json.Unmarshal(value, &state); err != nil {
		return state, err
	}
	return state, nil
}