func (cgCache *consumerGroupCache) cleanupChannels() {
	close(cgCache.msgsCh)
	close(cgCache.msgCacheCh)
	close(cgCache.msgsRedeliveryCh)
	close(cgCache.priorityMsgsRedeliveryCh)
	close(cgCache.msgCacheRedeliveredCh)
	close(cgCache.ackMsgCh)
	close(cgCache.nackMsgCh)
}
this is a utility routine which closes all message channels
to make sure we free up memory
this can be done outside of a mutex since this will happen only after we
closed everything.
package outputhost
import (
	"sync"
	"sync/atomic"
	"time"
	"github.com/uber-common/bark"
	"github.com/uber/tchannel-go/thrift"
	"github.com/uber/cherami-server/common"
	"github.com/uber/cherami-server/common/dconfig"
	"github.com/uber/cherami-server/common/metrics"
	"github.com/uber/cherami-server/services/outputhost/load"
	"github.com/uber/cherami-thrift/.generated/go/admin"
	"github.com/uber/cherami-thrift/.generated/go/cherami"
	"github.com/uber/cherami-thrift/.generated/go/controller"
	"github.com/uber/cherami-thrift/.generated/go/metadata"
	"github.com/uber/cherami-thrift/.generated/go/shared"
)
const (
	kafkaConnectedStoreUUID = `cafca000-0000-0caf-ca00-0000000cafca`  
)
type (
	cacheMsg struct {
		msg    *cherami.ConsumerMessage
		connID int
	}
	consumerGroupCache struct {
		extMutex sync.RWMutex
		cachedCGDesc shared.ConsumerGroupDescription
		cachedTime common.UnixNanoTime
		destPath string
		currID int
		msgsCh chan *cherami.ConsumerMessage
		msgsRedeliveryCh chan *cherami.ConsumerMessage
		priorityMsgsRedeliveryCh chan *cherami.ConsumerMessage
		msgCacheCh chan cacheMsg
		msgCacheRedeliveredCh chan cacheMsg
		ackMsgCh chan timestampedAckID
		nackMsgCh chan timestampedAckID
		notifyConsCloseCh chan int
		notifyReplicaCloseCh chan string
		connections map[int]*consConnection
		extentCache map[string]*extentCache
		msgDeliveryCache *cgMsgCache
		closeChannel chan struct{}
		shutdownWG *sync.WaitGroup
		metaClient metadata.TChanMetadataService
		tClients common.ClientFactory
		wsConnector common.WSConnector
		outputHostUUID string
		logger bark.Logger
		notifyUnloadCh chan<- string
		consumerM3Client metrics.Client
		m3Client metrics.Client
		notifier Notifier
		creditNotifyCh chan int32
		creditRequestCh chan string
		lastDisconnectTime time.Time
		dlqMerging int32
		sessionID uint16
		ackMgrLoadCh chan<- ackMgrLoadMsg
		ackMgrUnloadCh chan<- uint32
		ackIDGen common.HostAckIDGenerator
		loadReporterFactory common.LoadReporterDaemonFactory
		loadReporter common.LoadReporterDaemon
		hostMetrics *load.HostMetrics
		cgMetrics *load.CGMetrics
		unloadInProgress bool
		connsWG sync.WaitGroup
		manageMsgCacheWG sync.WaitGroup
		cfgMgr dconfig.ConfigManager
		kafkaCluster string
		kafkaTopics []string
		kafkaMessageConverterFactory KafkaMessageConverterFactory
	}
)
const getAddressCtxTimeout = 60 * time.Second
const metaPollTimeout = 1 * time.Minute
const idleTimeout = 30 * time.Minute
const unloadTickerTimeout = 10 * time.Minute
const defaultMaxResults = 500
const consGroupLoadReportingInterval = 2 * time.Second
const defaultNumOutstandingMsgs int32 = 10000
var ErrCgUnloaded = &cherami.InternalServiceError{Message: "ConsumerGroup already unloaded"}
var ErrConfigCast = &cherami.InternalServiceError{Message: "Unable to cast to OutputCgConfig"}
func newConsumerGroupCache(destPath string, cgDesc shared.ConsumerGroupDescription, cgLogger bark.Logger, h *OutputHost) *consumerGroupCache {
	cgCache := &consumerGroupCache{
		destPath:                 destPath,
		cachedCGDesc:             cgDesc,
		cachedTime:               common.Now(),
		extentCache:              make(map[string]*extentCache),
		msgsCh:                   make(chan *cherami.ConsumerMessage, defaultPrefetchBufferSize),
		msgsRedeliveryCh:         make(chan *cherami.ConsumerMessage, defaultPrefetchBufferSize),
		priorityMsgsRedeliveryCh: make(chan *cherami.ConsumerMessage, 1),
		notifyConsCloseCh:        make(chan int, 5),
		notifyReplicaCloseCh:     make(chan string, 5),
		connections:              make(map[int]*consConnection),
		msgCacheCh:               make(chan cacheMsg, defaultPrefetchBufferSize),
		msgCacheRedeliveredCh:    make(chan cacheMsg, defaultPrefetchBufferSize),
		ackMsgCh:                 make(chan timestampedAckID, ackChannelSize),  
		nackMsgCh:                make(chan timestampedAckID, ackChannelSize),
		closeChannel:             make(chan struct{}),
		outputHostUUID:           h.GetHostUUID(),
		tClients:                 h.GetClientFactory(),
		wsConnector:              h.GetWSConnector(),
		metaClient:               h.metaClient,
		shutdownWG:               &h.shutdownWG,
		logger:                   cgLogger.WithField(common.TagModule, `cgCache`),
		notifyUnloadCh:           h.unloadCacheCh,
		m3Client:                 h.m3Client,
		notifier:                 newNotifier(),
		creditNotifyCh:           make(chan int32, 50),
		creditRequestCh:          make(chan string, 50),
		lastDisconnectTime:       time.Now(),
		sessionID:                h.sessionID,
		ackIDGen:                 h.ackMgrIDGen,
		ackMgrLoadCh:             h.ackMgrLoadCh,
		ackMgrUnloadCh:           h.ackMgrUnloadCh,
		loadReporterFactory:      h.GetLoadReporterDaemonFactory(),
		hostMetrics:              h.hostMetrics,
		cgMetrics:                load.NewCGMetrics(),
		cfgMgr:                   h.cfgMgr,
		kafkaMessageConverterFactory: h.kafkaStreamFactory,
	}
	cgCache.consumerM3Client = metrics.NewClientWithTags(h.m3Client, metrics.Outputhost, cgCache.getConsumerGroupTags())
	cgCache.loadReporter = cgCache.loadReporterFactory.CreateReporter(consGroupLoadReportingInterval, cgCache, cgLogger)
	cgCache.loadReporter.Start()
	return cgCache
}
func (cgCache *consumerGroupCache) getConsumerGroupTags() map[string]string {
	destTagValue, tagErr := common.GetTagsFromPath(cgCache.destPath)
	if tagErr != nil {
		destTagValue = metrics.UnknownDirectoryTagValue
		cgCache.logger.WithField(common.TagDstPth, cgCache.destPath).
			WithField(common.TagUnknowPth, metrics.UnknownDirectoryTagValue).
			Error("unknown destination path, return default name")
	}
	cgTagValue, tagErr := common.GetTagsFromPath(cgCache.cachedCGDesc.GetConsumerGroupName())
	if tagErr != nil {
		cgTagValue = metrics.UnknownDirectoryTagValue
		cgCache.logger.WithField(common.TagCnsPth, cgCache.cachedCGDesc.GetConsumerGroupName()).
			WithField(common.TagUnknowPth, metrics.UnknownDirectoryTagValue).
			Error("unknown consumer group path, return default name")
	}
	tags := map[string]string{
		metrics.DestinationTagName:   destTagValue,
		metrics.ConsumerGroupTagName: cgTagValue,
	}
	return tags
}
func (cgCache *consumerGroupCache) loadExtentCache(ctx thrift.Context, destType shared.DestinationType, cge *shared.ConsumerGroupExtent) {
	var committer Committer
	extUUID := cge.GetExtentUUID()
	if extCache, exists := cgCache.extentCache[extUUID]; !exists {
		extCache = &extentCache{
			cgUUID:                       cgCache.cachedCGDesc.GetConsumerGroupUUID(),
			extUUID:                      extUUID,
			destUUID:                     cgCache.cachedCGDesc.GetDestinationUUID(),
			destType:                     destType,
			storeUUIDs:                   cge.StoreUUIDs,
			startFrom:                    time.Unix(0, cgCache.cachedCGDesc.GetStartFrom()),
			skipOlder:                    time.Duration(int64(cgCache.cachedCGDesc.GetSkipOlderMessagesSeconds()) * int64(time.Second)),
			delay:                        time.Duration(int64(cgCache.cachedCGDesc.GetDelaySeconds()) * int64(time.Second)),
			notifyReplicaCloseCh:         make(chan error, 5),
			closeChannel:                 make(chan struct{}),
			waitConsumedCh:               make(chan bool, 1),
			msgsCh:                       cgCache.msgsCh,
			connectionsClosedCh:          cgCache.notifyReplicaCloseCh,
			shutdownWG:                   &cgCache.connsWG,
			tClients:                     cgCache.tClients,
			wsConnector:                  cgCache.wsConnector,
			logger:                       cgCache.logger.WithFields(bark.Fields{common.TagExt: extUUID, common.TagModule: `extCache`}),
			creditNotifyCh:               cgCache.creditNotifyCh,
			creditRequestCh:              cgCache.creditRequestCh,
			initialCredits:               defaultNumOutstandingMsgs,
			loadMetrics:                  load.NewExtentMetrics(),
			consumerM3Client:             cgCache.consumerM3Client,
			kafkaMessageConverterFactory: cgCache.kafkaMessageConverterFactory,
		}
		cgCache.extentCache[extUUID] = extCache
		cgCache.cgMetrics.Increment(load.CGMetricNumOpenExtents)
		cgCache.hostMetrics.Increment(load.HostMetricNumOpenExtents)
		extCache.singleCGVisible = cgCache.checkSingleCGVisible(ctx, cge)
		if extCache.singleCGVisible {
			nDLQMergingExtents := atomic.AddInt32(&cgCache.dlqMerging, 1)
			extCache.logger.WithField(`nDLQMergingExtents`, nDLQMergingExtents).Info("Merging DLQ Extent(s)")
		}
		cfg, err := cgCache.getDynamicCgConfig()
		if err == nil {
			extCache.initialCredits = cgCache.getMessageCacheSize(cfg, defaultNumOutstandingMsgs)
		}
		if common.IsKafkaConsumerGroupExtent(cge) {
			committer = NewKafkaCommitter(
				cgCache.outputHostUUID,
				cgCache.cachedCGDesc.GetConsumerGroupUUID(),
				extCache.logger,
				&extCache.kafkaClient,
			)
		} else {
			committer = newCheramiCommitter(
				cgCache.metaClient,
				cgCache.outputHostUUID,
				cgCache.cachedCGDesc.GetConsumerGroupUUID(),
				extCache.extUUID,
				&extCache.connectedStoreUUID,
				cgCache.cachedCGDesc.GetIsMultiZone(),
				cgCache.tClients,
			)
		}
		extCache.ackMgr = newAckManager(
			cgCache,
			cgCache.ackIDGen.GetNextAckID(),
			cgCache.outputHostUUID,
			cgCache.cachedCGDesc.GetConsumerGroupUUID(),
			extCache.extUUID,
			&extCache.connectedStoreUUID,
			extCache.waitConsumedCh,
			cge,
			committer,
			extCache.logger,
		)
		extCache.loadReporter = cgCache.loadReporterFactory.CreateReporter(extentLoadReportingInterval, extCache, extCache.logger)
		extCache.shutdownWG.Add(1)
		go extCache.load(
			cgCache.outputHostUUID,
			cgCache.cachedCGDesc.GetConsumerGroupUUID(),
			cgCache.cachedCGDesc.GetConsumerGroupName(),
			cgCache.kafkaCluster,
			cgCache.kafkaTopics,
			cgCache.metaClient,
			cge,
			cgCache.consumerM3Client,
		)
		cgCache.ackMgrLoadCh <- ackMgrLoadMsg{uint32(extCache.ackMgr.ackMgrID), extCache.ackMgr}
	}
	return
}
func (cgCache *consumerGroupCache) unloadExtentCache(extUUID string) (ackMgrUnloadID uint32) {
	if extCache, ok := cgCache.extentCache[extUUID]; ok {
		extCache.logger.Info("removing extent")
		ackMgrUnloadID = uint32(extCache.ackMgr.ackMgrID)
		cgCache.cgMetrics.Decrement(load.CGMetricNumOpenExtents)
		cgCache.hostMetrics.Decrement(load.HostMetricNumOpenExtents)
		if extCache.singleCGVisible {
			dlqMerging := atomic.AddInt32(&cgCache.dlqMerging, -1)
			if dlqMerging <= 0 {  
				cgCache.logger.Info("done merging DLQ extent(s)")
			}
		}
		delete(cgCache.extentCache, extUUID)
	} else {
		cgCache.logger.WithField(common.TagExt, extUUID).Warn("tried to unload not-loaded extent")
	}
	return
}
func (cgCache *consumerGroupCache) manageConsumerGroupCache() {
	cgCache.logger.Info(`cgCache initialized`)
	defer cgCache.shutdownWG.Done()
	quit := false
	refreshTicker := time.NewTicker(metaPollTimeout)  
	defer refreshTicker.Stop()
	unloadTicker := time.NewTicker(unloadTickerTimeout)  
	defer unloadTicker.Stop()
	var ackMgrUnloadID uint32
	for !quit {
		select {
		case conn := <-cgCache.notifyConsCloseCh:
			cgCache.extMutex.Lock()
			if _, ok := cgCache.connections[conn]; ok {
				cgCache.logger.WithField(`conn`, conn).Info(`removing connection`)
				delete(cgCache.connections, conn)
				cgCache.cgMetrics.Decrement(load.CGMetricNumOpenConns)
			}
			if cgCache.getNumOpenConns() <= 0 {
				cgCache.lastDisconnectTime = time.Now()
			}
			cgCache.extMutex.Unlock()
		case extUUID := <-cgCache.notifyReplicaCloseCh:
			cgCache.extMutex.Lock()
			ackMgrUnloadID = cgCache.unloadExtentCache(extUUID)
			if len(cgCache.extentCache) == 0 {
				for _, conn := range cgCache.connections {
					go conn.close()
				}
			}
			cgCache.extMutex.Unlock()
			if ackMgrUnloadID > 0 {
				select {
				case cgCache.ackMgrUnloadCh <- ackMgrUnloadID:
					ackMgrUnloadID = 0
				default:
				}
			}
		case <-refreshTicker.C:
			cgCache.logger.Debug("refreshing all extents")
			cgCache.extMutex.Lock()
			cgCache.refreshCgCache(nil)
			cgCache.extMutex.Unlock()
		case <-unloadTicker.C:
			cgCache.extMutex.Lock()
			if cgCache.getNumOpenConns() <= 0 {
				if time.Since(cgCache.lastDisconnectTime) > idleTimeout {
					cgCache.logger.Info("unloading empty/idle cache")
					go cgCache.unloadConsumerGroupCache()
				}
			}
			cgCache.extMutex.Unlock()
		case <-cgCache.closeChannel:
			cgCache.msgDeliveryCache.stop()
			cgCache.manageMsgCacheWG.Wait()
			cgCache.cleanupChannels()
			cgCache.logger.Info("cg is stopped and all channels are cleanedup")
			quit = true
		}
	}
}
func (cgCache *consumerGroupCache) refreshCgCacheNoLock(ctx thrift.Context) error {
	cgCache.extMutex.Lock()
	defer cgCache.extMutex.Unlock()
	if cgCache.isClosed() || cgCache.unloadInProgress {
		return ErrCgUnloaded
	}
	return cgCache.refreshCgCache(ctx)
}
func (cgCache *consumerGroupCache) refreshCgCache(ctx thrift.Context) error {
	dstReq := &shared.ReadDestinationRequest{DestinationUUID: common.StringPtr(cgCache.cachedCGDesc.GetDestinationUUID())}
	dstDesc, errRD := cgCache.metaClient.ReadDestination(ctx, dstReq)
	if errRD != nil || dstDesc == nil {
		cgCache.logger.WithField(common.TagErr, errRD).Error(`ReadDestination failed on refresh`)
		return errRD
	}
	if dstDesc.GetStatus() == shared.DestinationStatus_DELETING ||
		dstDesc.GetStatus() == shared.DestinationStatus_DELETED {
		cgCache.logger.Info("destination deleted; unloading all extents")
		go cgCache.unloadConsumerGroupCache()
		return ErrCgUnloaded
	}
	if dstDesc.GetType() == shared.DestinationType_KAFKA {
		cgCache.kafkaCluster = dstDesc.GetKafkaCluster()
		cgCache.kafkaTopics = dstDesc.GetKafkaTopics()
	}
	readReq := &shared.ReadConsumerGroupRequest{
		DestinationUUID:   common.StringPtr(cgCache.cachedCGDesc.GetDestinationUUID()),
		ConsumerGroupName: common.StringPtr(cgCache.cachedCGDesc.GetConsumerGroupName()),
	}
	cgDesc, errRCG := cgCache.metaClient.ReadConsumerGroup(ctx, readReq)
	if errRCG != nil {
		cgCache.logger.WithField(common.TagErr, errRCG).Error(`ReadConsumerGroup failed on refresh`)
		return errRCG
	}
	if cgDesc.GetStatus() == shared.ConsumerGroupStatus_DELETING ||
		cgDesc.GetStatus() == shared.ConsumerGroupStatus_DELETED {
		cgCache.logger.Info("consumer group deleted; unloading all extents")
		go cgCache.unloadConsumerGroupCache()
		return ErrCgUnloaded
	}
	cgCache.cachedCGDesc.Status = cgDesc.Status
	cgCache.cachedCGDesc.MaxDeliveryCount = cgDesc.MaxDeliveryCount
	cgReq := &shared.ReadConsumerGroupExtentsRequest{
		DestinationUUID:   common.StringPtr(cgCache.cachedCGDesc.GetDestinationUUID()),
		ConsumerGroupUUID: common.StringPtr(cgCache.cachedCGDesc.GetConsumerGroupUUID()),
		OutputHostUUID:    common.StringPtr(cgCache.outputHostUUID),
		Status:            common.MetadataConsumerGroupExtentStatusPtr(shared.ConsumerGroupExtentStatus_OPEN),
		MaxResults:        common.Int32Ptr(defaultMaxResults),
	}
	for {
		cgRes, errRCGES := cgCache.metaClient.ReadConsumerGroupExtents(ctx, cgReq)
		if cgRes == nil || errRCGES != nil {
			cgCache.logger.WithField(common.TagErr, errRCGES).Error("ReadConsumerGroupExtents failed on refresh")
			return errRCGES
		}
		for _, cge := range cgRes.GetExtents() {
			cgCache.loadExtentCache(ctx, dstDesc.GetType(), cge)
		}
		if len(cgRes.GetNextPageToken()) == 0 {
			break
		} else {
			cgReq.PageToken = cgRes.GetNextPageToken()
		}
	}
	return nil
}
func (cgCache *consumerGroupCache) checkSingleCGVisible(ctx thrift.Context, cge *shared.ConsumerGroupExtent) (singleCgVisible bool) {
	req := &metadata.ReadExtentStatsRequest{
		DestinationUUID: common.StringPtr(cgCache.cachedCGDesc.GetDestinationUUID()),
		ExtentUUID:      common.StringPtr(cge.GetExtentUUID()),
	}
	ext, err := cgCache.metaClient.ReadExtentStats(ctx, req)
	if err == nil {
		if ext != nil && ext.GetExtentStats() != nil {
			singleCgVisible = ext.GetExtentStats().GetConsumerGroupVisibility() != ``
			if singleCgVisible {
				cgCache.logger.WithFields(bark.Fields{common.TagExt: common.FmtExt(cge.GetExtentUUID())}).Info("Consuming single CG visible extent")
			}
		}
	} else {
		cgCache.logger.WithFields(bark.Fields{common.TagErr: err, common.TagExt: common.FmtExt(cge.GetExtentUUID())}).Error("ReadExtentStats failed during extent load")
	}
	return
}
func (cgCache *consumerGroupCache) getDynamicCgConfig() (OutputCgConfig, error) {
	dCfgIface, err := cgCache.cfgMgr.Get(common.OutputServiceName, `*`, `*`, `*`)
	if err != nil {
		cgCache.logger.WithFields(bark.Fields{common.TagErr: err}).Error(`Couldn't get the configuration object`)
		return OutputCgConfig{}, err
	}
	cfg, ok := dCfgIface.(OutputCgConfig)
	if !ok {
		cgCache.logger.Error(`Couldn't cast cfg to OutputCgConfig`)
		return OutputCgConfig{}, ErrConfigCast
	}
	return cfg, nil
}
func (cgCache *consumerGroupCache) getMessageCacheSize(cfg OutputCgConfig, oldSize int32) (cacheSize int32) {
	logFn := func() bark.Logger {
		return cgCache.logger
	}
	ruleKey := cgCache.destPath + `/` + cgCache.cachedCGDesc.GetConsumerGroupName()
	cacheSize = int32(common.OverrideValueByPrefix(logFn, ruleKey, cfg.MessageCacheSize, int64(oldSize), `messagecachesize`))
	return cacheSize
}
func (cgCache *consumerGroupCache) loadConsumerGroupCache(ctx thrift.Context, exists bool) error {
	cgCache.extMutex.Lock()
	defer cgCache.extMutex.Unlock()
	if cgCache.isClosed() || cgCache.unloadInProgress {
		return ErrCgUnloaded
	}
	if !exists {
		dlq, err := newDeadLetterQueue(
			ctx,
			cgCache.logger,
			cgCache.cachedCGDesc,
			cgCache.metaClient,
			cgCache.m3Client,
			cgCache.consumerM3Client,
		)
		if err != nil {
			return err
		}
		cgCache.msgDeliveryCache = newMessageDeliveryCache(dlq, defaultNumOutstandingMsgs, cgCache)
		cgCache.shutdownWG.Add(1)
		go cgCache.manageConsumerGroupCache()  
		cgCache.manageMsgCacheWG.Add(1)        
		go cgCache.msgDeliveryCache.start()    
		return cgCache.refreshCgCache(ctx)
	}
	return nil
}
func (cgCache *consumerGroupCache) reconfigureClients(updateUUID string) {
	var notified, dropped int
	cgCache.extMutex.RLock()
	for _, conn := range cgCache.connections {
		select {
		case conn.reconfigureClientCh <- updateUUID:
			notified++
		default:
			dropped++
		}
	}
	cgCache.extMutex.RUnlock()
	cgCache.logger.WithFields(bark.Fields{
		common.TagUpdateUUID: updateUUID,
		`notified`:           notified,
		`dropped`:            dropped,
	}).Info(`reconfigureClients: notified clients`)
}
func (cgCache *consumerGroupCache) isClosed() bool {
	select {
	case <-cgCache.closeChannel:
		return true
	default:
		return false
	}
}
func (cgCache *consumerGroupCache) Report(reporter common.LoadReporter) {
	msgsOut := cgCache.cgMetrics.GetAndReset(load.CGMetricMsgsOut)
	bytesOut := cgCache.cgMetrics.GetAndReset(load.CGMetricBytesOut)
	numConns := cgCache.cgMetrics.Get(load.CGMetricNumOpenConns)
	numExtents := cgCache.cgMetrics.Get(load.CGMetricNumOpenExtents)
	smartRetryOn := cgCache.cgMetrics.Get(load.CGMetricSmartRetryOn)
	cgMetrics := controller.ConsumerGroupMetrics{
		NumberOfActiveExtents:   common.Int64Ptr(numExtents),
		NumberOfConnections:     common.Int64Ptr(numConns),
		OutgoingMessagesCounter: common.Int64Ptr(msgsOut),
		OutgoingBytesCounter:    common.Int64Ptr(bytesOut),
		SmartRetryOnCounter:     common.Int64Ptr(smartRetryOn),
	}
	dstID := cgCache.cachedCGDesc.GetDestinationUUID()
	cgID := cgCache.cachedCGDesc.GetConsumerGroupUUID()
	reporter.ReportConsumerGroupMetric(dstID, cgID, cgMetrics)
	cgCache.consumerM3Client.UpdateGauge(metrics.ConsConnectionScope, metrics.OutputhostCGConsConnection, numConns)
	cgCache.consumerM3Client.UpdateGauge(metrics.ConsConnectionScope, metrics.OutputhostCGNumExtents, numExtents)
}
func (cgCache *consumerGroupCache) getNumOpenConns() int32 {
	return int32(cgCache.cgMetrics.Get(load.CGMetricNumOpenConns))
}
func (cgCache *consumerGroupCache) incNumOpenConns(delta int32) int32 {
	return int32(cgCache.cgMetrics.Add(load.CGMetricNumOpenConns, int64(delta)))
}
func (cgCache *consumerGroupCache) utilSendCreditsToExtent(extUUID string, credits int32) bool {
	cgCache.extMutex.RLock()
	defer cgCache.extMutex.RUnlock()
	ret := false
	if extCache, ok := cgCache.extentCache[extUUID]; ok {
		ret = extCache.grantCredits(credits)
	}
	return ret
}
func (cgCache *consumerGroupCache) updateLastDisconnectTime() {
	cgCache.extMutex.Lock()
	defer cgCache.extMutex.Unlock()
	cgCache.lastDisconnectTime = time.Now()
}
func (cgCache *consumerGroupCache) unloadConsumerGroupCache() {
	cgCache.extMutex.Lock()
	if cgCache.isClosed() || cgCache.unloadInProgress {
		cgCache.extMutex.Unlock()
		return
	}
	cgCache.unloadInProgress = true
	for _, conn := range cgCache.connections {
		go conn.close()
	}
	for _, extCache := range cgCache.extentCache {
		go extCache.unload()
	}
	cgCache.extMutex.Unlock()
	cgCache.notifyUnloadCh <- cgCache.cachedCGDesc.GetConsumerGroupUUID()
	cgCache.connsWG.Wait()
	close(cgCache.closeChannel)
	cgCache.loadReporter.Stop()
	cgCache.consumerM3Client.UpdateGauge(metrics.ConsConnectionScope, metrics.OutputhostCGConsConnection, 0)
	cgCache.consumerM3Client.UpdateGauge(metrics.ConsConnectionScope, metrics.OutputhostCGNumExtents, 0)
}
func (cgCache *consumerGroupCache) getCgState(cgUUID string) *admin.ConsumerGroupState {
	cgCache.extMutex.RLock()
	defer cgCache.extMutex.RUnlock()
	cgState := admin.NewConsumerGroupState()
	cgState.CgUUID = common.StringPtr(cgUUID)
	cgState.NumOutstandingMsgs = common.Int32Ptr(cgCache.msgDeliveryCache.getOutstandingMsgs())
	cgState.MsgChSize = common.Int64Ptr(int64(len(cgCache.msgsCh)))
	cgState.NumConnections = common.Int64Ptr(int64(cgCache.getNumOpenConns()))
	cgState.CgExtents = make([]*admin.OutputCgExtent, 0)
	for _, extCache := range cgCache.extentCache {
		extState := extCache.getState()
		cgState.CgExtents = append(cgState.CgExtents, extState)
	}
	return cgState
}