func AssignmentDiffs(
	current []PartitionAssignment,
	desired []PartitionAssignment,
) []AssignmentDiff {
	diffsMap := map[int]AssignmentDiff{}
	for _, assignment := range current {
		diffsMap[assignment.ID] = AssignmentDiff{
			PartitionID: assignment.ID,
			Old:         assignment,
		}
	}
	for _, assignment := range desired {
		currDiff, ok := diffsMap[assignment.ID]
		if !ok {
			diffsMap[assignment.ID] = AssignmentDiff{
				PartitionID: assignment.ID,
				New:         assignment,
			}
		} else {
			diffsMap[assignment.ID] = AssignmentDiff{
				PartitionID: assignment.ID,
				Old:         currDiff.Old,
				New:         assignment,
			}
		}
	}
	partitionIDs := []int{}
	for partitionID := range diffsMap {
		partitionIDs = append(partitionIDs, partitionID)
	}
	sort.Slice(partitionIDs, func(a, b int) bool {
		return partitionIDs[a] < partitionIDs[b]
	})
	results := []AssignmentDiff{}
	for _, partitionID := range partitionIDs {
		results = append(results, diffsMap[partitionID])
	}
	return results
}
AssignmentDiffs returns the diffs implied by the argument current and
desired PartitionAssignments. Used for displaying diffs to user.
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
}
func MaxReplication(topics []TopicInfo) int {
	maxReplication := 0
	for _, topic := range topics {
		topicReplication := topic.MaxReplication()
		if topicReplication > maxReplication {
			maxReplication = topicReplication
		}
	}
	return maxReplication
}
func HasLeaders(topics []TopicInfo) bool {
	for _, topic := range topics {
		for _, partition := range topic.Partitions {
			if partition.Leader > 0 {
				return true
			}
		}
	}
	return false
}
func ThrottledTopicNames(topics []TopicInfo) []string {
	throttledNames := []string{}
	for _, topic := range topics {
		if topic.IsThrottled() {
			throttledNames = append(throttledNames, topic.Name)
		}
	}
	return throttledNames
}
func (p PartitionInfo) Racks(brokerRacks map[int]string) ([]string, error) {
	racks := []string{}
	for _, brokerID := range p.Replicas {
		rack, ok := brokerRacks[brokerID]
		if !ok {
			return nil, fmt.Errorf("Unrecognized broker ID: %d", brokerID)
		}
		racks = append(racks, rack)
	}
	return racks, nil
}
func (p PartitionInfo) NumRacks(brokerRacks map[int]string) (int, error) {
	racksMap := map[string]struct{}{}
	for _, brokerID := range p.Replicas {
		rack, ok := brokerRacks[brokerID]
		if !ok {
			return 0, fmt.Errorf("Unrecognized broker ID: %d", brokerID)
		}
		racksMap[rack] = struct{}{}
	}
	return len(racksMap), nil
}
func (t TopicInfo) ToAssignments() []PartitionAssignment {
	assignments := []PartitionAssignment{}
	for _, partitionInfo := range t.Partitions {
		assignments = append(
			assignments,
			PartitionAssignment{
				ID:       partitionInfo.ID,
				Replicas: util.CopyInts(partitionInfo.Replicas),
			},
		)
	}
	return assignments
}
func PartitionIDs(partitions []PartitionInfo) []int {
	ids := []int{}
	for _, partition := range partitions {
		ids = append(ids, partition.ID)
	}
	return ids
}
func (a PartitionAssignment) Index(replica int) int {
	for v, value := range a.Replicas {
		if value == replica {
			return v
		}
	}
	return -1
}
func (a PartitionAssignment) Copy() PartitionAssignment {
	return PartitionAssignment{
		ID:       a.ID,
		Replicas: util.CopyInts(a.Replicas),
	}
}
func CopyAssignments(
	curr []PartitionAssignment,
) []PartitionAssignment {
	copied := []PartitionAssignment{}
	for _, assignment := range curr {
		copied = append(copied, assignment.Copy())
	}
	return copied
}
func (a PartitionAssignment) DistinctRacks(
	brokerRacks map[int]string,
) map[string]struct{} {
	racksMap := map[string]struct{}{}
	for _, replica := range a.Replicas {
		racksMap[brokerRacks[replica]] = struct{}{}
	}
	return racksMap
}
func CheckAssignments(assignments []PartitionAssignment) error {
	if len(assignments) == 0 {
		return errors.New("Got zero-length slice")
	}
	var minReplicas, maxReplicas int
	for a, assignment := range assignments {
		numReplicas := len(assignment.Replicas)
		if a == 0 {
			minReplicas = numReplicas
			maxReplicas = numReplicas
		} else {
			if numReplicas < minReplicas {
				minReplicas = numReplicas
			}
			if numReplicas > maxReplicas {
				maxReplicas = numReplicas
			}
		}
		if a != assignment.ID {
			return errors.New("Slice elements not in order")
		}
		if hasRepeats(assignment) {
			return fmt.Errorf(
				"Found repeated partition in assignment: %+v",
				assignment,
			)
		}
	}
	if minReplicas != maxReplicas {
		return fmt.Errorf(
			"Partition replicas do not have consistent length (min: %d, max: %d)",
			minReplicas,
			maxReplicas,
		)
	}
	return nil
}
func hasRepeats(assignment PartitionAssignment) bool {
	replicasMap := map[int]struct{}{}
	for _, replica := range assignment.Replicas {
		if _, ok := replicasMap[replica]; ok {
			return true
		}
		replicasMap[replica] = struct{}{}
	}
	return false
}
func ReplicasToAssignments(
	replicaSlices [][]int,
) []PartitionAssignment {
	assignments := []PartitionAssignment{}
	for p, replicas := range replicaSlices {
		assignments = append(
			assignments,
			PartitionAssignment{
				ID:       p,
				Replicas: util.CopyInts(replicas),
			},
		)
	}
	return assignments
}
func AssignmentsToReplicas(assignments []PartitionAssignment) ([][]int, error) {
	replicaSlices := [][]int{}
	for a, assignment := range assignments {
		if a != assignment.ID {
			return nil, errors.New("Assignments are not in order")
		}
		replicaSlices = append(
			replicaSlices,
			assignment.Replicas,
		)
	}
	return replicaSlices, nil
}
func MaxPartitionsPerBroker(
	allAssignments ...[]PartitionAssignment,
) map[int]int {
	assignmentsPerPartition := map[int][]PartitionAssignment{}
	for _, assignments := range allAssignments {
		for _, assignment := range assignments {
			assignmentsPerPartition[assignment.ID] = append(
				assignmentsPerPartition[assignment.ID],
				assignment,
			)
		}
	}
	partitionsPerBrokerMap := map[int]map[int]struct{}{}
	for partition, assignments := range assignmentsPerPartition {
		partitionsPerBrokerMap[partition] = map[int]struct{}{}
		for _, assignment := range assignments {
			for _, replica := range assignment.Replicas {
				partitionsPerBrokerMap[partition][replica] = struct{}{}
			}
		}
	}
	partitionsPerBroker := map[int]int{}
	for _, brokersMap := range partitionsPerBrokerMap {
		for broker := range brokersMap {
			partitionsPerBroker[broker]++
		}
	}
	return partitionsPerBroker
}
func SameBrokers(
	a PartitionAssignment,
	b PartitionAssignment,
) bool {
	return util.SameElements(a.Replicas, b.Replicas)
}
type AssignmentDiff struct {
	PartitionID int
	Old         PartitionAssignment
	New         PartitionAssignment
}