func MaxReplication(topics []TopicInfo) int {
	maxReplication := 0
	for _, topic := range topics {
		topicReplication := topic.MaxReplication()
		if topicReplication > maxReplication {
			maxReplication = topicReplication
		}
	}
	return maxReplication
}
MaxReplication returns the maximum amount of replication across all partitions
in the argument topics.
package admin
import (
	"errors"
	"fmt"
	"reflect"
	"sort"
	"strconv"
	"time"
	"github.com/segmentio/topicctl/pkg/util"
)
const (
	RetentionKey = "retention.ms"
	LeaderThrottledKey = "leader.replication.throttled.rate"
	FollowerThrottledKey = "follower.replication.throttled.rate"
	LeaderReplicasThrottledKey = "leader.replication.throttled.replicas"
	FollowerReplicasThrottledKey = "follower.replication.throttled.replicas"
)
type BrokerInfo struct {
	ID               int               `json:"id"`
	Endpoints        []string          `json:"endpoints"`
	Host             string            `json:"host"`
	Port             int32             `json:"port"`
	InstanceID       string            `json:"instanceID"`
	AvailabilityZone string            `json:"availabilityZone"`
	Rack             string            `json:"rack"`
	InstanceType     string            `json:"instanceType"`
	Version          int               `json:"version"`
	Timestamp        time.Time         `json:"timestamp"`
	Config           map[string]string `json:"config"`
}
type TopicInfo struct {
	Name       string            `json:"name"`
	Config     map[string]string `json:"config"`
	Partitions []PartitionInfo   `json:"partitions"`
	Version    int               `json:"version"`
}
type PartitionInfo struct {
	Topic           string `json:"topic"`
	ID              int    `json:"ID"`
	Leader          int    `json:"leader"`
	Version         int    `json:"version"`
	Replicas        []int  `json:"replicas"`
	ISR             []int  `json:"isr"`
	ControllerEpoch int    `json:"controllerEpoch"`
	LeaderEpoch     int    `json:"leaderEpoch"`
}
type PartitionAssignment struct {
	ID       int   `json:"id"`
	Replicas []int `json:"replicas"`
}
type zkClusterID struct {
	Version string `json:"version"`
	ID      string `json:"id"`
}
type zkBrokerInfo struct {
	Endpoints    []string `json:"endpoints"`
	Host         string   `json:"host"`
	Port         int32    `json:"port"`
	Rack         string   `json:"rack"`
	TimestampStr string   `json:"timestamp"`
	Version      int      `json:"version"`
}
type zkBrokerConfig struct {
	Version int               `json:"version"`
	Config  map[string]string `json:"config"`
}
type zkTopicInfo struct {
	Version    int              `json:"version"`
	Partitions map[string][]int `json:"partitions"`
}
type zkTopicConfig struct {
	Version int               `json:"version"`
	Config  map[string]string `json:"config"`
}
type zkPartitionInfo struct {
	Leader          int   `json:"leader"`
	Version         int   `json:"version"`
	ISR             []int `json:"isr"`
	ControllerEpoch int   `json:"controller_epoch"`
	LeaderEpoch     int   `json:"leader_epoch"`
}
type zkAssignment struct {
	Version    int                     `json:"version"`
	Partitions []zkAssignmentPartition `json:"partitions"`
}
type zkAssignmentPartition struct {
	Topic     string `json:"topic"`
	Partition int    `json:"partition"`
	Replicas  []int  `json:"replicas"`
}
type zkElection struct {
	Version    int                        `json:"version"`
	Partitions []zkElectionTopicPartition `json:"partitions"`
}
type zkElectionTopicPartition struct {
	Topic     string `json:"topic"`
	Partition int    `json:"partition"`
}
type zkChangeNotification struct {
	Version    int    `json:"version"`
	EntityPath string `json:"entity_path"`
}
func (b BrokerInfo) Addr() string {
	return fmt.Sprintf("%s:%d", b.Host, b.Port)
}
func (b BrokerInfo) IsThrottled() bool {
	_, leaderOk := b.Config[LeaderThrottledKey]
	_, followerOk := b.Config[FollowerThrottledKey]
	return leaderOk || followerOk
}
func BrokerIDs(brokers []BrokerInfo) []int {
	brokerIDs := []int{}
	for _, broker := range brokers {
		brokerIDs = append(brokerIDs, broker.ID)
	}
	return brokerIDs
}
func ThrottledBrokerIDs(brokers []BrokerInfo) []int {
	brokerIDs := []int{}
	for _, broker := range brokers {
		if broker.IsThrottled() {
			brokerIDs = append(brokerIDs, broker.ID)
		}
	}
	return brokerIDs
}
func BrokerRacks(brokers []BrokerInfo) map[int]string {
	brokerRacks := map[int]string{}
	for _, broker := range brokers {
		brokerRacks[broker.ID] = broker.Rack
	}
	return brokerRacks
}
func BrokersPerRack(brokers []BrokerInfo) map[string][]int {
	brokersPerRack := map[string][]int{}
	for _, broker := range brokers {
		rack := broker.Rack
		brokersPerRack[rack] = append(
			brokersPerRack[rack],
			broker.ID,
		)
	}
	return brokersPerRack
}
func BrokerCountsPerRack(brokers []BrokerInfo) map[string]int {
	brokersPerRack := BrokersPerRack(brokers)
	brokerCountsPerRack := map[string]int{}
	for rack, brokers := range brokersPerRack {
		brokerCountsPerRack[rack] = len(brokers)
	}
	return brokerCountsPerRack
}
func DistinctRacks(brokers []BrokerInfo) []string {
	brokersPerRack := BrokersPerRack(brokers)
	racks := []string{}
	for rack := range brokersPerRack {
		racks = append(racks, rack)
	}
	sort.Slice(racks, func(a, b int) bool {
		return racks[a] < racks[b]
	})
	return racks
}
func LeadersPerRack(brokers []BrokerInfo, topic TopicInfo) map[string]int {
	leadersPerRack := map[string]int{}
	brokerRacks := BrokerRacks(brokers)
	for _, partition := range topic.Partitions {
		leaderRack := brokerRacks[partition.Leader]
		leadersPerRack[leaderRack]++
	}
	return leadersPerRack
}
func (t TopicInfo) Retention() time.Duration {
	retentionStr, ok := t.Config[RetentionKey]
	if !ok {
		return 0
	}
	retention, err := strconv.ParseInt(retentionStr, 10, 64)
	if err != nil {
		return 0
	}
	return time.Duration(retention) * time.Millisecond
}
func (t TopicInfo) PartitionIDs() []int {
	ids := []int{}
	for _, partition := range t.Partitions {
		ids = append(ids, partition.ID)
	}
	return ids
}
func (t TopicInfo) MaxReplication() int {
	maxReplication := 0
	for _, partition := range t.Partitions {
		if len(partition.Replicas) > maxReplication {
			maxReplication = len(partition.Replicas)
		}
	}
	return maxReplication
}
func (t TopicInfo) MaxISR() int {
	maxISR := 0
	for _, partition := range t.Partitions {
		if len(partition.ISR) > maxISR {
			maxISR = len(partition.ISR)
		}
	}
	return maxISR
}
func (t TopicInfo) RackCounts(brokerRacks map[int]string) (int, int, error) {
	var minRacks, maxRacks int
	for p, partition := range t.Partitions {
		numRacks, err := partition.NumRacks(brokerRacks)
		if err != nil {
			return 0, 0, err
		}
		if p == 0 {
			minRacks = numRacks
			maxRacks = numRacks
		} else {
			if numRacks < minRacks {
				minRacks = numRacks
			}
			if numRacks > maxRacks {
				maxRacks = numRacks
			}
		}
	}
	return minRacks, maxRacks, nil
}
func (t TopicInfo) AllReplicasInSync() bool {
	for _, partition := range t.Partitions {
		if !util.SameElements(partition.Replicas, partition.ISR) {
			return false
		}
	}
	return true
}
func (t TopicInfo) OutOfSyncPartitions(subset []int) []PartitionInfo {
	outOfSync := []PartitionInfo{}
	subsetMap := map[int]struct{}{}
	for _, id := range subset {
		subsetMap[id] = struct{}{}
	}
	for _, partition := range t.Partitions {
		if _, ok := subsetMap[partition.ID]; len(subset) > 0 && !ok {
			continue
		}
		if !util.SameElements(partition.Replicas, partition.ISR) {
			outOfSync = append(outOfSync, partition)
		}
	}
	return outOfSync
}
func (t TopicInfo) AllLeadersCorrect() bool {
	for _, partition := range t.Partitions {
		if partition.Leader != partition.Replicas[0] {
			return false
		}
	}
	return true
}
func (t TopicInfo) WrongLeaderPartitions(subset []int) []PartitionInfo {
	wrongLeaders := []PartitionInfo{}
	subsetMap := map[int]struct{}{}
	for _, id := range subset {
		subsetMap[id] = struct{}{}
	}
	for _, partition := range t.Partitions {
		if _, ok := subsetMap[partition.ID]; len(subset) > 0 && !ok {
			continue
		}
		if partition.Leader != partition.Replicas[0] {
			wrongLeaders = append(wrongLeaders, partition)
		}
	}
	return wrongLeaders
}
func (t TopicInfo) IsThrottled() bool {
	_, leaderOk := t.Config[LeaderReplicasThrottledKey]
	_, followerOk := t.Config[FollowerReplicasThrottledKey]
	return leaderOk || followerOk
}