func (c *ZKAdminClient) AssignPartitions(
	ctx context.Context,
	topic string,
	assignments []PartitionAssignment,
) error {
	if c.readOnly {
		return errors.New("Cannot assign partitions in read-only mode")
	}
	zkAssignmentObj := zkAssignment{
		Version:    1,
		Partitions: []zkAssignmentPartition{},
	}
	for _, assignment := range assignments {
		zkAssignmentObj.Partitions = append(
			zkAssignmentObj.Partitions,
			zkAssignmentPartition{
				Topic:     topic,
				Partition: assignment.ID,
				Replicas:  util.CopyInts(assignment.Replicas),
			},
		)
	}
	zNode := c.zNode(assignmentPath)
	log.Infof(
		"Writing reassignment config to zk path %s: %+v",
		zNode,
		zkAssignmentObj,
	)
	return c.zkClient.CreateJSON(
		ctx,
		zNode,
		zkAssignmentObj,
		false,
	)
}
AssignPartitions notifies the cluster to begin a partition reassignment.
This should only be used for existing partitions; to create new partitions,
use the AddPartitions method.
package admin
import (
	"context"
	"errors"
	"fmt"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"time"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/ec2"
	"github.com/segmentio/kafka-go"
	"github.com/segmentio/topicctl/pkg/util"
	"github.com/segmentio/topicctl/pkg/zk"
	log "github.com/sirupsen/logrus"
)
const (
	assignmentPath    = "/admin/reassign_partitions"
	electionPath      = "/admin/preferred_replica_election"
	brokersPath       = "/brokers/ids"
	topicsPath        = "/brokers/topics"
	clusterIDPath     = "/cluster/id"
	brokerConfigsPath = "/config/brokers"
	configChangesPath = "/config/changes/config_change_"
	topicConfigsPath  = "/config/topics"
	maxPoolSize = 20
)
var (
	ErrTopicDoesNotExist = errors.New("Topic does not exist")
)
type ZKAdminClient struct {
	zkClient       zk.Client
	zkPrefix       string
	bootstrapAddrs []string
	Connector      *Connector
	sess           *session.Session
	readOnly       bool
}
var _ Client = (*ZKAdminClient)(nil)
type ZKAdminClientConfig struct {
	ZKAddrs           []string
	ZKPrefix          string
	BootstrapAddrs    []string
	ExpectedClusterID string
	Sess              *session.Session
	ReadOnly          bool
}
func NewZKAdminClient(
	ctx context.Context,
	config ZKAdminClientConfig,
) (*ZKAdminClient, error) {
	zkClient, err := zk.NewPooledClient(
		config.ZKAddrs,
		time.Minute,
		&zk.ZKDebugLogger{},
		10,
		config.ReadOnly,
	)
	if err != nil {
		return nil, err
	}
	zkPrefix := config.ZKPrefix
	if zkPrefix != "" {
		if !strings.HasPrefix(zkPrefix, "/") {
			zkPrefix = fmt.Sprintf("/%s", zkPrefix)
		}
		if strings.HasSuffix(zkPrefix, "/") {
			zkPrefix = zkPrefix[:len(zkPrefix)-1]
		}
	}
	client := &ZKAdminClient{
		zkClient: zkClient,
		zkPrefix: zkPrefix,
		sess:     config.Sess,
		readOnly: config.ReadOnly,
	}
	if config.ExpectedClusterID != "" {
		log.Info("Checking cluster ID against version in cluster")
		clusterID, err := client.GetClusterID(ctx)
		if err != nil {
			return nil, err
		}
		if clusterID != config.ExpectedClusterID {
			return nil, fmt.Errorf(
				"ID in cluster (%s) does not match expected one (%s)",
				clusterID,
				config.ExpectedClusterID,
			)
		}
	}
	var bootstrapAddrs []string
	if len(config.BootstrapAddrs) == 0 {
		log.Debug("No bootstrap addresses provided, getting one from zookeeper")
		ids, err := client.GetBrokerIDs(ctx)
		if err != nil {
			return nil, err
		}
		brokers, err := client.GetBrokers(ctx, []int{ids[0]})
		if err != nil {
			return nil, err
		}
		bootstrapAddrs = []string{brokers[0].Addr()}
	} else {
		bootstrapAddrs = append(bootstrapAddrs, config.BootstrapAddrs...)
	}
	client.bootstrapAddrs = bootstrapAddrs
	client.Connector, err = NewConnector(
		ConnectorConfig{
			BrokerAddr: bootstrapAddrs[0],
		},
	)
	return client, nil
}
func (c *ZKAdminClient) GetClusterID(
	ctx context.Context,
) (string, error) {
	zkClusterIDPath := c.zNode(clusterIDPath)
	zkClusterIDObj := zkClusterID{}
	_, err := c.zkClient.GetJSON(ctx, zkClusterIDPath, &zkClusterIDObj)
	if err != nil {
		return "", err
	}
	return zkClusterIDObj.ID, nil
}
func (c *ZKAdminClient) GetBrokers(
	ctx context.Context,
	ids []int,
) ([]BrokerInfo, error) {
	var brokerIDs []int
	var err error
	if len(ids) > 0 {
		brokerIDs = ids
	} else {
		brokerIDs, err = c.GetBrokerIDs(ctx)
		if err != nil {
			return nil, err
		}
	}
	brokers := []BrokerInfo{}
	brokerHosts := []string{}
	for _, id := range brokerIDs {
		zkBrokerInfo := zkBrokerInfo{}
		_, err = c.zkClient.GetJSON(
			ctx,
			c.zNode(brokersPath, fmt.Sprintf("%d", id)),
			&zkBrokerInfo,
		)
		if err != nil {
			return nil, err
		}
		zBrokerConfigPath := c.zNode(brokerConfigsPath, fmt.Sprintf("%d", id))
		zkBrokerConfig := zkBrokerConfig{}
		exists, _, err := c.zkClient.Exists(
			ctx,
			zBrokerConfigPath,
		)
		if err != nil {
			return nil, err
		}
		if exists {
			_, err = c.zkClient.GetJSON(
				ctx,
				zBrokerConfigPath,
				&zkBrokerConfig,
			)
			if err != nil {
				return nil, err
			}
		}
		epochMillis, err := strconv.ParseInt(
			zkBrokerInfo.TimestampStr,
			10,
			64,
		)
		if err != nil {
			return nil, err
		}
		brokerInfo := BrokerInfo{
			ID:        int(id),
			Endpoints: zkBrokerInfo.Endpoints,
			Host:      zkBrokerInfo.Host,
			Port:      zkBrokerInfo.Port,
			Rack:      zkBrokerInfo.Rack,
			Version:   zkBrokerInfo.Version,
			Timestamp: time.Unix(epochMillis/1000, 0),
			Config:    zkBrokerConfig.Config,
		}
		brokerHosts = append(
			brokerHosts,
			zkBrokerInfo.Host,
		)
		brokers = append(
			brokers,
			brokerInfo,
		)
	}
	instances, err := c.getInstances(ctx, brokerHosts)
	if err != nil {
		log.Debugf("Could not get instance info from EC2: %+v", err)
	}
	for b := 0; b < len(brokers); b++ {
		instance, ok := instances[brokers[b].Host]
		if !ok {
			continue
		}
		brokers[b].InstanceID = aws.StringValue(instance.InstanceId)
		brokers[b].InstanceType = aws.StringValue(instance.InstanceType)
		brokers[b].AvailabilityZone = aws.StringValue(
			instance.Placement.AvailabilityZone,
		)
	}
	sort.Slice(brokers, func(i, j int) bool {
		return brokers[i].ID < brokers[j].ID
	})
	return brokers, nil
}
func (c *ZKAdminClient) GetBrokerIDs(ctx context.Context) ([]int, error) {
	zPath := c.zNode(brokersPath)
	brokerIDStrs, _, err := c.zkClient.Children(ctx, zPath)
	if err != nil {
		return nil, fmt.Errorf(
			"Error getting children at path %s: %+v",
			zPath,
			err,
		)
	}
	brokerIDs := []int{}
	for _, idStr := range brokerIDStrs {
		id, err := strconv.ParseInt(idStr, 10, 32)
		if err != nil {
			return nil, err
		}
		brokerIDs = append(brokerIDs, int(id))
	}
	return brokerIDs, nil
}
func (c *ZKAdminClient) GetConnector() *Connector {
	return c.Connector
}
func (c *ZKAdminClient) GetTopics(
	ctx context.Context,
	names []string,
	detailed bool,
) ([]TopicInfo, error) {
	var topicNames []string
	var err error
	if len(names) > 0 {
		topicNames = names
	} else {
		topicNames, err = c.GetTopicNames(ctx)
		if err != nil {
			return nil, err
		}
	}
	log.Debugf(
		"Looking up %d topic names: %+v",
		len(topicNames),
		topicNames,
	)
	topics := []TopicInfo{}
	type topicReq struct {
		name     string
		detailed bool
	}
	type topicResp struct {
		topic TopicInfo
		err   error
	}
	topicReqChan := make(chan topicReq, len(topicNames))
	topicRespChan := make(chan topicResp, len(topicNames))
	defer close(topicReqChan)
	for _, name := range topicNames {
		topicReqChan <- topicReq{
			name:     name,
			detailed: detailed,
		}
	}
	poolSize := minInt(len(topicNames), maxPoolSize)
	for i := 0; i < poolSize; i++ {
		go func() {
			for {
				topicReq, ok := <-topicReqChan
				if !ok {
					return
				}
				topic, err := c.getTopic(ctx, topicReq.name, topicReq.detailed)
				topicRespChan <- topicResp{
					topic: topic,
					err:   err,
				}
			}
		}()
	}
	for i := 0; i < len(topicNames); i++ {
		topicResp := <-topicRespChan
		if topicResp.err != nil {
			return nil, err
		}
		topics = append(topics, topicResp.topic)
	}
	sort.Slice(topics, func(i, j int) bool {
		return topics[i].Name < topics[j].Name
	})
	return topics, nil
}
func (c *ZKAdminClient) GetTopicNames(ctx context.Context) ([]string, error) {
	zPath := c.zNode(topicsPath)
	topicNames, _, err := c.zkClient.Children(ctx, zPath)
	if err != nil {
		return nil, fmt.Errorf(
			"Error getting children at path %s: %+v",
			zPath,
			err,
		)
	}
	return topicNames, nil
}
func (c *ZKAdminClient) GetTopic(
	ctx context.Context,
	name string,
	detailed bool,
) (TopicInfo, error) {
	topics, err := c.GetTopics(ctx, []string{name}, detailed)
	if err != nil {
		if strings.Contains(err.Error(), "node does not exist") {
			return TopicInfo{}, ErrTopicDoesNotExist
		}
		return TopicInfo{}, err
	}
	if len(topics) != 1 {
		return TopicInfo{}, ErrTopicDoesNotExist
	}
	return topics[0], nil
}
func (c *ZKAdminClient) UpdateTopicConfig(
	ctx context.Context,
	name string,
	configEntries []kafka.ConfigEntry,
	overwrite bool,
) ([]string, error) {
	var updatedKeys []string
	if c.readOnly {
		return updatedKeys, errors.New("Cannot update topic in read-only mode")
	}
	log.Debugf("Updating config for topic %s", name)
	configMap := map[string]interface{}{}
	zPath := c.zNode(topicConfigsPath, name)
	stats, err := c.zkClient.GetJSON(
		ctx,
		zPath,
		&configMap,
	)
	if err != nil {
		return updatedKeys, err
	}
	updatedKeys, err = updateConfig(configMap, configEntries, overwrite)
	if err != nil {
		return updatedKeys, err
	}
	_, err = c.zkClient.SetJSON(ctx, zPath, configMap, stats.Version)
	if err != nil {
		return updatedKeys, err
	}
	changeObj := zkChangeNotification{
		Version:    2,
		EntityPath: fmt.Sprintf("topics/%s", name),
	}
	log.Debugf("Setting change notification: %+v", changeObj)
	return updatedKeys,
		c.zkClient.CreateJSON(ctx, c.zNode(configChangesPath), changeObj, true)
}
func (c *ZKAdminClient) UpdateBrokerConfig(
	ctx context.Context,
	id int,
	configEntries []kafka.ConfigEntry,
	overwrite bool,
) ([]string, error) {
	var updatedKeys []string
	if c.readOnly {
		return updatedKeys, errors.New("Cannot update topic in read-only mode")
	}
	log.Debugf("Updating config for broker %d", id)
	idStr := fmt.Sprintf("%d", id)
	zBrokerRoot := c.zNode(brokerConfigsPath)
	exists, _, err := c.zkClient.Exists(ctx, zBrokerRoot)
	if err != nil {
		return nil, err
	}
	if !exists {
		log.Infof("Creating broker configs path: %s", zBrokerRoot)
		err := c.zkClient.Create(ctx, zBrokerRoot, nil, false)
		if err != nil {
			return updatedKeys, err
		}
	}
	zBrokerPath := c.zNode(brokerConfigsPath, idStr)
	exists, _, err = c.zkClient.Exists(ctx, zBrokerPath)
	if err != nil {
		return nil, err
	}
	if !exists {
		zkBrokerConfigObj := zkBrokerConfig{
			Version: 1,
			Config:  map[string]string{},
		}
		log.Infof("Creating broker config at %s: %+v", zBrokerPath, zkBrokerConfigObj)
		err := c.zkClient.CreateJSON(ctx, zBrokerPath, zkBrokerConfigObj, false)
		if err != nil {
			return updatedKeys, err
		}
	}
	configMap := map[string]interface{}{}
	zPath := c.zNode(brokerConfigsPath, idStr)
	stats, err := c.zkClient.GetJSON(
		ctx,
		zPath,
		&configMap,
	)
	if err != nil {
		return updatedKeys, err
	}
	updatedKeys, err = updateConfig(configMap, configEntries, overwrite)
	if err != nil {
		return updatedKeys, err
	}
	_, err = c.zkClient.SetJSON(ctx, zPath, configMap, stats.Version)
	if err != nil {
		return updatedKeys, err
	}
	changeObj := zkChangeNotification{
		Version:    2,
		EntityPath: fmt.Sprintf("brokers/%s", idStr),
	}
	log.Debugf("Setting change notification: %+v", changeObj)
	return updatedKeys,
		c.zkClient.CreateJSON(ctx, c.zNode(configChangesPath), changeObj, true)
}
func (c *ZKAdminClient) CreateTopic(
	ctx context.Context,
	config kafka.TopicConfig,
) error {
	if c.readOnly {
		return errors.New("Cannot create topic in read-only mode")
	}
	req := kafka.CreateTopicsRequest{
		Topics: []kafka.TopicConfig{config},
	}
	log.Debugf("Creating topic with config %+v", config)
	_, err := c.Connector.KafkaClient.CreateTopics(ctx, &req)
	return err
}