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
}
GetBrokers gets information on one or more cluster brokers from zookeeper.
If the argument ids is unset, then it fetches all brokers.
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
}