func (c *CLIRunner) GetPartitions(ctx context.Context, topic string) error {
	c.startSpinner()
	topicInfo, err := c.adminClient.GetTopic(ctx, topic, true)
	if err != nil {
		c.stopSpinner()
		return err
	}
	brokers, err := c.adminClient.GetBrokers(ctx, nil)
	c.stopSpinner()
	if err != nil {
		return err
	}
	c.printer(
		"Partitions for topic %s:\n%s",
		topic,
		admin.FormatTopicPartitions(topicInfo.Partitions, brokers),
	)
	return nil
}
GetPartitions fetches the details of each partition in a topic and prints out a summary for
user inspection.
package cli
import (
	"context"
	"fmt"
	"io/ioutil"
	"os"
	"path/filepath"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"time"
	"github.com/briandowns/spinner"
	"github.com/segmentio/topicctl/pkg/admin"
	"github.com/segmentio/topicctl/pkg/apply"
	"github.com/segmentio/topicctl/pkg/check"
	"github.com/segmentio/topicctl/pkg/config"
	"github.com/segmentio/topicctl/pkg/groups"
	"github.com/segmentio/topicctl/pkg/messages"
	log "github.com/sirupsen/logrus"
)
const (
	spinnerCharSet  = 36
	spinnerDuration = 200 * time.Millisecond
)
type CLIRunner struct {
	adminClient admin.Client
	printer     func(f string, a ...interface{})
	spinnerObj  *spinner.Spinner
}
func NewCLIRunner(
	adminClient admin.Client,
	printer func(f string, a ...interface{}),
	showSpinner bool,
) *CLIRunner {
	var spinnerObj *spinner.Spinner
	if showSpinner {
		spinnerObj = spinner.New(
			spinner.CharSets[spinnerCharSet],
			spinnerDuration,
			spinner.WithWriter(os.Stderr),
			spinner.WithHiddenCursor(true),
		)
		spinnerObj.Prefix = "Loading: "
	}
	cliRunner := &CLIRunner{
		adminClient: adminClient,
		printer:     printer,
		spinnerObj:  spinnerObj,
	}
	return cliRunner
}
func (c *CLIRunner) GetBrokers(ctx context.Context, full bool) error {
	c.startSpinner()
	brokers, err := c.adminClient.GetBrokers(ctx, nil)
	c.stopSpinner()
	if err != nil {
		return err
	}
	c.printer("Brokers:\n%s", admin.FormatBrokers(brokers, full))
	c.printer("Brokers per rack:\n%s", admin.FormatBrokersPerRack(brokers))
	return nil
}
func (c *CLIRunner) ApplyTopic(
	ctx context.Context,
	applierConfig apply.TopicApplierConfig,
) error {
	applier, err := apply.NewTopicApplier(
		ctx,
		c.adminClient,
		applierConfig,
	)
	if err != nil {
		return err
	}
	c.printer(
		"Starting apply for topic %s in environment %s, cluster %s",
		applierConfig.TopicConfig.Meta.Name,
		applierConfig.TopicConfig.Meta.Environment,
		applierConfig.TopicConfig.Meta.Cluster,
	)
	err = applier.Apply(ctx)
	if err != nil {
		return err
	}
	c.printer("Apply completed successfully!")
	return nil
}
func (c *CLIRunner) BootstrapTopics(
	ctx context.Context,
	topics []string,
	clusterConfig config.ClusterConfig,
	matchRegexpStr string,
	excludeRegexpStr string,
	outputDir string,
	overwrite bool,
) error {
	topicInfoObjs, err := c.adminClient.GetTopics(ctx, topics, false)
	if err != nil {
		return err
	}
	matchRegexp, err := regexp.Compile(matchRegexpStr)
	if err != nil {
		return err
	}
	excludeRegexp, err := regexp.Compile(excludeRegexpStr)
	if err != nil {
		return err
	}
	topicConfigs := []config.TopicConfig{}
	for _, topicInfo := range topicInfoObjs {
		if strings.HasPrefix(topicInfo.Name, "__") {
			continue
		} else if !matchRegexp.MatchString(topicInfo.Name) {
			continue
		} else if excludeRegexp.MatchString(topicInfo.Name) {
			continue
		}
		topicConfig := config.TopicConfigFromTopicInfo(
			clusterConfig,
			topicInfo,
		)
		topicConfigs = append(topicConfigs, topicConfig)
	}
	for _, topicConfig := range topicConfigs {
		yamlStr, err := topicConfig.ToYAML()
		if err != nil {
			return err
		}
		if outputDir != "" {
			outputPath := filepath.Join(
				outputDir,
				fmt.Sprintf("%s.yaml", topicConfig.Meta.Name),
			)
			var isNew bool
			_, err := os.Stat(outputPath)
			if err != nil && !os.IsNotExist(err) {
				return err
			}
			isNew = os.IsNotExist(err)
			if isNew || overwrite {
				log.Infof("Writing config to %s", outputPath)
				err = ioutil.WriteFile(outputPath, []byte(yamlStr), 0644)
				if err != nil {
					return err
				}
			} else {
				log.Infof("Skipping over existing config %s", outputPath)
			}
		} else {
			log.Infof("Config for topic %s:\n%s", topicConfig.Meta.Name, yamlStr)
		}
	}
	return nil
}
func (c *CLIRunner) CheckTopic(
	ctx context.Context,
	checkConfig check.CheckConfig,
) (bool, error) {
	results, err := check.CheckTopic(ctx, checkConfig)
	if results.AllOK() {
		c.printer(
			"Topic %s (cluster=%s, env=%s) OK",
			checkConfig.TopicConfig.Meta.Name,
			checkConfig.ClusterConfig.Meta.Name,
			checkConfig.ClusterConfig.Meta.Environment,
		)
	} else {
		c.printer(
			"Check failed for topic %s (cluster=%s, env=%s):\n%s",
			checkConfig.TopicConfig.Meta.Name,
			checkConfig.ClusterConfig.Meta.Name,
			checkConfig.ClusterConfig.Meta.Environment,
			check.FormatResults(results),
		)
	}
	return results.AllOK(), err
}
func (c *CLIRunner) GetBrokerBalance(ctx context.Context, topicName string) error {
	c.startSpinner()
	brokers, err := c.adminClient.GetBrokers(ctx, nil)
	if err != nil {
		c.stopSpinner()
		return err
	}
	var topicNames []string
	if topicName != "" {
		topicNames = []string{topicName}
	}
	topics, err := c.adminClient.GetTopics(ctx, topicNames, false)
	if err != nil {
		c.stopSpinner()
		return err
	}
	c.stopSpinner()
	c.printer("Broker replicas:\n%s", admin.FormatBrokerReplicas(brokers, topics))
	c.printer("Broker rack replicas:\n%s", admin.FormatBrokerRackReplicas(brokers, topics))
	return nil
}
func (c *CLIRunner) GetConfig(ctx context.Context, brokerOrTopic string) error {
	c.startSpinner()
	brokerIDs, err := c.adminClient.GetBrokerIDs(ctx)
	if err != nil {
		c.stopSpinner()
		return err
	}
	topicNames, err := c.adminClient.GetTopicNames(ctx)
	if err != nil {
		c.stopSpinner()
		return err
	}
	for _, brokerID := range brokerIDs {
		if fmt.Sprintf("%d", brokerID) == brokerOrTopic {
			brokers, err := c.adminClient.GetBrokers(ctx, []int{brokerID})
			if err != nil {
				c.stopSpinner()
				return err
			}
			if len(brokers) != 1 {
				c.stopSpinner()
				return fmt.Errorf("Could not find broker %d", brokerID)
			}
			c.stopSpinner()
			c.printer(
				"Config for broker %d:\n%s",
				brokerID,
				admin.FormatConfig(brokers[0].Config),
			)
			return nil
		}
	}
	for _, topicName := range topicNames {
		if topicName == brokerOrTopic {
			topics, err := c.adminClient.GetTopics(ctx, []string{topicName}, false)
			if err != nil {
				c.stopSpinner()
				return err
			}
			if len(topics) != 1 {
				c.stopSpinner()
				return fmt.Errorf("Could not find topic %s", topicName)
			}
			c.stopSpinner()
			c.printer(
				"Config for topic %s:\n%s",
				topicName,
				admin.FormatConfig(topics[0].Config),
			)
			return nil
		}
	}
	c.stopSpinner()
	return fmt.Errorf("Could not find broker or topic named %s", brokerOrTopic)
}
func (c *CLIRunner) GetGroups(ctx context.Context) error {
	c.startSpinner()
	groupCoordinators, err := groups.GetGroups(ctx, c.adminClient.GetConnector())
	c.stopSpinner()
	if err != nil {
		return err
	}
	c.printer("Groups:\n%s", groups.FormatGroupCoordinators(groupCoordinators))
	return nil
}
func (c *CLIRunner) GetGroupMembers(ctx context.Context, groupID string, full bool) error {
	c.startSpinner()
	groupDetails, err := groups.GetGroupDetails(
		ctx,
		c.adminClient.GetConnector(),
		groupID,
	)
	c.stopSpinner()
	if err != nil {
		return err
	}
	c.printer("Group state: %s", groupDetails.State)
	c.printer(
		"Group members (%d):\n%s",
		len(groupDetails.Members),
		groups.FormatGroupMembers(groupDetails.Members, full),
	)
	c.printer(
		"Member frequency by partition count:\n%s",
		groups.FormatMemberPartitionCounts(groupDetails.Members),
	)
	return nil
}
func (c *CLIRunner) GetMemberLags(
	ctx context.Context,
	topic string,
	groupID string,
	full bool,
	sortByValues bool,
) error {
	c.startSpinner()
	_, err := c.adminClient.GetTopic(ctx, topic, false)
	if err != nil {
		c.stopSpinner()
		return fmt.Errorf("Error fetching topic info: %+v", err)
	}
	memberLags, err := groups.GetMemberLags(
		ctx,
		c.adminClient.GetConnector(),
		topic,
		groupID,
	)
	c.stopSpinner()
	if err != nil {
		return err
	}
	if sortByValues {
		sort.Slice(memberLags, func(a, b int) bool {
			return memberLags[a].TimeLag() < memberLags[b].TimeLag()
		})
	}
	c.printer(
		"Group member lags:\n%s",
		groups.FormatMemberLags(memberLags, full),
	)
	return nil
}