func (s *CassandraMetadataService) UpdateDestination(ctx thrift.Context, updateRequest *shared.UpdateDestinationRequest) (*shared.DestinationDescription, error) {
	if updateRequest.GetStatus() == shared.DestinationStatus_DELETED {
		return nil, &shared.BadRequestError{
			Message: "UpdateDestination: Update to DELETED status is not allowed. Use DeleteDestination instead.",
		}
	}
	getDestination := &shared.ReadDestinationRequest{
		DestinationUUID: common.StringPtr(updateRequest.GetDestinationUUID()),
	}
	existing, err := s.ReadDestination(nil, getDestination)
	if err != nil {
		return nil, err
	}
	if existing == nil {
		return nil, &shared.BadRequestError{
			Message: "Bad destination UUID",
		}
	}
	if !updateRequest.IsSetStatus() {
		updateRequest.Status = common.InternalDestinationStatusPtr(existing.GetStatus())
	}
	if !updateRequest.IsSetConsumedMessagesRetention() {
		updateRequest.ConsumedMessagesRetention = common.Int32Ptr(existing.GetConsumedMessagesRetention())
	}
	if !updateRequest.IsSetUnconsumedMessagesRetention() {
		updateRequest.UnconsumedMessagesRetention = common.Int32Ptr(existing.GetUnconsumedMessagesRetention())
	}
	if !updateRequest.IsSetOwnerEmail() {
		updateRequest.OwnerEmail = common.StringPtr(existing.GetOwnerEmail())
	}
	if !updateRequest.IsSetChecksumOption() {
		updateRequest.ChecksumOption = common.InternalChecksumOptionPtr(existing.GetChecksumOption())
	}
	isMultiZone := existing.GetIsMultiZone()
	if !updateRequest.IsSetZoneConfigs() {
		updateRequest.ZoneConfigs = existing.GetZoneConfigs()
	} else {
		isMultiZone = true
	}
	batch := s.session.NewBatch(gocql.LoggedBatch)  
	batch.Query(
		sqlUpdateDstByUUID,
		updateRequest.GetDestinationUUID(),
		existing.GetPath(),
		existing.GetType(),
		updateRequest.GetStatus(),
		updateRequest.GetConsumedMessagesRetention(),
		updateRequest.GetUnconsumedMessagesRetention(),
		updateRequest.GetOwnerEmail(),
		updateRequest.GetChecksumOption(),
		isMultiZone,
		marshalDstZoneConfigs(updateRequest.GetZoneConfigs()),
		isMultiZone,
		updateRequest.GetDestinationUUID())
	batch.Query(
		sqlUpdateDstByPath,
		updateRequest.GetDestinationUUID(),
		existing.GetPath(),
		existing.GetType(),
		updateRequest.GetStatus(),
		updateRequest.GetConsumedMessagesRetention(),
		updateRequest.GetUnconsumedMessagesRetention(),
		updateRequest.GetOwnerEmail(),
		updateRequest.GetChecksumOption(),
		isMultiZone,
		marshalDstZoneConfigs(updateRequest.GetZoneConfigs()),
		isMultiZone,
		existing.GetPath(),
		directoryUUID)
	batch.Cons = s.highConsLevel
	if err = s.session.ExecuteBatch(batch); err != nil {
		return nil, &shared.InternalServiceError{
			Message: "UpdateDestination: " + err.Error(),
		}
	}
	callerUserName := getThriftContextValue(ctx, common.CallerUserName)
	callerHostName := getThriftContextValue(ctx, common.CallerHostName)
	callerServiceName := getThriftContextValue(ctx, common.CallerServiceName)
	s.recordUserOperation(
		existing.GetPath(),
		updateRequest.GetDestinationUUID(),
		entityTypeDst,
		callerUserName,
		"",  
		callerServiceName,
		callerHostName,
		opsUpdate,
		time.Now(),
		marshalRequest(updateRequest))
	if updateRequest.IsSetStatus() {
		existing.Status = common.InternalDestinationStatusPtr(updateRequest.GetStatus())
	}
	if updateRequest.IsSetConsumedMessagesRetention() {
		existing.ConsumedMessagesRetention = common.Int32Ptr(updateRequest.GetConsumedMessagesRetention())
	}
	if updateRequest.IsSetUnconsumedMessagesRetention() {
		existing.UnconsumedMessagesRetention = common.Int32Ptr(updateRequest.GetUnconsumedMessagesRetention())
	}
	if updateRequest.IsSetOwnerEmail() {
		existing.OwnerEmail = common.StringPtr(updateRequest.GetOwnerEmail())
	}
	if updateRequest.IsSetChecksumOption() {
		existing.ChecksumOption = common.InternalChecksumOptionPtr(updateRequest.GetChecksumOption())
	}
	if updateRequest.IsSetZoneConfigs() {
		existing.ZoneConfigs = updateRequest.GetZoneConfigs()
	}
	existing.IsMultiZone = common.BoolPtr(isMultiZone)
	s.destinationCache.Delete("dstpath:" + existing.GetPath())
	s.destinationCache.Delete("dstuuid:" + updateRequest.GetDestinationUUID())
	return existing, nil
}
UpdateDestination implements the corresponding TChanMetadataServiceClient API
package metadata
import (
	"encoding/json"
	"fmt"
	"reflect"
	"regexp"
	"strings"
	"time"
	"github.com/uber/cherami-server/common"
	"github.com/uber/cherami-server/common/cache"
	"github.com/uber/cherami-server/common/configure"
	m "github.com/uber/cherami-thrift/.generated/go/metadata"
	"github.com/uber/cherami-thrift/.generated/go/shared"
	"github.com/gocql/gocql"
	"github.com/pborman/uuid"
	"github.com/sirupsen/logrus"
	"github.com/uber-common/bark"
	"github.com/uber/tchannel-go/thrift"
)
const directoryUUID string = "CC3B477C-E6F2-4465-9A98-7FE71B68CD1F"
var uuidRegex, _ = regexp.Compile(`^[[:xdigit:]]{8}-[[:xdigit:]]{4}-[[:xdigit:]]{4}-[[:xdigit:]]{4}-[[:xdigit:]]{12}$`)
const (
	defaultSessionTimeout         = 30 * time.Second
	defaultDLQConsumedRetention   = 7 * 24 * 3600  
	defaultDLQUnconsumedRetention = 7 * 24 * 3600  
)
const (
	opsCreate = "create"
	opsDelete = "delete"
	opsUpdate = "update"
)
const (
	entityTypeDst = "destinaton"
	entityTypeCG  = "consumer_group"
)
const (
	tableConsumerGroupExtents   = "consumer_group_extents"
	tableConsumerGroups         = "consumer_groups"
	tableConsumerGroupsByName   = "consumer_groups_by_name"
	tableDestinationExtents     = "destination_extents"
	tableDestinations           = "destinations"
	tableDestinationsByPath     = "destinations_by_path"
	tableHostAddrToUUID         = "host_addr_to_uuid"
	tableInputHostExtents       = "input_host_extents"
	tableOperationsByEntityName = "user_operations_by_entity_name"
	tableOperationsByEntityUUID = "user_operations_by_entity_uuid"
	tableStoreExtents           = "store_extents"
	tableUUIDToHostAddr         = "uuid_to_host_addr"
)
const (
	columnAckLevelOffset                 = "ack_level_offset"
	columnAckLevelSequence               = "ack_level_sequence"
	columnAckLevelSequenceRate           = "ack_level_sequence_rate"
	columnActiveZone                     = "active_zone"
	columnAllowConsume                   = "allow_consume"
	columnAllowPublish                   = "allow_publish"
	columnAlwaysReplicatedTo             = "always_replicate_to"
	columnArchivalLocation               = "archival_location"
	columnAvailableAddress               = "available_address"
	columnAvailableEnqueueTime           = "available_enqueue_time"
	columnAvailableSequence              = "available_sequence"
	columnAvailableSequenceRate          = "available_sequence_rate"
	columnBeginAddress                   = "begin_address"
	columnBeginEnqueueTime               = "begin_enqueue_time"
	columnBeginSequence                  = "begin_sequence"
	columnBeginTime                      = "begin_time"
	columnCallerHostName                 = "caller_host_name"
	columnCallerServiceName              = "caller_service_name"
	columnChecksumOption                 = "checksum_option"
	columnConnectedStore                 = "connected_store"
	columnConsumedMessagesRetention      = "consumed_messages_retention"
	columnConsumerGroup                  = "consumer_group"
	columnConsumerGroupUUID              = "consumer_group_uuid"
	columnConsumerGroupVisibility        = "consumer_group_visibility"
	columnCreatedTime                    = "created_time"
	columnDelaySeconds                   = "delay_seconds"
	columnDLQConsumerGroup               = "dlq_consumer_group"
	columnDLQMergeBefore                 = "dlq_merge_before"
	columnDLQPurgeBefore                 = "dlq_purge_before"
	columnDeadLetterQueueDestinationUUID = "dead_letter_queue_destination_uuid"
	columnDestination                    = "destination"
	columnDestinationUUID                = "destination_uuid"
	columnDirectoryUUID                  = "directory_uuid"
	columnEndTime                        = "end_time"
	columnEntityName                     = "entity_name"
	columnEntityType                     = "entity_type"
	columnEntityUUID                     = "entity_uuid"
	columnExtent                         = "extent"
	columnExtentUUID                     = "extent_uuid"
	columnHostAddr                       = "hostaddr"
	columnHostName                       = "hostname"
	columnInitiatorInfo                  = "initiator"
	columnInputHostUUID                  = "input_host_uuid"
	columnIsMultiZone                    = "is_multi_zone"
	columnKafkaCluster                   = "kafka_cluster"
	columnKafkaTopics                    = "kafka_topics"
	columnLastAddress                    = "last_address"
	columnLastEnqueueTime                = "last_enqueue_time"
	columnLastSequence                   = "last_sequence"
	columnLastSequenceRate               = "last_sequence_rate"
	columnLockTimeoutSeconds             = "lock_timeout_seconds"
	columnMaxDeliveryCount               = "max_delivery_count"
	columnName                           = "name"
	columnOpsContent                     = "operation_content"
	columnOpsTime                        = "operation_time"
	columnOpsType                        = "operation_type"
	columnOriginZone                     = "origin_zone"
	columnOutputHostUUID                 = "output_host_uuid"
	columnOwnerEmail                     = "owner_email"
	columnPath                           = "path"
	columnReceivedLevelOffset            = "received_level_offset"
	columnReceivedLevelSequence          = "received_level_sequence"
	columnReceivedLevelSequenceRate      = "received_level_sequence_rate"
	columnRemoteExtentPrimaryStore       = "remote_extent_primary_store"
	columnRemoteExtentReplicaNum         = "remote_extent_replica_num"
	columnReplicaStats                   = "replica_stats"
	columnReplicationStatus              = "replication_status"
	columnSizeInBytes                    = "size_in_bytes"
	columnSizeInBytesRate                = "size_in_bytes_rate"
	columnSkipOlderMessagesSeconds       = "skip_older_messages_seconds"
	columnStartFrom                      = "start_from"
	columnStatus                         = "status"
	columnStatusUpdatedTime              = "status_updated_time"
	columnStore                          = "store"
	columnStoreUUID                      = "store_uuid"
	columnStoreUUIDS                     = "store_uuids"
	columnStoreVersion                   = "store_version"
	columnType                           = "type"
	columnUUID                           = "uuid"
	columnUnconsumedMessagesRetention    = "unconsumed_messages_retention"
	columnUserEmail                      = "user_email"
	columnUserName                       = "user_name"
	columnVisible                        = "visible"
	columnZone                           = "zone"
	columnZoneConfigs                    = "zone_configs"
	columnOptions                        = "options"
)
const userOperationTTL = "2592000"  
const cassandraProtoVersion = 4
const deleteExtentTTLSeconds = int64(time.Hour*24) / int64(time.Second)
const defaultDeleteTTLSeconds = int64(time.Hour*24*30) / int64(time.Second)
const destinationCacheSize = 1048576
const consumerGroupCacheSize = 1048576
const cacheTTL = time.Second
type CassandraMetadataService struct {
	session       *gocql.Session
	lowConsLevel  gocql.Consistency
	midConsLevel  gocql.Consistency
	highConsLevel gocql.Consistency  
	clusterName   string
	log           bark.Logger
	destinationCache   cache.Cache
	consumerGroupCache cache.Cache
}
var _ m.TChanMetadataService = (*CassandraMetadataService)(nil)
func parseConsistency(cfgCons string) (lowCons gocql.Consistency, midCons gocql.Consistency, highCons gocql.Consistency) {
	lowCons, midCons, highCons = gocql.One, gocql.One, gocql.One
	switch cons := strings.Split(cfgCons, ","); len(cons) {
	case 3:
		lowCons = gocql.ParseConsistency(strings.TrimSpace(cons[2]))
		fallthrough
	case 2:
		midCons = gocql.ParseConsistency(strings.TrimSpace(cons[1]))
		if len(cons) == 2 {
			lowCons = midCons
		}
		fallthrough
	case 1:
		highCons = gocql.ParseConsistency(strings.TrimSpace(cons[0]))
	}
	return
}
func NewCassandraMetadataService(cfg configure.CommonMetadataConfig, log bark.Logger) (*CassandraMetadataService, error) {
	if log == nil {
		log = bark.NewLoggerFromLogrus(logrus.StandardLogger())
	}
	lowCons, midCons, highCons := parseConsistency(cfg.GetConsistency())
	if highCons == gocql.One && configure.NewCommonConfigure().GetEnvironment() == configure.EnvProduction {
		log.Panic("Highest consistency level of ONE should only be used in TestEnvironment")
	}
	clusterName := cfg.GetClusterName()
	if len(clusterName) < 1 {
		clusterName = "unknown"
	}
	cluster := newCluster(cfg.GetCassandraHosts())
	cluster.Port = cfg.GetPort()
	cluster.Keyspace = cfg.GetKeyspace()
	cluster.ProtoVersion = cassandraProtoVersion
	if auth := cfg.GetAuthentication(); auth.Enabled {
		cluster.Authenticator = gocql.PasswordAuthenticator{
			Username: auth.Username,
			Password: auth.Password,
		}
	}
	if filter, ok := cfg.GetDcFilter()[clusterName]; ok {
		cluster.HostFilter = gocql.DataCentreHostFilter(strings.TrimSpace(filter))
	}
	numConns := cfg.GetNumConns()
	if numConns > 0 {
		cluster.NumConns = numConns
	}
	cluster.Consistency = highCons
	cluster.Timeout = defaultSessionTimeout
	session, err := cluster.CreateSession()
	if err != nil {
		log.WithField(common.TagErr, err).Error("NewCassandraMetadataService failed")
		return nil, fmt.Errorf("createSession: %v", err)
	}
	return &CassandraMetadataService{
		session:       session,
		lowConsLevel:  lowCons,
		midConsLevel:  midCons,
		highConsLevel: highCons,
		clusterName:   clusterName,
		log:           log.WithField(common.TagModule, `metadata`),
		destinationCache:   cache.New(destinationCacheSize, &cache.Options{TTL: cacheTTL}),
		consumerGroupCache: cache.New(consumerGroupCacheSize, &cache.Options{TTL: cacheTTL}),
	}, nil
}
func (s *CassandraMetadataService) GetSession() *gocql.Session {
	return s.session
}
const (
	sqlUserInfoType = `{` +
		columnUserName + `: ?, ` +
		columnUserEmail + `: ?}`
	sqlRecordUserOperationByEntityName = `INSERT INTO ` + tableOperationsByEntityName +
		`(` + columnEntityName + `, ` + columnEntityUUID + `, ` + columnEntityType + `, ` + columnInitiatorInfo +
		`, ` + columnCallerServiceName + `, ` + columnCallerHostName + `, ` + columnOpsType + `, ` + columnOpsTime + `, ` + columnOpsContent + `) ` +
		` VALUES (?, ?, ?, ` + sqlUserInfoType + `, ?, ?, ?, ?, ?) USING TTL ` + userOperationTTL
	sqlRecordUserOperationByEntityUUID = `INSERT INTO ` + tableOperationsByEntityUUID +
		`(` + columnEntityName + `, ` + columnEntityUUID + `, ` + columnEntityType + `, ` + columnInitiatorInfo +
		`, ` + columnCallerServiceName + `, ` + columnCallerHostName + `, ` + columnOpsType + `, ` + columnOpsTime + `, ` + columnOpsContent + `) ` +
		` VALUES (?, ?, ?, ` + sqlUserInfoType + `, ?, ?, ?, ?, ?) USING TTL ` + userOperationTTL
)
func marshalRequest(request interface{}) (opsContentStr string) {
	opsContent, err := json.Marshal(request)
	opsContentStr = ""
	if err == nil {
		opsContentStr = string(opsContent)
	}
	return
}
func (s *CassandraMetadataService) recordUserOperation(entityName string, entityUUID string, entityType string, userName string, userEmail string, callerServiceName string, callerHostName string, operationType string, operationTime time.Time, operationContent string) error {
	batch := s.session.NewBatch(gocql.LoggedBatch)
	batch.Query(
		sqlRecordUserOperationByEntityName,
		entityName,
		entityUUID,
		entityType,
		userName,
		userEmail,
		callerServiceName,
		callerHostName,
		operationType,
		operationTime,
		operationContent)
	batch.Query(
		sqlRecordUserOperationByEntityUUID,
		entityName,
		entityUUID,
		entityType,
		userName,
		userEmail,
		callerServiceName,
		callerHostName,
		operationType,
		operationTime,
		operationContent)
	batch.Cons = s.highConsLevel
	if err := s.session.ExecuteBatch(batch); err != nil {
		s.log.WithFields(bark.Fields{common.TagErr: err}).Error("recordUserOperation failed")
		return fmt.Errorf("recordUserOperation error: %v", err)
	}
	return nil
}
const (
	sqlDstType = `{` +
		columnUUID + `: ?, ` +
		columnPath + `: ?, ` +
		columnType + `: ?, ` +
		columnStatus + `: ?, ` +
		columnConsumedMessagesRetention + `: ?, ` +
		columnUnconsumedMessagesRetention + `: ?, ` +
		columnOwnerEmail + `: ?, ` +
		columnChecksumOption + `: ?, ` +
		columnIsMultiZone + `: ?, ` +
		columnZoneConfigs + `: ?}`
	sqlInsertDstByUUID = `INSERT INTO ` + tableDestinations +
		`(` + columnUUID + `, ` + columnIsMultiZone + `, ` + columnDestination + `, ` + columnKafkaCluster + `, ` + columnKafkaTopics + `, ` + columnDLQConsumerGroup + `, ` + columnDLQPurgeBefore + `, ` + columnDLQMergeBefore + `) ` +
		` VALUES (?, ?, ` + sqlDstType + `, ?, ?, ?, ?, ?)`
	sqlInsertDstByPath = `INSERT INTO ` + tableDestinationsByPath +
		`(` +
		columnDirectoryUUID + `, ` +
		columnPath + `, ` +
		columnIsMultiZone + `, ` +
		columnDestination + `, ` +
		columnKafkaCluster + `, ` +
		columnKafkaTopics +
		`) VALUES (?, ?, ?, ` + sqlDstType + `, ?, ?) IF NOT EXISTS`
	sqlGetDstByPath = `SELECT ` +
		columnDestination + `.` + columnUUID + `, ` +
		columnPath + `, ` +
		columnDestination + `.` + columnType + `, ` +
		columnDestination + `.` + columnStatus + `, ` +
		columnDestination + `.` + columnConsumedMessagesRetention + `, ` +
		columnDestination + `.` + columnUnconsumedMessagesRetention + `, ` +
		columnDestination + `.` + columnOwnerEmail + `, ` +
		columnDestination + `.` + columnChecksumOption + `, ` +
		columnDestination + `.` + columnIsMultiZone + `, ` +
		columnDestination + `.` + columnZoneConfigs + `, ` +
		columnKafkaCluster + `, ` +
		columnKafkaTopics +
		` FROM ` + tableDestinationsByPath +
		` WHERE ` + columnDirectoryUUID + `=? and ` + columnPath + `=?`
	sqlListDestinationsByUUID = `SELECT ` +
		columnDestination + `.` + columnUUID + `, ` +
		columnDestination + `.` + columnPath + `, ` +
		columnDestination + `.` + columnType + `, ` +
		columnDestination + `.` + columnStatus + `, ` +
		columnDestination + `.` + columnConsumedMessagesRetention + `, ` +
		columnDestination + `.` + columnUnconsumedMessagesRetention + `, ` +
		columnDestination + `.` + columnOwnerEmail + `, ` +
		columnDestination + `.` + columnChecksumOption + `, ` +
		columnDestination + `.` + columnIsMultiZone + `, ` +
		columnDestination + `.` + columnZoneConfigs + `, ` +
		columnDLQConsumerGroup + `, ` +
		columnDLQPurgeBefore + `, ` +
		columnDLQMergeBefore + `, ` +
		columnKafkaCluster + `, ` +
		columnKafkaTopics +
		` FROM ` + tableDestinations
	sqlUpdateDstByUUID = `UPDATE ` + tableDestinations + ` SET ` +
		columnDestination + `=` + sqlDstType +
		`, ` + columnIsMultiZone + ` = ? ` +
		` WHERE ` + columnUUID + `=?`
	sqlUpdateDstDLQCursors = `UPDATE ` + tableDestinations + ` SET ` +
		columnDLQPurgeBefore + ` = ?,` +
		columnDLQMergeBefore + ` = ?` +
		` WHERE ` + columnUUID + `=?`
	sqlUpdateDstByPath = `UPDATE ` + tableDestinationsByPath + ` SET ` +
		columnDestination + `=` + sqlDstType +
		`, ` + columnIsMultiZone + ` = ? ` +
		` WHERE ` + columnPath + `=? and ` + columnDirectoryUUID + `=?`
	sqlDeleteDst = `DELETE FROM ` + tableDestinationsByPath +
		` WHERE ` + columnDirectoryUUID + `=? and ` + columnPath + `=?`
	sqlDeleteDstUUID = sqlInsertDstByUUID + ` USING TTL ?`
)
func (s *CassandraMetadataService) CreateDestination(ctx thrift.Context, request *shared.CreateDestinationRequest) (*shared.DestinationDescription, error) {
	uuidRequest := shared.NewCreateDestinationUUIDRequest()
	uuidRequest.Request = request
	uuidRequest.DestinationUUID = common.StringPtr(uuid.New())
	return s.CreateDestinationUUID(ctx, uuidRequest)
}
func (s *CassandraMetadataService) CreateDestinationUUID(ctx thrift.Context, uuidRequest *shared.CreateDestinationUUIDRequest) (*shared.DestinationDescription, error) {
	destinationUUID := uuidRequest.GetDestinationUUID()
	request := uuidRequest.GetRequest()
	if request.GetDLQConsumerGroupUUID() == `` {
		request.DLQConsumerGroupUUID = nil  
	}
	if err := s.session.Query(
		sqlInsertDstByUUID,
		destinationUUID,
		request.GetIsMultiZone(),
		destinationUUID,
		request.GetPath(),
		request.GetType(),
		shared.DestinationStatus_ENABLED,
		request.GetConsumedMessagesRetention(),
		request.GetUnconsumedMessagesRetention(),
		request.GetOwnerEmail(),
		request.GetChecksumOption(),
		request.GetIsMultiZone(),
		marshalDstZoneConfigs(request.GetZoneConfigs()),
		request.KafkaCluster,
		request.KafkaTopics,
		request.DLQConsumerGroupUUID,  
		int64(0), int64(0),            
	).Consistency(s.highConsLevel).Exec(); err != nil {
		return nil, &shared.InternalServiceError{
			Message: fmt.Sprintf("CreateDestination failure while inserting into destinations: %v", err),
		}
	}
	if !common.PathDLQRegex.MatchString(request.GetPath()) {
		previous := make(map[string]interface{})
		query := s.session.Query(
			sqlInsertDstByPath,
			directoryUUID,
			request.GetPath(),
			request.GetIsMultiZone(),
			destinationUUID,
			request.GetPath(),
			request.GetType(),
			shared.DestinationStatus_ENABLED,
			request.GetConsumedMessagesRetention(),
			request.GetUnconsumedMessagesRetention(),
			request.GetOwnerEmail(),
			request.GetChecksumOption(),
			request.GetIsMultiZone(),
			marshalDstZoneConfigs(request.GetZoneConfigs()),
			request.KafkaCluster,
			request.KafkaTopics).Consistency(s.highConsLevel)
		applied, err := query.MapScanCAS(previous)
		if err != nil {
			return nil, &shared.InternalServiceError{
				Message: fmt.Sprintf("CreateDestination failure while inserting into destinations_by_path: %v", err),
			}
		}
		if !applied {
			deleteOrphanRecord := `DELETE FROM ` + tableDestinations + ` WHERE ` + columnUUID + `=?`
			if err = s.session.Query(deleteOrphanRecord, destinationUUID).Exec(); err != nil {
				s.log.WithFields(bark.Fields{common.TagDst: common.FmtDst(destinationUUID), common.TagErr: err}).Warn(`CreateDestination failure deleting orphan record from destinations`)
			}
			return nil, &shared.EntityAlreadyExistsError{
				Message: fmt.Sprintf("CreateDestination: Destination \"%v\" already exists with id=%v",
					request.GetPath(), previous[columnDirectoryUUID]),
			}
		}
	}
	callerUserName := getThriftContextValue(ctx, common.CallerUserName)
	callerHostName := getThriftContextValue(ctx, common.CallerHostName)
	callerServiceName := getThriftContextValue(ctx, common.CallerServiceName)
	s.recordUserOperation(
		request.GetPath(),
		destinationUUID,
		entityTypeDst,
		callerUserName,
		"",  
		callerServiceName,
		callerHostName,
		opsCreate,
		time.Now(),
		marshalRequest(request))
	return &shared.DestinationDescription{
		DestinationUUID: common.StringPtr(destinationUUID),
		Path:            common.StringPtr(request.GetPath()),
		Type:            common.InternalDestinationTypePtr(request.GetType()),
		Status:          common.InternalDestinationStatusPtr(shared.DestinationStatus_ENABLED),
		ConsumedMessagesRetention:   common.Int32Ptr(request.GetConsumedMessagesRetention()),
		UnconsumedMessagesRetention: common.Int32Ptr(request.GetUnconsumedMessagesRetention()),
		OwnerEmail:                  common.StringPtr(request.GetOwnerEmail()),
		ChecksumOption:              common.InternalChecksumOptionPtr(request.GetChecksumOption()),
		IsMultiZone:                 common.BoolPtr(request.GetIsMultiZone()),
		ZoneConfigs:                 request.GetZoneConfigs(),
		DLQConsumerGroupUUID:        common.StringPtr(request.GetDLQConsumerGroupUUID()),
		KafkaCluster:                common.StringPtr(request.GetKafkaCluster()),
		KafkaTopics:                 request.KafkaTopics,
	}, nil
}
func getUtilDestinationDescription() *shared.DestinationDescription {
	result := &shared.DestinationDescription{}
	result.DestinationUUID = common.StringPtr("")
	result.Path = common.StringPtr("")
	result.Type = common.InternalDestinationTypePtr(0)
	result.Status = common.InternalDestinationStatusPtr(0)
	result.ConsumedMessagesRetention = common.Int32Ptr(0)
	result.UnconsumedMessagesRetention = common.Int32Ptr(0)
	result.OwnerEmail = common.StringPtr("")
	result.ChecksumOption = common.InternalChecksumOptionPtr(0)
	result.DLQConsumerGroupUUID = common.StringPtr("")
	result.DLQPurgeBefore = common.Int64Ptr(0)
	result.DLQMergeBefore = common.Int64Ptr(0)
	result.IsMultiZone = common.BoolPtr(false)
	result.ZoneConfigs = shared.DestinationDescription_ZoneConfigs_DEFAULT
	result.KafkaCluster = common.StringPtr("")
	result.KafkaTopics = make([]string, 0)
	return result
}
func getUtilConsumerGroupDescription() *shared.ConsumerGroupDescription {
	result := &shared.ConsumerGroupDescription{}
	result.ConsumerGroupUUID = common.StringPtr("")
	result.DestinationUUID = common.StringPtr("")
	result.ConsumerGroupName = common.StringPtr("")
	result.StartFrom = common.Int64Ptr(0)
	result.Status = common.InternalConsumerGroupStatusPtr(0)
	result.LockTimeoutSeconds = common.Int32Ptr(0)
	result.MaxDeliveryCount = common.Int32Ptr(0)
	result.SkipOlderMessagesSeconds = common.Int32Ptr(0)
	result.DelaySeconds = common.Int32Ptr(0)
	result.OwnerEmail = common.StringPtr("")
	result.IsMultiZone = common.BoolPtr(false)
	result.ActiveZone = common.StringPtr("")
	result.ZoneConfigs = shared.ConsumerGroupDescription_ZoneConfigs_DEFAULT
	result.Options = make(map[string]string, 0)
	return result
}
func getUtilEntityOpsDescription() *shared.EntityOpsDescription {
	result := &shared.EntityOpsDescription{}
	result.EntityUUID = common.StringPtr("")
	result.EntityName = common.StringPtr("")
	result.EntityType = common.StringPtr("")
	result.HostName = common.StringPtr("")
	result.ServiceName = common.StringPtr("")
	result.UserName = common.StringPtr("")
	result.OpsType = common.StringPtr("")
	result.OpsTime = common.StringPtr("")
	result.OpsContent = common.StringPtr("")
	return result
}
func getUtilHostDescription() *m.HostDescription {
	result := &m.HostDescription{}
	result.HostUUID = common.StringPtr("")
	result.HostAddr = common.StringPtr("")
	result.HostName = common.StringPtr("")
	return result
}
func marshalDstZoneConfigs(configs []*shared.DestinationZoneConfig) []map[string]interface{} {
	configsData := make([]map[string]interface{}, len(configs))
	for i, config := range configs {
		configsData[i] = map[string]interface{}{
			columnZone:                   config.GetZone(),
			columnAllowPublish:           config.GetAllowPublish(),
			columnAllowConsume:           config.GetAllowConsume(),
			columnAlwaysReplicatedTo:     config.GetAlwaysReplicateTo(),
			columnRemoteExtentReplicaNum: config.GetRemoteExtentReplicaNum(),
		}
	}
	return configsData
}
func unmarshalDstZoneConfigs(configsData []map[string]interface{}) []*shared.DestinationZoneConfig {
	configs := make([]*shared.DestinationZoneConfig, len(configsData))
	for i, configMap := range configsData {
		configs[i] = &shared.DestinationZoneConfig{
			Zone:                   common.StringPtr(toString(configMap[columnZone])),
			AllowPublish:           common.BoolPtr(toBool(configMap[columnAllowPublish])),
			AllowConsume:           common.BoolPtr(toBool(configMap[columnAllowConsume])),
			AlwaysReplicateTo:      common.BoolPtr(toBool(configMap[columnAlwaysReplicatedTo])),
			RemoteExtentReplicaNum: common.Int32Ptr(int32(toInt(configMap[columnRemoteExtentReplicaNum]))),
		}
	}
	return configs
}
func (s *CassandraMetadataService) ReadDestination(ctx thrift.Context, getRequest *shared.ReadDestinationRequest) (result *shared.DestinationDescription, err error) {
	var query *gocql.Query
	var uuid, sql string
	if getRequest.Path != nil {
		if getRequest.DestinationUUID != nil {
			return nil, &shared.BadRequestError{Message: "ReadDestination: both DestinationUUID and Path are specified"}
		}
		if uuidRegex.MatchString(getRequest.GetPath()) {
			getRequest.DestinationUUID = getRequest.Path
			getRequest.Path = nil
		}
	}
	var key string
	if getRequest.Path != nil {
		key = "dstpath:" + getRequest.GetPath()
	} else {
		key = "dstuuid:" + getRequest.GetDestinationUUID()
	}
	cached := s.destinationCache.Get(key)
	if cached != nil {
		return cached.(*shared.DestinationDescription), nil
	}
	result = getUtilDestinationDescription()
	var zoneConfigsData []map[string]interface{}
	if getRequest.Path != nil {
		query = s.session.Query(sqlGetDstByPath).Consistency(s.lowConsLevel)
		query.Bind(directoryUUID, getRequest.GetPath())
		err = query.Scan(
			result.DestinationUUID,
			result.Path,
			result.Type,
			result.Status,
			result.ConsumedMessagesRetention,
			result.UnconsumedMessagesRetention,
			result.OwnerEmail,
			result.ChecksumOption,
			result.IsMultiZone,
			&zoneConfigsData,
			result.KafkaCluster,
			&result.KafkaTopics,
		)
	} else {
		sql = sqlListDestinationsByUUID + ` WHERE ` + columnUUID + `=?`
		query = s.session.Query(sql).Consistency(s.lowConsLevel)
		uuid = getRequest.GetDestinationUUID()
		query.Bind(uuid)
		err = query.Scan(
			result.DestinationUUID,
			result.Path,
			result.Type,
			result.Status,
			result.ConsumedMessagesRetention,
			result.UnconsumedMessagesRetention,
			result.OwnerEmail,
			result.ChecksumOption,
			result.IsMultiZone,
			&zoneConfigsData,
			result.DLQConsumerGroupUUID,  
			result.DLQPurgeBefore,        
			result.DLQMergeBefore,        
			result.KafkaCluster,
			&result.KafkaTopics,
		)
	}
	result.ZoneConfigs = unmarshalDstZoneConfigs(zoneConfigsData)
	if err != nil {
		if err == gocql.ErrNotFound {
			var dest string
			if getRequest.Path != nil {
				dest = getRequest.GetPath()
			} else {
				dest = getRequest.GetDestinationUUID()
			}
			return nil, &shared.EntityNotExistsError{
				Message: fmt.Sprintf("Destination %s does not exist", dest),
			}
		}
		return nil, &shared.InternalServiceError{
			Message: err.Error(),
		}
	}
	*result.DLQPurgeBefore = int64(cqlTimestampToUnixNano(*result.DLQPurgeBefore))
	*result.DLQMergeBefore = int64(cqlTimestampToUnixNano(*result.DLQMergeBefore))
	s.destinationCache.Put(key, result)
	return result, nil
}