func (r *Replicator) CreateExtent(ctx thrift.Context, createRequest *shared.CreateExtentRequest) (*shared.CreateExtentResult_, error) {
	lcllg := r.logger.WithFields(bark.Fields{
		common.TagDst:      common.FmtDst(createRequest.GetExtent().GetDestinationUUID()),
		common.TagExt:      common.FmtExt(createRequest.GetExtent().GetExtentUUID()),
		common.TagZoneName: common.FmtZoneName(createRequest.GetExtent().GetOriginZone()),
	})
	r.m3Client.IncCounter(metrics.ReplicatorCreateExtentScope, metrics.ReplicatorRequests)
	controller, err := r.GetClientFactory().GetControllerClient()
	if err != nil {
		lcllg.WithField(common.TagErr, err).Error(`Error getting controller client`)
		r.m3Client.IncCounter(metrics.ReplicatorCreateExtentScope, metrics.ReplicatorFailures)
		return nil, err
	}
	res, err := controller.CreateRemoteZoneExtent(ctx, createRequest)
	if err != nil {
		lcllg.WithField(common.TagErr, err).Error(`Error calling controller to create extent`)
		r.m3Client.IncCounter(metrics.ReplicatorCreateExtentScope, metrics.ReplicatorFailures)
		return nil, err
	}
	lcllg.Info(`Called controller to create extent`)
	return res, nil
}
CreateExtent create extent at local zone, expect to be called by remote replicator
package replicator
import (
	"fmt"
	"math/rand"
	"net"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"
	"github.com/uber-common/bark"
	"github.com/uber/tchannel-go/thrift"
	ccommon "github.com/uber/cherami-client-go/common"
	"github.com/uber/cherami-server/common"
	"github.com/uber/cherami-server/common/configure"
	dconfig "github.com/uber/cherami-server/common/dconfigclient"
	mm "github.com/uber/cherami-server/common/metadata"
	"github.com/uber/cherami-server/common/metrics"
	"github.com/uber/cherami-server/common/set"
	storeStream "github.com/uber/cherami-server/stream"
	"github.com/uber/cherami-thrift/.generated/go/admin"
	"github.com/uber/cherami-thrift/.generated/go/metadata"
	rgen "github.com/uber/cherami-thrift/.generated/go/replicator"
	"github.com/uber/cherami-thrift/.generated/go/shared"
)
type (
	Replicator struct {
		logger   bark.Logger
		m3Client metrics.Client
		common.SCommon
		hostIDHeartbeater        common.HostIDHeartbeater
		AppConfig                configure.CommonAppConfig
		uconfigClient            dconfig.Client
		metaClient               metadata.TChanMetadataService
		allZones                 map[string][]string
		localZone                string
		authoritativeZone        string
		tenancy                  string
		defaultAuthoritativeZone string
		replicatorclientFactory  ClientFactory
		remoteReplicatorConn      map[string]*outConnection
		remoteReplicatorConnMutex sync.RWMutex
		storehostConn             map[string]*outConnection
		storehostConnMutex        sync.RWMutex
		knownCgExtents set.Set
		metadataReconciler MetadataReconciler
	}
)
const (
	remoteReplicatorCallTimeOut = 30 * time.Second
	remoteReplicatorSetAcklevelCallTimeOut = 5 * time.Second
	localReplicatorCallTimeOut = 30 * time.Second
)
var _ rgen.TChanReplicator = (*Replicator)(nil)
func getAllZones(replicatorHosts map[string][]string) map[string][]string {
	allZones := make(map[string][]string)
	for zone := range replicatorHosts {
		zone = strings.ToLower(zone)
		tenancy := common.TenancyProd
		parts := strings.Split(zone, "_")
		if len(parts) == 2 {
			tenancy = parts[0]
			zone = parts[1]
		}
		if _, ok := allZones[tenancy]; !ok {
			allZones[tenancy] = make([]string, 0)
		}
		allZones[tenancy] = append(allZones[tenancy], zone)
	}
	return allZones
}
func NewReplicator(serviceName string, sVice common.SCommon, metadataClient metadata.TChanMetadataService, replicatorClientFactory ClientFactory, config configure.CommonAppConfig) (*Replicator, []thrift.TChanServer) {
	deployment := strings.ToLower(sVice.GetConfig().GetDeploymentName())
	zone, tenancy := common.GetLocalClusterInfo(deployment)
	allZones := getAllZones(replicatorClientFactory.GetHostsForAllDeployment())
	logger := (sVice.GetConfig().GetLogger()).WithFields(bark.Fields{
		common.TagReplicator: common.FmtOut(sVice.GetHostUUID()),
		common.TagDplName:    common.FmtDplName(deployment),
		common.TagZoneName:   common.FmtZoneName(zone),
		common.TagTenancy:    common.FmtTenancy(tenancy),
	})
	r := &Replicator{
		logger:                   logger,
		m3Client:                 metrics.NewClient(sVice.GetMetricsReporter(), metrics.Replicator),
		SCommon:                  sVice,
		AppConfig:                config,
		allZones:                 allZones,
		localZone:                zone,
		defaultAuthoritativeZone: config.GetReplicatorConfig().GetDefaultAuthoritativeZone(),
		tenancy:                  tenancy,
		replicatorclientFactory:  replicatorClientFactory,
		remoteReplicatorConn:     make(map[string]*outConnection),
		storehostConn:            make(map[string]*outConnection),
		knownCgExtents:           set.NewConcurrent(0),
	}
	r.metaClient = mm.NewMetadataMetricsMgr(metadataClient, r.m3Client, r.logger)
	r.uconfigClient = sVice.GetDConfigClient()
	r.dynamicConfigManage()
	return r, []thrift.TChanServer{rgen.NewTChanReplicatorServer(r), admin.NewTChanReplicatorAdminServer(r)}
}
func (r *Replicator) Start(thriftService []thrift.TChanServer) {
	r.SCommon.Start(thriftService)
	r.hostIDHeartbeater = common.NewHostIDHeartbeater(r.metaClient, r.GetHostUUID(), r.GetHostPort(), r.GetHostName(), r.logger)
	r.hostIDHeartbeater.Start()
	r.replicatorclientFactory.SetTChannel(r.GetTChannel())
	r.metadataReconciler = NewMetadataReconciler(r.metaClient, r, r.localZone, r.logger, r.m3Client)
	r.metadataReconciler.Start()
}
func (r *Replicator) Stop() {
	r.hostIDHeartbeater.Stop()
	for _, conn := range r.remoteReplicatorConn {
		conn.shutdown()
	}
	for _, conn := range r.storehostConn {
		conn.shutdown()
	}
	r.metadataReconciler.Stop()
	r.SCommon.Stop()
}
func (r *Replicator) RegisterWSHandler() *http.ServeMux {
	mux := http.NewServeMux()
	mux.HandleFunc(fmt.Sprintf(ccommon.HTTPHandlerPattern, ccommon.EndpointOpenReplicationRemoteReadStream), r.OpenReplicationRemoteReadStreamHandler)
	mux.HandleFunc(fmt.Sprintf(ccommon.HTTPHandlerPattern, ccommon.EndpointOpenReplicationReadStream), r.OpenReplicationReadStreamHandler)
	return mux
}
func (r *Replicator) getAuthoritativeZone() string {
	return r.authoritativeZone
}
func (r *Replicator) setAuthoritativeZone(zone string) {
	r.authoritativeZone = zone
}
func (r *Replicator) OpenReplicationReadStreamHandler(w http.ResponseWriter, req *http.Request) {
	r.m3Client.IncCounter(metrics.OpenReplicationReadScope, metrics.ReplicatorRequests)
	request, err := common.GetOpenReplicationReadStreamRequestHTTP(req.Header)
	if err != nil {
		r.logger.WithField(common.TagErr, err).Error("unable to parse all needed headers")
		r.m3Client.IncCounter(metrics.OpenReplicationReadScope, metrics.ReplicatorBadRequest)
		r.m3Client.IncCounter(metrics.OpenReplicationReadScope, metrics.ReplicatorFailures)
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	inStream, err := r.GetWSConnector().AcceptReplicationReadStream(w, req)
	if err != nil {
		r.logger.WithField(common.TagErr, err).Error("unable to upgrade websocket connection")
		r.m3Client.IncCounter(metrics.OpenReplicationReadScope, metrics.ReplicatorCreateInStreamFailure)
		r.m3Client.IncCounter(metrics.OpenReplicationReadScope, metrics.ReplicatorFailures)
		return
	}
	r.logger.WithFields(bark.Fields{
		common.TagExt:  common.FmtExt(*request.OpenReadStreamRequest.ExtentUUID),
		common.TagDst:  common.FmtDst(*request.OpenReadStreamRequest.DestinationUUID),
		common.TagAddr: common.FmtAddr(request.OpenReadStreamRequest.GetAddress()),
	}).Info(`Received OpenReplicationRead request`)
	destUUID := *request.DestinationUUID
	extUUID := *request.ExtentUUID
	outStream, err := r.createStoreHostReadStream(destUUID, extUUID, request)
	if err != nil {
		r.logger.WithFields(bark.Fields{
			common.TagErr: err,
			common.TagExt: common.FmtExt(*request.OpenReadStreamRequest.ExtentUUID),
			common.TagDst: common.FmtDst(*request.OpenReadStreamRequest.DestinationUUID),
		}).Error("Can't open store host read stream")
		r.m3Client.IncCounter(metrics.OpenReplicationReadScope, metrics.ReplicatorCreateOutStreamFailure)
		r.m3Client.IncCounter(metrics.OpenReplicationReadScope, metrics.ReplicatorFailures)
		inStream.Done()
		return
	}
	destDesc, err := r.metaClient.ReadDestination(nil, &shared.ReadDestinationRequest{
		DestinationUUID: common.StringPtr(destUUID),
	})
	if err != nil {
		r.logger.WithFields(bark.Fields{
			common.TagErr: err,
			common.TagExt: common.FmtExt(*request.OpenReadStreamRequest.ExtentUUID),
			common.TagDst: common.FmtDst(*request.OpenReadStreamRequest.DestinationUUID),
		}).Error("Failed to read destination")
		r.m3Client.IncCounter(metrics.OpenReplicationReadScope, metrics.ReplicatorFailures)
		inStream.Done()
		return
	}
	outConn := newOutConnection(extUUID, destDesc.GetPath(), outStream, r.logger, r.m3Client, metrics.OpenReplicationReadScope)
	outConn.open()
	r.addStoreHostConn(outConn)
	defer r.removeStoreHostConn(outConn)
	inConn := newInConnection(extUUID, destDesc.GetPath(), inStream, outConn.msgsCh, r.logger, r.m3Client, metrics.OpenReplicationReadScope, metrics.OpenReplicationReadPerDestScope)
	inConn.open()
	outConn.WaitUntilDone()
	inConn.WaitUntilDone()
	return
}
func (r *Replicator) OpenReplicationRemoteReadStreamHandler(w http.ResponseWriter, req *http.Request) {
	r.m3Client.IncCounter(metrics.OpenReplicationRemoteReadScope, metrics.ReplicatorRequests)
	request, err := common.GetOpenReplicationRemoteReadStreamRequestHTTP(req.Header)
	if err != nil {
		r.logger.WithField(common.TagErr, err).Error("unable to parse all needed headers")
		r.m3Client.IncCounter(metrics.OpenReplicationRemoteReadScope, metrics.ReplicatorBadRequest)
		r.m3Client.IncCounter(metrics.OpenReplicationRemoteReadScope, metrics.ReplicatorFailures)
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	inStream, err := r.GetWSConnector().AcceptReplicationRemoteReadStream(w, req)
	if err != nil {
		r.logger.WithField(common.TagErr, err).Error("unable to upgrade websocket connection")
		r.m3Client.IncCounter(metrics.OpenReplicationRemoteReadScope, metrics.ReplicatorCreateInStreamFailure)
		r.m3Client.IncCounter(metrics.OpenReplicationRemoteReadScope, metrics.ReplicatorFailures)
		return
	}
	r.logger.WithFields(bark.Fields{
		common.TagExt:  common.FmtExt(*request.OpenReadStreamRequest.ExtentUUID),
		common.TagDst:  common.FmtDst(*request.OpenReadStreamRequest.DestinationUUID),
		common.TagAddr: common.FmtAddr(request.OpenReadStreamRequest.GetAddress()),
	}).Info(`Received OpenReplicationRemoteRead request`)
	extUUID := request.GetExtentUUID()
	destUUID := request.GetDestinationUUID()
	outStream, err := r.createRemoteReplicationReadStream(extUUID, destUUID, request)
	if err != nil {
		r.logger.WithFields(bark.Fields{
			common.TagErr: err,
			common.TagExt: common.FmtExt(*request.OpenReadStreamRequest.ExtentUUID),
			common.TagDst: common.FmtDst(*request.OpenReadStreamRequest.DestinationUUID),
		}).Error("Can't open remote replication read stream")
		r.m3Client.IncCounter(metrics.OpenReplicationRemoteReadScope, metrics.ReplicatorCreateOutStreamFailure)
		r.m3Client.IncCounter(metrics.OpenReplicationRemoteReadScope, metrics.ReplicatorFailures)
		inStream.Done()
		return
	}
	destDesc, err := r.metaClient.ReadDestination(nil, &shared.ReadDestinationRequest{
		DestinationUUID: common.StringPtr(destUUID),
	})
	if err != nil {
		r.logger.WithFields(bark.Fields{
			common.TagErr: err,
			common.TagExt: common.FmtExt(*request.OpenReadStreamRequest.ExtentUUID),
			common.TagDst: common.FmtDst(*request.OpenReadStreamRequest.DestinationUUID),
		}).Error("Failed to read destination")
		r.m3Client.IncCounter(metrics.OpenReplicationRemoteReadScope, metrics.ReplicatorFailures)
		inStream.Done()
		return
	}
	outConn := newOutConnection(extUUID, destDesc.GetPath(), outStream, r.logger, r.m3Client, metrics.OpenReplicationRemoteReadScope)
	outConn.open()
	r.addRemoteReplicatorConn(outConn)
	defer r.removeRemoteReplicatorConn(outConn)
	inConn := newInConnection(extUUID, destDesc.GetPath(), inStream, outConn.msgsCh, r.logger, r.m3Client, metrics.OpenReplicationRemoteReadScope, metrics.OpenReplicationRemoteReadPerDestScope)
	inConn.open()
	outConn.WaitUntilDone()
	inConn.WaitUntilDone()
	return
}
func (r *Replicator) CreateDestinationUUID(ctx thrift.Context, createRequest *shared.CreateDestinationUUIDRequest) (*shared.DestinationDescription, error) {
	r.m3Client.IncCounter(metrics.ReplicatorCreateDestUUIDScope, metrics.ReplicatorRequests)
	destDesc, err := r.metaClient.CreateDestinationUUID(ctx, createRequest)
	if err != nil {
		r.logger.WithFields(bark.Fields{
			common.TagDst:    common.FmtDst(createRequest.GetDestinationUUID()),
			common.TagDstPth: common.FmtDstPth(createRequest.GetRequest().GetPath()),
			common.TagErr:    err,
		}).Error(`Error creating destination`)
		r.m3Client.IncCounter(metrics.ReplicatorCreateDestUUIDScope, metrics.ReplicatorFailures)
		return nil, err
	}
	r.logger.WithFields(bark.Fields{
		common.TagDst:                 common.FmtDst(destDesc.GetDestinationUUID()),
		common.TagDstPth:              common.FmtDstPth(destDesc.GetPath()),
		`Type`:                        destDesc.GetType(),
		`Status`:                      destDesc.GetStatus(),
		`ConsumedMessagesRetention`:   destDesc.GetConsumedMessagesRetention(),
		`UnconsumedMessagesRetention`: destDesc.GetUnconsumedMessagesRetention(),
		`OwnerEmail`:                  destDesc.GetOwnerEmail(),
		`ChecksumOption`:              destDesc.GetChecksumOption(),
		`IsMultiZone`:                 destDesc.GetIsMultiZone(),  
	}).Info(`Created Destination`)
	return destDesc, nil
}
func (r *Replicator) CreateRemoteDestinationUUID(ctx thrift.Context, createRequest *shared.CreateDestinationUUIDRequest) error {
	r.m3Client.IncCounter(metrics.ReplicatorCreateRmtDestUUIDScope, metrics.ReplicatorRequests)
	if createRequest == nil || !createRequest.IsSetRequest() || !createRequest.IsSetDestinationUUID() {
		r.m3Client.IncCounter(metrics.ReplicatorCreateRmtDestUUIDScope, metrics.ReplicatorFailures)
		err := &shared.BadRequestError{Message: `Create remote destination request has nil request or nil uuid`}
		r.logger.WithField(common.TagErr, err).Error(`Create remote destination request verification failed`)
		return err
	}
	lclLg := r.logger.WithFields(bark.Fields{
		common.TagDst:    common.FmtDst(createRequest.GetDestinationUUID()),
		common.TagDstPth: common.FmtDstPth(createRequest.GetRequest().GetPath()),
	})
	if !createRequest.GetRequest().GetIsMultiZone() || !createRequest.GetRequest().IsSetZoneConfigs() || len(createRequest.GetRequest().GetZoneConfigs()) == 0 {
		r.m3Client.IncCounter(metrics.ReplicatorCreateRmtDestUUIDScope, metrics.ReplicatorFailures)
		err := &shared.BadRequestError{Message: fmt.Sprintf(`Not a valid create remote dest request for IsMultiZone [%v] or ZoneConfigs not set`, createRequest.GetRequest().GetIsMultiZone())}
		lclLg.WithField(common.TagErr, err).Error(`Create remote destination request verification failed`)
		return err
	}
	if _, ok := r.allZones[r.tenancy]; !ok {
		r.m3Client.IncCounter(metrics.ReplicatorCreateRmtDestUUIDScope, metrics.ReplicatorFailures)
		err := &shared.BadRequestError{Message: fmt.Sprintf(`Unknown tenancy [%s]`, r.tenancy)}
		lclLg.WithField(common.TagErr, err).Error(`Create remote destination failed with unknown tenancy`)
		return err
	}
	for _, zone := range r.allZones[r.tenancy] {
		if strings.EqualFold(zone, r.localZone) {
			continue
		}
		go r.createDestinationRemoteCall(zone, lclLg, createRequest)
	}
	return nil
}
func (r *Replicator) createDestinationRemoteCall(zone string, logger bark.Logger, createRequest *shared.CreateDestinationUUIDRequest) error {
	client, err := r.replicatorclientFactory.GetReplicatorClient(zone)
	if err != nil {
		r.m3Client.IncCounter(metrics.ReplicatorCreateRmtDestUUIDScope, metrics.ReplicatorFailures)
		logger.WithFields(bark.Fields{
			common.TagErr:      err,
			common.TagZoneName: common.FmtZoneName(zone),
		}).Error(`Get remote replicator client failed`)
		return err
	}
	ctx, cancel := thrift.NewContext(remoteReplicatorCallTimeOut)
	defer cancel()
	_, err = client.CreateDestinationUUID(ctx, createRequest)
	if err != nil {
		r.m3Client.IncCounter(metrics.ReplicatorCreateRmtDestUUIDScope, metrics.ReplicatorFailures)
		logger.WithFields(bark.Fields{
			common.TagErr:      err,
			common.TagZoneName: common.FmtZoneName(zone),
		}).Error(`Create remote destination call failed`)
		return err
	}
	return nil
}
func (r *Replicator) UpdateDestination(ctx thrift.Context, updateRequest *shared.UpdateDestinationRequest) (*shared.DestinationDescription, error) {
	r.m3Client.IncCounter(metrics.ReplicatorUpdateDestScope, metrics.ReplicatorRequests)
	destDesc, err := r.metaClient.UpdateDestination(ctx, updateRequest)
	if err != nil {
		r.logger.WithFields(bark.Fields{
			common.TagDst: common.FmtDst(updateRequest.GetDestinationUUID()),
			common.TagErr: err,
		}).Error(`Error updating destination`)
		r.m3Client.IncCounter(metrics.ReplicatorUpdateDestScope, metrics.ReplicatorFailures)
		return nil, err
	}
	r.logger.WithFields(bark.Fields{
		common.TagDst:                 common.FmtDst(updateRequest.GetDestinationUUID()),
		`Type`:                        destDesc.GetType(),
		`Status`:                      destDesc.GetStatus(),
		`ConsumedMessagesRetention`:   destDesc.GetConsumedMessagesRetention(),
		`UnconsumedMessagesRetention`: destDesc.GetUnconsumedMessagesRetention(),
		`OwnerEmail`:                  destDesc.GetOwnerEmail(),
		`ChecksumOption`:              destDesc.GetChecksumOption(),
		`IsMultiZone`:                 destDesc.GetIsMultiZone(),  
	}).Info(`Updated destination`)
	return destDesc, nil
}
func (r *Replicator) UpdateRemoteDestination(ctx thrift.Context, updateRequest *shared.UpdateDestinationRequest) error {
	r.m3Client.IncCounter(metrics.ReplicatorUpdateRmtDestScope, metrics.ReplicatorRequests)
	if updateRequest == nil {
		r.m3Client.IncCounter(metrics.ReplicatorUpdateRmtDestScope, metrics.ReplicatorFailures)
		err := &shared.BadRequestError{Message: `Update remote destination request has nil request`}
		r.logger.WithField(common.TagErr, err).Error(`Update remote destination request verification failed`)
		return err
	}
	lclLg := r.logger.WithFields(bark.Fields{
		common.TagDst: common.FmtDst(updateRequest.GetDestinationUUID()),
	})
	if _, ok := r.allZones[r.tenancy]; !ok {
		r.m3Client.IncCounter(metrics.ReplicatorUpdateRmtDestScope, metrics.ReplicatorFailures)
		err := &shared.BadRequestError{Message: fmt.Sprintf(`Unknown tenancy [%s]`, r.tenancy)}
		lclLg.WithField(common.TagErr, err).Error(`Update remote destination failed with unknown tenancy`)
		return err
	}
	for _, zone := range r.allZones[r.tenancy] {
		if strings.EqualFold(zone, r.localZone) {
			continue
		}
		go r.updateDestinationRemoteCall(zone, lclLg, updateRequest)
	}
	return nil
}
func (r *Replicator) updateDestinationRemoteCall(zone string, logger bark.Logger, updateRequest *shared.UpdateDestinationRequest) error {
	client, err := r.replicatorclientFactory.GetReplicatorClient(zone)
	if err != nil {
		r.m3Client.IncCounter(metrics.ReplicatorUpdateRmtDestScope, metrics.ReplicatorFailures)
		logger.WithFields(bark.Fields{
			common.TagErr:      err,
			common.TagZoneName: common.FmtZoneName(zone),
		}).Error(`Get remote replicator client failed`)
		return err
	}
	ctx, cancel := thrift.NewContext(remoteReplicatorCallTimeOut)
	defer cancel()
	_, err = client.UpdateDestination(ctx, updateRequest)
	if err != nil {
		r.m3Client.IncCounter(metrics.ReplicatorUpdateRmtDestScope, metrics.ReplicatorFailures)
		logger.WithFields(bark.Fields{
			common.TagErr:      err,
			common.TagZoneName: common.FmtZoneName(zone),
		}).Error(`Update remote destination call failed`)
		return err
	}
	return nil
}
func (r *Replicator) DeleteDestination(ctx thrift.Context, deleteRequest *shared.DeleteDestinationRequest) error {
	r.m3Client.IncCounter(metrics.ReplicatorDeleteDestScope, metrics.ReplicatorRequests)
	err := r.metaClient.DeleteDestination(ctx, deleteRequest)
	if err != nil {
		r.logger.WithFields(bark.Fields{
			common.TagDstPth: common.FmtDstPth(deleteRequest.GetPath()),
			common.TagErr:    err,
		}).Error(`Error deleting destination`)
		r.m3Client.IncCounter(metrics.ReplicatorDeleteDestScope, metrics.ReplicatorFailures)
		return err
	}
	return nil
}
func (r *Replicator) DeleteRemoteDestination(ctx thrift.Context, deleteRequest *shared.DeleteDestinationRequest) error {
	r.m3Client.IncCounter(metrics.ReplicatorDeleteRmtDestScope, metrics.ReplicatorRequests)
	if deleteRequest == nil {
		r.m3Client.IncCounter(metrics.ReplicatorDeleteRmtDestScope, metrics.ReplicatorFailures)
		err := &shared.BadRequestError{Message: `Delete remote destination request has nil request`}
		r.logger.WithField(common.TagErr, err).Error(`Delete remote destination request verification failed`)
		return err
	}
	lclLg := r.logger.WithFields(bark.Fields{
		common.TagDstPth: common.FmtDstPth(deleteRequest.GetPath()),
	})
	if _, ok := r.allZones[r.tenancy]; !ok {
		r.m3Client.IncCounter(metrics.ReplicatorDeleteRmtDestScope, metrics.ReplicatorFailures)
		err := &shared.BadRequestError{Message: fmt.Sprintf(`Unknown tenancy [%s]`, r.tenancy)}
		lclLg.WithField(common.TagErr, err).Error(`Update remote destination failed with unknown tenancy`)
		return err
	}
	for _, zone := range r.allZones[r.tenancy] {
		if strings.EqualFold(zone, r.localZone) {
			continue
		}
		go r.deleteDestinationRemoteCall(zone, lclLg, deleteRequest)
	}
	return nil
}
func (r *Replicator) deleteDestinationRemoteCall(zone string, logger bark.Logger, deleteRequest *shared.DeleteDestinationRequest) error {
	client, err := r.replicatorclientFactory.GetReplicatorClient(zone)
	if err != nil {
		r.m3Client.IncCounter(metrics.ReplicatorDeleteRmtDestScope, metrics.ReplicatorFailures)
		logger.WithFields(bark.Fields{
			common.TagErr:      err,
			common.TagZoneName: common.FmtZoneName(zone),
		}).Error(`Get remote replicator client failed`)
		return err
	}
	ctx, cancel := thrift.NewContext(remoteReplicatorCallTimeOut)
	defer cancel()
	err = client.DeleteDestination(ctx, deleteRequest)
	if err != nil {
		r.m3Client.IncCounter(metrics.ReplicatorDeleteRmtDestScope, metrics.ReplicatorFailures)
		logger.WithFields(bark.Fields{
			common.TagErr:      err,
			common.TagZoneName: common.FmtZoneName(zone),
		}).Error(`Delete remote destination call failed`)
		return err
	}
	return nil
}
func (r *Replicator) CreateConsumerGroupUUID(ctx thrift.Context, createRequest *shared.CreateConsumerGroupUUIDRequest) (*shared.ConsumerGroupDescription, error) {
	r.m3Client.IncCounter(metrics.ReplicatorCreateCgUUIDScope, metrics.ReplicatorRequests)
	cgDesc, err := r.metaClient.CreateConsumerGroupUUID(ctx, createRequest)
	if err != nil {
		r.logger.WithFields(bark.Fields{
			common.TagCnsm:   common.FmtCnsm(createRequest.GetConsumerGroupUUID()),
			common.TagCnsPth: common.FmtCnsPth(createRequest.GetRequest().GetConsumerGroupName()),
			common.TagDstPth: common.FmtDstPth(createRequest.GetRequest().GetDestinationPath()),
			common.TagErr:    err,
		}).Error(`Error creating consumer group`)
		r.m3Client.IncCounter(metrics.ReplicatorCreateCgUUIDScope, metrics.ReplicatorFailures)
		return nil, err
	}
	r.logger.WithFields(bark.Fields{
		common.TagCnsm:   common.FmtCnsm(createRequest.GetConsumerGroupUUID()),
		common.TagCnsPth: common.FmtCnsPth(createRequest.GetRequest().GetConsumerGroupName()),
		common.TagDstPth: common.FmtDstPth(createRequest.GetRequest().GetDestinationPath()),
		common.TagDLQID:  common.FmtDLQID(cgDesc.GetDeadLetterQueueDestinationUUID()),
		`IsMultiZone`:    cgDesc.GetIsMultiZone(),
		`ActiveZone`:     cgDesc.GetActiveZone(),
	}).Info(`Created consumer group`)
	return cgDesc, nil
}
func (r *Replicator) CreateRemoteConsumerGroupUUID(ctx thrift.Context, createRequest *shared.CreateConsumerGroupUUIDRequest) error {
	r.m3Client.IncCounter(metrics.ReplicatorCreateRmtCgUUIDScope, metrics.ReplicatorRequests)
	if createRequest == nil || !createRequest.IsSetRequest() || !createRequest.IsSetConsumerGroupUUID() {
		r.m3Client.IncCounter(metrics.ReplicatorCreateRmtCgUUIDScope, metrics.ReplicatorFailures)
		err := &shared.BadRequestError{Message: `Create remote cg request has nil request or nil uuid`}
		r.logger.WithField(common.TagErr, err).Error(`Create remote cg request verification failed`)
		return err
	}
	lclLg := r.logger.WithFields(bark.Fields{
		common.TagCnsm:   common.FmtCnsm(createRequest.GetConsumerGroupUUID()),
		common.TagDstPth: common.FmtDstPth(createRequest.GetRequest().GetDestinationPath()),
		common.TagCnsPth: common.FmtCnsPth(createRequest.GetRequest().GetConsumerGroupName()),
	})
	if !createRequest.GetRequest().GetIsMultiZone() || !createRequest.GetRequest().IsSetZoneConfigs() || len(createRequest.GetRequest().GetZoneConfigs()) == 0 {
		r.m3Client.IncCounter(metrics.ReplicatorCreateRmtCgUUIDScope, metrics.ReplicatorFailures)
		err := &shared.BadRequestError{Message: fmt.Sprintf(`Not a valid create remote cg request for IsMultiZone [%v] or ZoneConfigs not set`, createRequest.GetRequest().GetIsMultiZone())}
		lclLg.WithField(common.TagErr, err).Error(`Create remote cg request verification failed`)
		return err
	}
	if _, ok := r.allZones[r.tenancy]; !ok {
		r.m3Client.IncCounter(metrics.ReplicatorCreateRmtCgUUIDScope, metrics.ReplicatorFailures)
		err := &shared.BadRequestError{Message: fmt.Sprintf(`Unknown tenancy [%s]`, r.tenancy)}
		lclLg.WithField(common.TagErr, err).Error(`Create remote cg failed with unknown tenancy`)
		return err
	}
	for _, zone := range r.allZones[r.tenancy] {
		if strings.EqualFold(zone, r.localZone) {
			continue
		}
		go r.createConsumerGroupRemoteCall(zone, lclLg, createRequest)
	}
	return nil
}
func (r *Replicator) createConsumerGroupRemoteCall(zone string, logger bark.Logger, createRequest *shared.CreateConsumerGroupUUIDRequest) error {
	client, err := r.replicatorclientFactory.GetReplicatorClient(zone)
	if err != nil {
		r.m3Client.IncCounter(metrics.ReplicatorCreateRmtCgUUIDScope, metrics.ReplicatorFailures)
		logger.WithFields(bark.Fields{
			common.TagErr:      err,
			common.TagZoneName: common.FmtZoneName(zone),
		}).Error(`Get remote replicator client failed`)
		return err
	}
	ctx, cancel := thrift.NewContext(remoteReplicatorCallTimeOut)
	defer cancel()
	_, err = client.CreateConsumerGroupUUID(ctx, createRequest)
	if err != nil {
		r.m3Client.IncCounter(metrics.ReplicatorCreateRmtCgUUIDScope, metrics.ReplicatorFailures)
		logger.WithFields(bark.Fields{
			common.TagErr:      err,
			common.TagZoneName: common.FmtZoneName(zone),
		}).Error(`Create remote cg call failed`)
		return err
	}
	return nil
}
func (r *Replicator) UpdateConsumerGroup(ctx thrift.Context, updateRequest *shared.UpdateConsumerGroupRequest) (*shared.ConsumerGroupDescription, error) {
	r.m3Client.IncCounter(metrics.ReplicatorUpdateCgScope, metrics.ReplicatorRequests)
	cgDesc, err := r.metaClient.UpdateConsumerGroup(ctx, updateRequest)
	if err != nil {
		r.logger.WithFields(bark.Fields{
			common.TagCnsPth: common.FmtCnsPth(updateRequest.GetConsumerGroupName()),
			common.TagDstPth: common.FmtDstPth(updateRequest.GetDestinationPath()),
			common.TagDst:    common.FmtDst(cgDesc.GetDestinationUUID()),
			common.TagErr:    err,
		}).Error(`Error updating cg`)
		r.m3Client.IncCounter(metrics.ReplicatorUpdateCgScope, metrics.ReplicatorFailures)
		return nil, err
	}
	r.logger.WithFields(bark.Fields{
		common.TagCnsPth: common.FmtCnsPth(updateRequest.GetConsumerGroupName()),
		common.TagCnsm:   common.FmtCnsm(cgDesc.GetConsumerGroupUUID()),
		common.TagDstPth: common.FmtDstPth(updateRequest.GetDestinationPath()),
		common.TagDst:    common.FmtDst(cgDesc.GetDestinationUUID()),
		common.TagDLQID:  common.FmtDLQID(cgDesc.GetDeadLetterQueueDestinationUUID()),
		`IsMultiZone`:    cgDesc.GetIsMultiZone(),
		`ActiveZone`:     cgDesc.GetActiveZone(),
	}).Info(`Updated cg`)
	return cgDesc, nil
}
func (r *Replicator) UpdateRemoteConsumerGroup(ctx thrift.Context, updateRequest *shared.UpdateConsumerGroupRequest) error {
	r.m3Client.IncCounter(metrics.ReplicatorUpdateRmtCgScope, metrics.ReplicatorRequests)
	if updateRequest == nil {
		r.m3Client.IncCounter(metrics.ReplicatorUpdateRmtCgScope, metrics.ReplicatorFailures)
		err := &shared.BadRequestError{Message: `Update remote cg request has nil request`}
		r.logger.WithField(common.TagErr, err).Error(`Update remote cg request verification failed`)
		return err
	}
	lclLg := r.logger.WithFields(bark.Fields{
		common.TagCnsPth: common.FmtCnsPth(updateRequest.GetConsumerGroupName()),
		common.TagDstPth: common.FmtDstPth(updateRequest.GetDestinationPath()),
	})
	if _, ok := r.allZones[r.tenancy]; !ok {
		r.m3Client.IncCounter(metrics.ReplicatorUpdateRmtCgScope, metrics.ReplicatorFailures)
		err := &shared.BadRequestError{Message: fmt.Sprintf(`Unknown tenancy [%s]`, r.tenancy)}
		lclLg.WithField(common.TagErr, err).Error(`Update remote destination failed with unknown tenancy`)
		return err
	}
	for _, zone := range r.allZones[r.tenancy] {
		if strings.EqualFold(zone, r.localZone) {
			continue
		}
		go r.updateCgRemoteCall(zone, lclLg, updateRequest)
	}
	return nil
}
func (r *Replicator) updateCgRemoteCall(zone string, logger bark.Logger, updateRequest *shared.UpdateConsumerGroupRequest) error {
	client, err := r.replicatorclientFactory.GetReplicatorClient(zone)
	if err != nil {
		r.m3Client.IncCounter(metrics.ReplicatorUpdateRmtCgScope, metrics.ReplicatorFailures)
		logger.WithFields(bark.Fields{
			common.TagErr:      err,
			common.TagZoneName: common.FmtZoneName(zone),
		}).Error(`Get remote replicator client failed`)
		return err
	}
	ctx, cancel := thrift.NewContext(remoteReplicatorCallTimeOut)
	defer cancel()
	_, err = client.UpdateConsumerGroup(ctx, updateRequest)
	if err != nil {
		r.m3Client.IncCounter(metrics.ReplicatorUpdateRmtCgScope, metrics.ReplicatorFailures)
		logger.WithFields(bark.Fields{
			common.TagErr:      err,
			common.TagZoneName: common.FmtZoneName(zone),
		}).Error(`Update remote cg call failed`)
		return err
	}
	return nil
}
func (r *Replicator) DeleteConsumerGroup(ctx thrift.Context, deleteRequest *shared.DeleteConsumerGroupRequest) error {
	r.m3Client.IncCounter(metrics.ReplicatorDeleteCgScope, metrics.ReplicatorRequests)
	err := r.metaClient.DeleteConsumerGroup(ctx, deleteRequest)
	if err != nil {
		r.logger.WithFields(bark.Fields{
			common.TagCnsPth: common.FmtCnsPth(deleteRequest.GetConsumerGroupName()),
			common.TagDstPth: common.FmtDstPth(deleteRequest.GetDestinationPath()),
			common.TagErr:    err,
		}).Error(`Error deleting cg`)
		r.m3Client.IncCounter(metrics.ReplicatorDeleteCgScope, metrics.ReplicatorFailures)
		return err
	}
	r.logger.WithFields(bark.Fields{
		common.TagCnsPth: common.FmtCnsPth(deleteRequest.GetConsumerGroupName()),
		common.TagDstPth: common.FmtDstPth(deleteRequest.GetDestinationPath()),
	}).Info(`Deleted cg`)
	return nil
}
func (r *Replicator) DeleteRemoteConsumerGroup(ctx thrift.Context, deleteRequest *shared.DeleteConsumerGroupRequest) error {
	r.m3Client.IncCounter(metrics.ReplicatorDeleteRmtCgScope, metrics.ReplicatorRequests)
	if deleteRequest == nil {
		r.m3Client.IncCounter(metrics.ReplicatorDeleteRmtCgScope, metrics.ReplicatorFailures)
		err := &shared.BadRequestError{Message: `Delete remote cg request has nil request`}
		r.logger.WithField(common.TagErr, err).Error(`Delete remote cg request verification failed`)
		return err
	}
	lclLg := r.logger.WithFields(bark.Fields{
		common.TagCnsPth: common.FmtCnsPth(deleteRequest.GetConsumerGroupName()),
		common.TagDstPth: common.FmtDstPth(deleteRequest.GetDestinationPath()),
	})
	if _, ok := r.allZones[r.tenancy]; !ok {
		r.m3Client.IncCounter(metrics.ReplicatorDeleteRmtCgScope, metrics.ReplicatorFailures)
		err := &shared.BadRequestError{Message: fmt.Sprintf(`Unknown tenancy [%s]`, r.tenancy)}
		lclLg.WithField(common.TagErr, err).Error(`Delete remote cg failed with unknown tenancy`)
		return err
	}
	for _, zone := range r.allZones[r.tenancy] {
		if strings.EqualFold(zone, r.localZone) {
			continue
		}
		go r.deleteCgRemoteCall(zone, lclLg, deleteRequest)
	}
	return nil
}
func (r *Replicator) deleteCgRemoteCall(zone string, logger bark.Logger, deleteRequest *shared.DeleteConsumerGroupRequest) error {
	client, err := r.replicatorclientFactory.GetReplicatorClient(zone)
	if err != nil {
		r.m3Client.IncCounter(metrics.ReplicatorDeleteRmtCgScope, metrics.ReplicatorFailures)
		logger.WithFields(bark.Fields{
			common.TagErr:      err,
			common.TagZoneName: common.FmtZoneName(zone),
		}).Error(`Get remote replicator client failed`)
		return err
	}
	ctx, cancel := thrift.NewContext(remoteReplicatorCallTimeOut)
	defer cancel()
	err = client.DeleteConsumerGroup(ctx, deleteRequest)
	if err != nil {
		r.m3Client.IncCounter(metrics.ReplicatorDeleteRmtCgScope, metrics.ReplicatorFailures)
		logger.WithFields(bark.Fields{
			common.TagErr:      err,
			common.TagZoneName: common.FmtZoneName(zone),
		}).Error(`Delete remote cg call failed`)
		return err
	}
	return nil
}