func CreateConsumerGroup(c *cli.Context, cliHelper common.CliHelper, serviceName string) {
	CreateConsumerGroupSecure(c, cliHelper, serviceName, nil)
}
CreateConsumerGroup creates consumer group based on cli.Context
package common
import (
	"bufio"
	"encoding/json"
	"errors"
	"fmt"
	"math"
	"math/rand"
	"os"
	"regexp"
	"runtime/debug"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"
	"github.com/apache/thrift/lib/go/thrift"
	"github.com/codegangsta/cli"
	ccli "github.com/uber/cherami-client-go/client/cherami"
	"github.com/uber/cherami-server/clients/inputhost"
	mcli "github.com/uber/cherami-server/clients/metadata"
	"github.com/uber/cherami-server/clients/outputhost"
	"github.com/uber/cherami-server/clients/storehost"
	"github.com/uber/cherami-server/common"
	"github.com/uber/cherami-thrift/.generated/go/admin"
	"github.com/uber/cherami-thrift/.generated/go/cherami"
	"github.com/uber/cherami-thrift/.generated/go/metadata"
	"github.com/uber/cherami-thrift/.generated/go/shared"
	"github.com/uber/cherami-thrift/.generated/go/store"
)
type GlobalOptions struct {
	hyperbahn              bool
	hyperbahnBootstrapFile string
	env                    string
	frontendHost           string
	frontendPort           int
	timeoutSecs            int
}
const (
	DefaultPageSize = 1000
	UnknownUUID = " UNKNOWN_HOST_UUID"
	DestinationType = "DST"
	ConsumerGroupType = "CG"
	DefaultUnconsumedMessagesRetention = 3 * 24 * 3600  
	DefaultConsumedMessagesRetention = 1 * 24 * 3600  
	MinUnconsumedMessagesRetentionForMultiZoneDest = 3 * 24 * 3600
	MinConsumedMessagesRetention = 180
	MaxConsumedMessagesRetention = 7 * 24 * 3600
	MinUnconsumedMessagesRetention = 180
	MaxUnconsumedMessagesRetention = 7 * 24 * 3600
	DefaultLockTimeoutSeconds = 60
	MinLockTimeoutSeconds = 10
	MaxLockTimeoutSeconds = 3600
	DefaultMaxDeliveryCount = 10
	MinMaxDeliveryCount = 1
	MaxMaxDeliveryCount = 1000
	DefaultSkipOlderMessageSeconds = 0
	MinSkipOlderMessageSeconds = 1800
	MaxSkipOlderMessageSeconds = 2 * 24 * 3600
	DefaultDelayMessageSeconds = 0
	MinDelayMessageSeconds = 0
	MaxDelayMessageSeconds = 2 * 24 * 3600
	kafkaPrefix = `/kafka_`
)
const (
	strNotEnoughArgs                  = "Not enough arguments. Try \"-help\""
	strTooManyArgs                    = "Too many arguments. Try \"-help\""
	strNoChange                       = "Update must update something. Try \"-help\""
	strCGSpecIncorrectArgs            = "Incorrect consumer group specification. Use \"<cg_uuid>\" or \"<dest_path> <cg_name>\""
	strDestStatus                     = "Destination status must be \"enabled\", \"disabled\", \"sendonly\", or \"recvonly\""
	strCGStatus                       = "Consumer group status must be \"enabled\", or \"disabled\""
	strWrongDestZoneConfig            = "Format of destination zone config is wrong, should be \"ZoneName,AllowPublish,AllowConsume,ReplicaCount\". For example: \"zone1,true,true,3\""
	strWrongReplicaCount              = "Replica count must be within 1 to 3"
	strWrongZoneConfigCount           = "Multi zone destination or consumer group must have at least 2 zone configs"
	strWrongCgZoneConfig              = "Format of consumer group zone config is wrong, should be \"ZoneName,PreferedActiveZone\". For example: \"zone1,false\""
	strMultiZoneCgWithSingleZoneDest  = "Multi zone consumer group must be created with a multi zone destination"
	strMultiplePreferedActiveZone     = "At most one zone can be prefered active zone"
	strUnconsumedRetentionTooSmall    = "Unconsumed retention period for multi zone destination should be at least 3 days"
	strKafkaNaming                    = "Kafka destinations and consumer groups must begin with \"" + kafkaPrefix + "\""
	strKafkaNotEnoughArgs             = "Kafka destinations must specify the Kafka cluster and at least one topic"
	strInvalidConsumedRetention       = "Invalid consumed message retention %v (must be between %v and %v), please contact cherami team"
	strInvalidUnconsumedRetention     = "Invalid unconsumed message retention %v (must be between %v and %v), please contact cherami team"
	strInvalidLockTimeoutSeconds      = "Invalid lock timeout seconds %v (must be between %v and %v), please contact cherami team"
	strInvalidMaxDeliveryCount        = "Invalid max delivery count %v (must be between %v and %v), please contact cherami team"
	strInvalidSkipOlderMessageSeconds = "Invalid skip older message seconds %v (must be zero or between %v and %v), please contact cherami team"
	strInvalidDelayMessageSeconds     = "Invalid delay message seconds %v (must be between %v and %v), please contact cherami team"
)
var uuidRegex, _ = regexp.Compile(`^[[:xdigit:]]{8}-[[:xdigit:]]{4}-[[:xdigit:]]{4}-[[:xdigit:]]{4}-[[:xdigit:]]{12}$`)
func ExitIfError(err error) {
	const stacksEnv = `CHERAMI_SHOW_STACKS`
	envReminder := "\n-env=staging is now the default. Did you mean '-env=prod' ?\n"
	if err != nil {
		fmt.Fprintln(os.Stderr, err.Error())
		fmt.Fprintln(os.Stderr, envReminder)
		if os.Getenv(stacksEnv) != `` {
			debug.PrintStack()
		} else {
			fmt.Fprintf(os.Stderr, "('export %s=1' to see stack traces)\n", stacksEnv)
		}
		os.Exit(1)
	}
}
func newGlobalOptionsFromCLIContext(c *cli.Context) *GlobalOptions {
	host, port, err := common.SplitHostPort(c.GlobalString("hostport"))
	ExitIfError(err)
	environment := c.GlobalString("env")
	if strings.HasPrefix(environment, `prod`) {
		environment = ``
	}
	return &GlobalOptions{
		hyperbahn:              c.GlobalBool("hyperbahn"),
		hyperbahnBootstrapFile: c.GlobalString("hyperbahn_bootstrap_file"),
		env:          environment,
		frontendHost: host,
		frontendPort: port,
		timeoutSecs:  c.GlobalInt("timeout"),
	}
}
func GetCClient(c *cli.Context, serviceName string) ccli.Client {
	return GetCClientSecure(c, serviceName, nil)
}
func GetCClientSecure(c *cli.Context, serviceName string, authProvider ccli.AuthProvider) ccli.Client {
	gOpts := newGlobalOptionsFromCLIContext(c)
	var cClient ccli.Client
	var err error
	cOpts := ccli.ClientOptions{
		Timeout:       time.Duration(gOpts.timeoutSecs) * time.Second,
		DeploymentStr: gOpts.env,
		AuthProvider:  authProvider,
	}
	if !(len(gOpts.frontendHost) > 0 || gOpts.frontendPort > 0) && gOpts.hyperbahn {
		cClient, err = ccli.NewHyperbahnClient(serviceName, gOpts.hyperbahnBootstrapFile, &cOpts)
	} else {
		cClient, err = ccli.NewClient(serviceName, gOpts.frontendHost, gOpts.frontendPort, &cOpts)
	}
	ExitIfError(err)
	return cClient
}
func GetMClient(c *cli.Context, serviceName string) mcli.Client {
	gOpts := newGlobalOptionsFromCLIContext(c)
	var mClient mcli.Client
	var err error
	cOpts := ccli.ClientOptions{
		Timeout:       time.Duration(gOpts.timeoutSecs) * time.Second,
		DeploymentStr: gOpts.env,
	}
	if !(len(gOpts.frontendHost) > 0 || gOpts.frontendPort > 0) && gOpts.hyperbahn {
		mClient, err = mcli.NewHyperbahnClient(serviceName, gOpts.hyperbahnBootstrapFile, &cOpts)
	} else {
		mClient, err = mcli.NewClient(serviceName, gOpts.frontendHost, gOpts.frontendPort, &cOpts)
	}
	ExitIfError(err)
	return mClient
}
func Jsonify(obj thrift.TStruct) string {
	transport := thrift.NewTMemoryBufferLen(1024)
	protocol := thrift.NewTSimpleJSONProtocol(transport)
	obj.Write(protocol)
	protocol.Flush()
	transport.Flush()
	return transport.String()
}
func getChecksumOptionParam(optionStr string) cherami.ChecksumOption {
	checksumstr := strings.ToLower(optionStr)
	if checksumstr == "md5" {
		return cherami.ChecksumOption_MD5
	}
	return cherami.ChecksumOption_CRC32IEEE
}
func CreateDestination(c *cli.Context, cliHelper common.CliHelper, serviceName string) {
	CreateDestinationSecure(c, cliHelper, serviceName, nil)
}
func CreateDestinationSecure(
	c *cli.Context,
	cliHelper common.CliHelper,
	serviceName string,
	authProvider ccli.AuthProvider,
) {
	cClient := GetCClientSecure(c, serviceName, authProvider)
	if len(c.Args()) < 1 {
		ExitIfError(errors.New(strNotEnoughArgs))
	}
	adminMode := c.GlobalBool("admin_mode")
	path := c.Args().First()
	kafkaCluster := c.String("kafka_cluster")
	kafkaTopics := c.StringSlice("kafka_topics")
	dType := cherami.DestinationType_PLAIN
	if c.String("type") == "timer" {
		dType = cherami.DestinationType_TIMER
	}
	if c.String("type") == "kafka" {  
		dType = cherami.DestinationType_KAFKA
		if !strings.HasPrefix(path, kafkaPrefix) {
			ExitIfError(errors.New(strKafkaNaming))
		}
		if kafkaCluster == `` || common.ContainsEmpty(kafkaTopics) || len(kafkaTopics) == 0 {  
			ExitIfError(errors.New(strKafkaNotEnoughArgs))
		}
	}
	consumedMessagesRetention := int32(c.Int("consumed_messages_retention"))
	if !adminMode && (consumedMessagesRetention < MinConsumedMessagesRetention || consumedMessagesRetention > MaxConsumedMessagesRetention) {
		ExitIfError(fmt.Errorf(strInvalidConsumedRetention, consumedMessagesRetention, MinConsumedMessagesRetention, MaxConsumedMessagesRetention))
	}
	unconsumedMessagesRetention := int32(c.Int("unconsumed_messages_retention"))
	if !adminMode && (unconsumedMessagesRetention < MinUnconsumedMessagesRetention || unconsumedMessagesRetention > MaxUnconsumedMessagesRetention) {
		ExitIfError(fmt.Errorf(strInvalidUnconsumedRetention, unconsumedMessagesRetention, MinUnconsumedMessagesRetention, MaxUnconsumedMessagesRetention))
	}
	checksumOption := getChecksumOptionParam(string(c.String("checksum_option")))
	ownerEmail := string(c.String("owner_email"))
	if len(ownerEmail) == 0 {
		cliHelper.GetDefaultOwnerEmail()
	}
	zoneConfigs := getDestZoneConfigs(c, cliHelper)
	if len(zoneConfigs.Configs) > 1 && unconsumedMessagesRetention < MinUnconsumedMessagesRetentionForMultiZoneDest {
		ExitIfError(errors.New(strUnconsumedRetentionTooSmall))
	}
	desc, err := cClient.CreateDestination(&cherami.CreateDestinationRequest{
		Path: &path,
		Type: &dType,
		ConsumedMessagesRetention:   &consumedMessagesRetention,
		UnconsumedMessagesRetention: &unconsumedMessagesRetention,
		ChecksumOption:              checksumOption,
		OwnerEmail:                  &ownerEmail,
		IsMultiZone:                 common.BoolPtr(len(zoneConfigs.Configs) > 0),
		ZoneConfigs:                 &zoneConfigs,
		KafkaCluster:                &kafkaCluster,
		KafkaTopics:                 kafkaTopics,
	})
	ExitIfError(err)
	fmt.Printf("%v\n", Jsonify(desc))
}
func getDestZoneConfigs(c *cli.Context, cliHelper common.CliHelper) cherami.DestinationZoneConfigs {
	var zoneConfigs cherami.DestinationZoneConfigs
	configs := c.StringSlice("zone_config")
	for _, config := range configs {
		parts := strings.Split(config, `,`)
		if len(parts) != 4 {
			ExitIfError(errors.New(strWrongDestZoneConfig))
		}
		zone, err := cliHelper.GetCanonicalZone(parts[0])
		ExitIfError(err)
		allowPublish, err := strconv.ParseBool(parts[1])
		ExitIfError(err)
		allowConsume, err := strconv.ParseBool(parts[2])
		ExitIfError(err)
		replicaCount, err := strconv.ParseInt(parts[3], 10, 0)
		ExitIfError(err)
		if replicaCount < 1 || replicaCount > 3 {
			ExitIfError(errors.New(strWrongReplicaCount))
		}
		zoneConfigs.Configs = append(zoneConfigs.Configs, &cherami.DestinationZoneConfig{
			Zone:                   common.StringPtr(zone),
			AllowPublish:           common.BoolPtr(allowPublish),
			AllowConsume:           common.BoolPtr(allowConsume),
			RemoteExtentReplicaNum: common.Int32Ptr(int32(replicaCount)),
		})
	}
	if len(zoneConfigs.Configs) == 1 {
		ExitIfError(errors.New(strWrongZoneConfigCount))
	}
	return zoneConfigs
}
func UpdateDestination(c *cli.Context, cliHelper common.CliHelper, serviceName string) {
	UpdateDestinationSecure(c, cliHelper, serviceName, nil)
}
func UpdateDestinationSecure(
	c *cli.Context,
	cliHelper common.CliHelper,
	serviceName string,
	authProvider ccli.AuthProvider,
) {
	mClient := GetMClient(c, serviceName)
	cClient := GetCClientSecure(c, serviceName, authProvider)
	if len(c.Args()) < 1 {
		ExitIfError(errors.New(strNotEnoughArgs))
	}
	path := c.Args().First()
	status := cherami.DestinationStatus_ENABLED
	switch c.String("status") {
	case "disabled":
		status = cherami.DestinationStatus_DISABLED
	case "sendonly":
		status = cherami.DestinationStatus_SENDONLY
	case "recvonly:":
		status = cherami.DestinationStatus_RECEIVEONLY
	case "enabled":
	default:
		if c.IsSet(`status`) {
			ExitIfError(errors.New(strDestStatus))
		}
	}
	setCount := 0
	request := &cherami.UpdateDestinationRequest{
		Path: &path,
		ConsumedMessagesRetention:   getIfSetInt32(c, `consumed_messages_retention`, &setCount),
		UnconsumedMessagesRetention: getIfSetInt32(c, `unconsumed_messages_retention`, &setCount),
		ZoneConfigs:                 getIfSetDestZoneConfig(c, &setCount, cliHelper),
	}
	existingDesc, err := readDestinationFromMetadata(mClient, path)
	ExitIfError(err)
	if existingDesc.GetIsMultiZone() || (request.IsSetZoneConfigs() && len(request.GetZoneConfigs().GetConfigs()) > 0) {
		if c.IsSet(`unconsumed_messages_retention`) && int32(c.Int(`unconsumed_messages_retention`)) < MinUnconsumedMessagesRetentionForMultiZoneDest {
			ExitIfError(errors.New(strUnconsumedRetentionTooSmall))
		}
		if !c.IsSet(`unconsumed_messages_retention`) && existingDesc.GetUnconsumedMessagesRetention() < MinUnconsumedMessagesRetentionForMultiZoneDest {
			ExitIfError(errors.New(strUnconsumedRetentionTooSmall))
		}
	}
	if c.IsSet(`status`) {
		setCount++
		request.Status = &status
	}
	if c.IsSet(`checksum_option`) {
		checksumOption := getChecksumOptionParam(string(c.String("checksum_option")))
		request.ChecksumOption = &checksumOption
		setCount++
	}
	if setCount == 0 {
		ExitIfError(errors.New(strNoChange))
	}
	desc, err := cClient.UpdateDestination(request)
	ExitIfError(err)
	fmt.Printf("%v\n", Jsonify(desc))
}