func (h *OutputHost) UnloadConsumerGroups(ctx thrift.Context, request *admin.UnloadConsumerGroupsRequest) (err error) {
	defer atomic.AddInt32(&h.loadShutdownRef, -1)
	sw := h.m3Client.StartTimer(metrics.UnloadConsumerGroupsScope, metrics.OutputhostLatencyTimer)
	defer sw.Stop()
	h.m3Client.IncCounter(metrics.UnloadConsumerGroupsScope, metrics.OutputhostRequests)
	if atomic.AddInt32(&h.loadShutdownRef, 1) <= 0 {
		h.logger.Error("not unloading the CG cache outputHost already shutdown")
		h.m3Client.IncCounter(metrics.UnloadConsumerGroupsScope, metrics.OutputhostFailures)
		return ErrHostShutdown
	}
	for _, cgUUID := range request.CgUUIDs {
		h.cgMutex.RLock()
		cg, ok := h.cgCache[cgUUID]
		if ok {
			go cg.unloadConsumerGroupCache()
		} else {
			h.logger.WithField(common.TagCnsm, common.FmtCnsm(cgUUID)).
				Error("consumer group is not cached at all")
			err = ErrCgAlreadyUnloaded
			h.m3Client.IncCounter(metrics.UnloadConsumerGroupsScope, metrics.OutputhostFailures)
		}
		h.cgMutex.RUnlock()
	}
	return err
}
UnloadConsumerGroups is the API used to unload consumer groups to clear the cache
package outputhost
import (
	"fmt"
	"net/http"
	"strconv"
	"sync"
	"sync/atomic"
	"time"
	"github.com/uber-common/bark"
	"github.com/uber/tchannel-go/thrift"
	"github.com/Shopify/sarama"
	ccommon "github.com/uber/cherami-client-go/common"
	"github.com/uber/cherami-server/common"
	"github.com/uber/cherami-server/common/configure"
	cassDconfig "github.com/uber/cherami-server/common/dconfig"
	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/services/outputhost/load"
	"github.com/uber/cherami-server/stream"
	"github.com/uber/cherami-thrift/.generated/go/admin"
	"github.com/uber/cherami-thrift/.generated/go/cherami"
	ccherami "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 (
	defaultIdleTimeout        = 10 * time.Minute
	defaultPrefetchBufferSize = 1000  
	defaultUnloadChSize       = 50
	defaultAckMgrMapChSize    = 500
	defaultAckMgrIDStartFrom  = 0                 
	defaultMaxConnLimitPerCg  = 1000              
	hostLoadReportingInterval = 2 * time.Second   
	msgCacheWriteTimeout      = 10 * time.Minute  
)
var thisOutputHost *OutputHost
type (
	OutputHost struct {
		logger             bark.Logger
		loadShutdownRef    int32
		cgMutex            sync.RWMutex
		cgCache            map[string]*consumerGroupCache
		shutdownWG         sync.WaitGroup
		cacheTimeout       time.Duration
		metaClient         metadata.TChanMetadataService
		frontendClient     ccherami.TChanBFrontend
		hostIDHeartbeater  common.HostIDHeartbeater
		loadReporter       common.LoadReporterDaemon
		shutdownCh         chan struct{}
		unloadCacheCh      chan string
		m3Client           metrics.Client
		dClient            dconfig.Client
		numConsConn        int32                   
		ackMgrMap          map[uint32]*ackManager  
		ackMgrMutex        sync.RWMutex            
		sessionID          uint16
		ackMgrIDGen        common.HostAckIDGenerator  
		ackMgrLoadCh       chan ackMgrLoadMsg
		ackMgrUnloadCh     chan uint32
		hostMetrics        *load.HostMetrics
		cfgMgr             cassDconfig.ConfigManager
		kafkaCfg           configure.CommonKafkaConfig
		kafkaStreamFactory KafkaMessageConverterFactory
		common.SCommon
	}
	OutOptions struct {
		CacheIdleTimeout time.Duration
		KStreamFactory KafkaMessageConverterFactory
	}
	ackMgrLoadMsg struct {
		ackMgrID uint32
		ackMgr   *ackManager
	}
)
var _ cherami.TChanBOut = &OutputHost{}
var _ admin.TChanOutputHostAdmin = &OutputHost{}
var _ common.WSService = &OutputHost{}
var ErrHostShutdown = &cherami.InternalServiceError{Message: "OutputHost already shutdown"}
var ErrCgAlreadyUnloaded = &cherami.InternalServiceError{Message: "Consumer Group already unloaded"}
var ErrLimit = &cherami.InternalServiceError{Message: "Outputhost cons connection limit exceeded"}
func (h *OutputHost) getConsumerGroup(ctx thrift.Context, path string, cgName string, rejectDisabled bool) (*shared.ConsumerGroupDescription, metrics.ErrorClass, error) {
	mGetRequest := shared.ReadConsumerGroupRequest{
		DestinationPath:   common.StringPtr(path),
		ConsumerGroupName: common.StringPtr(cgName),
	}
	cgDesc, err := h.metaClient.ReadConsumerGroup(ctx, &mGetRequest)
	if cgDesc != nil && err == nil {
		if rejectDisabled {
			if cgDesc.GetStatus() != shared.ConsumerGroupStatus_ENABLED {
				return nil, metrics.UserError, cherami.NewEntityDisabledError()
			}
		}
		return cgDesc, metrics.NoError, nil
	}
	return nil, common.ClassifyErrorByType(err), err
}
func (h *OutputHost) createConsumerGroupCache(cgDesc shared.ConsumerGroupDescription, destPath string,
	cgLogger bark.Logger) (cgCache *consumerGroupCache, exists bool) {
	h.cgMutex.Lock()
	if cgCache, exists = h.cgCache[cgDesc.GetConsumerGroupUUID()]; !exists {
		cgCache = newConsumerGroupCache(destPath, cgDesc, cgLogger, h)
		h.cgCache[cgDesc.GetConsumerGroupUUID()] = cgCache
	}
	h.cgMutex.Unlock()
	return
}
func (h *OutputHost) createAndLoadCGCache(ctx thrift.Context, cgDesc shared.ConsumerGroupDescription, destPath string,
	cgLogger bark.Logger) (*consumerGroupCache, error) {
	if atomic.AddInt32(&h.loadShutdownRef, 1) <= 0 {
		cgLogger.Warn("not loading the CG cache outputHost already shutdown")
		return nil, ErrHostShutdown
	}
	cgCache, exists := h.createConsumerGroupCache(cgDesc, destPath, cgLogger)
	return h.loadCGCache(ctx, exists, cgCache)
}
func (h *OutputHost) loadCGCache(ctx thrift.Context, exists bool, cgCache *consumerGroupCache) (*consumerGroupCache, error) {
	err := cgCache.loadConsumerGroupCache(ctx, exists)
	if err != nil {
		cgCache.logger.Error("outhost: error loading the CG cache")
		if !exists && err != ErrCgUnloaded {
			cgCache.logger.Warn("outputhost: unloading consumer group cache.")
			go cgCache.unloadConsumerGroupCache()
		}
		return nil, err
	}
	return cgCache, nil
}
func (h *OutputHost) getAckMgr(ackMgrID uint16) (ackmgr *ackManager) {
	h.ackMgrMutex.RLock()
	ackmgr, _ = h.ackMgrMap[uint32(ackMgrID)]
	h.ackMgrMutex.RUnlock()
	return
}
func (h *OutputHost) OpenConsumerStreamHandler(w http.ResponseWriter, r *http.Request) {
	path := r.Header.Get("path")
	if len(path) == 0 {
		err := &cherami.BadRequestError{Message: `please set path as part of the header`}
		h.logger.WithField(common.TagErr, err).Error("OpenConsumerStreamHandler failed")
		h.incFailureCounter(metrics.OpenConsumerStreamScope, metrics.UserError)
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	cgName := r.Header.Get("consumerGroupName")
	if len(cgName) == 0 {
		err := &cherami.BadRequestError{Message: `please set consumerGroupName as part of the header`}
		h.logger.WithField(common.TagErr, err).Error("OpenConsumerStreamScope failed")
		h.incFailureCounter(metrics.OpenConsumerStreamScope, metrics.UserError)
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	wsStream, err := h.GetWSConnector().AcceptConsumerStream(w, r)
	if err != nil {
		h.logger.WithFields(bark.Fields{
			common.TagDstPth: common.FmtDstPth(path),
			common.TagCnsPth: common.FmtCnsPth(cgName),
			common.TagErr:    err,
		}).Error("unable to upgrade websocket connection")
		h.incFailureCounter(metrics.OpenConsumerStreamScope, metrics.InternalError)
		return
	}
	ctx, cancel := thrift.NewContext(common.MaxDuration)
	defer cancel()
	ctx = thrift.WithHeaders(ctx, map[string]string{
		"path":              path,
		"consumerGroupName": cgName,
	})
	h.OpenConsumerStream(ctx, wsStream)
}
func (h *OutputHost) OpenConsumerStream(ctx thrift.Context, call stream.BOutOpenConsumerStreamInCall) error {
	h.m3Client.IncCounter(metrics.OpenConsumerStreamScope, metrics.OutputhostRequests)
	path, ok := ctx.Headers()["path"]
	if !ok {
		err := &cherami.BadRequestError{Message: `please set path as part of the header`}
		h.logger.WithField(common.TagErr, err).Error("OpenConsumerStream failed")
		h.incFailureCounter(metrics.OpenConsumerStreamScope, metrics.UserError)
		call.Done()
		return err
	}
	cgName, ok := ctx.Headers()["consumerGroupName"]
	if !ok {
		err := &cherami.BadRequestError{Message: `please set consumerGroupName as part of the header`}
		h.logger.WithField(common.TagErr, err).Error("OpenConsumerStream failed")
		h.incFailureCounter(metrics.OpenConsumerStreamScope, metrics.UserError)
		call.Done()
		return err
	}
	cgLogger := h.logger.WithFields(bark.Fields{
		common.TagDstPth: common.FmtDstPth(path),
		common.TagCnsPth: common.FmtCnsPth(cgName),
	})
	cgDesc, errC, err := h.getConsumerGroup(ctx, path, cgName, false  )
	if err != nil || cgDesc == nil || len(cgDesc.GetDestinationUUID()) == 0 || len(cgDesc.GetConsumerGroupUUID()) == 0 {
		cgLogger.WithField(common.TagErr, err).Error(`error translating dest/cg name to uuid`)
		call.Done()
		h.incFailureCounter(metrics.OpenConsumerStreamScope, errC)
		return err
	}
	cgLogger = cgLogger.WithFields(bark.Fields{
		common.TagCnsm: cgDesc.GetConsumerGroupUUID(),
		common.TagDst:  cgDesc.GetDestinationUUID(),
	})
	cgCache, err := h.createAndLoadCGCache(ctx, *cgDesc, path, cgLogger)
	if err != nil {
		cgLogger.WithField(common.TagErr, err).Error(`error loading consumer group`)
		atomic.AddInt32(&h.loadShutdownRef, -1)
		call.Done()
		h.incFailureCounter(metrics.OpenConsumerStreamScope, errC)
		return err
	}
	cgCache.extMutex.Lock()
	if h.IsLimitsEnabled() && (cgCache.getNumOpenConns() > defaultMaxConnLimitPerCg) {
		cgCache.extMutex.Unlock()
		cgLogger.WithFields(bark.Fields{
			common.TagHostConnLimit: common.FmtHostConnLimit(int(cgCache.getNumOpenConns())),
			common.TagErr:           ErrLimit,
		}).Error("per-cg limit exceeded; rejecting connection")
		atomic.AddInt32(&h.loadShutdownRef, -1)
		call.Done()
		h.incFailureCounter(metrics.OpenConsumerStreamScope, metrics.UserError)
		return ErrLimit
	}
	conn := newConsConnection(cgCache.currID, cgCache, call, h.cacheTimeout, cgLogger)
	cgCache.connections[cgCache.currID] = conn
	connWG := conn.open()
	cgCache.currID++
	cgCache.incNumOpenConns(1)
	cgCache.extMutex.Unlock()
	atomic.AddInt32(&h.loadShutdownRef, -1)
	h.hostMetrics.Increment(load.HostMetricNumOpenConns)
	connWG.Wait()
	h.hostMetrics.Decrement(load.HostMetricNumOpenConns)
	return nil
}
func (h *OutputHost) processAcks(ackIds []string, isNack bool) (invalidIDs []string) {
	for _, ackIDStr := range ackIds {
		ackID := AckID(ackIDStr)
		ackIDObj, err := common.AckIDFromString(string(ackID))
		if err != nil {
			h.logger.WithFields(bark.Fields{
				common.TagErr:   fmt.Sprintf("%v", err),
				common.TagAckID: ackIDStr,
			}).Error("error parsing the ackID")
			invalidIDs = append(invalidIDs, ackIDStr)
			if !isNack {
				h.m3Client.IncCounter(metrics.AckMessagesScope, metrics.OutputhostMessageAckFailures)
			} else {
				h.m3Client.IncCounter(metrics.AckMessagesScope, metrics.OutputhostMessageNackFailures)
			}
			continue
		}
		sessionID, ackMgrID, seqNum := ackIDObj.MutatedID.DeconstructCombinedID()
		if sessionID != h.sessionID {
			h.logger.WithFields(bark.Fields{
				"OutputSessionID": fmt.Sprintf("%v", h.sessionID),
				"ackSessionID":    fmt.Sprintf("%v", sessionID),
			}).Error("ackID intended for a different outputhost")
			h.m3Client.IncCounter(metrics.AckMessagesScope, metrics.OutputhostMessageDiffSession)
			continue
		}
		ackMgr := h.getAckMgr(ackMgrID)
		if ackMgr == nil {
			h.logger.WithFields(bark.Fields{
				common.TagAckID: ackIDStr,
				`ackMgrID`:      ackMgrID,
			}).Info("processAcks could not get ack manager (probably extent is consumed)")
			h.m3Client.IncCounter(metrics.AckMessagesScope, metrics.OutputhostMessageNoAckManager)
			continue
		}
		if err = ackMgr.acknowledgeMessage(ackID, ackIndex(seqNum), ackIDObj.Address, isNack); err == nil {
			continue
		} else {
			h.logger.WithFields(bark.Fields{
				common.TagErr:   fmt.Sprintf("%v", err),
				common.TagAckID: ackIDStr,
			}).Warn("processAcks ack manager returned error")
			h.m3Client.IncCounter(metrics.AckMessagesScope, metrics.OutputhostMessageAckManagerError)
		}
	}
	return
}
func (h *OutputHost) AckMessages(ctx thrift.Context, ackRequest *cherami.AckMessagesRequest) error {
	sw := h.m3Client.StartTimer(metrics.AckMessagesScope, metrics.OutputhostLatencyTimer)
	defer sw.Stop()
	h.m3Client.IncCounter(metrics.AckMessagesScope, metrics.OutputhostRequests)
	invalidAckIds := h.processAcks(ackRequest.AckIds, false  )
	invalidNackIds := h.processAcks(ackRequest.NackIds, true  )
	sw.Stop()
	if len(invalidAckIds) > 0 || len(invalidNackIds) > 0 {
		h.logger.WithFields(bark.Fields{
			"inv_ackid_count":  strconv.Itoa(len(invalidAckIds)),
			"inv_nackid_count": strconv.Itoa(len(invalidNackIds)),
		}).Warn("Invalid AckId or NackId returned")
		return &cherami.InvalidAckIdError{
			Message: "some ackIds not recognized",
			AckIds:  invalidAckIds,
			NackIds: invalidNackIds,
		}
	}
	return nil
}
func (h *OutputHost) ReceiveMessageBatch(ctx thrift.Context, request *cherami.ReceiveMessageBatchRequest) (*cherami.ReceiveMessageBatchResult_, error) {
	sw := h.m3Client.StartTimer(metrics.ReceiveMessageBatchOutputHostScope, metrics.OutputhostLatencyTimer)
	defer sw.Stop()
	h.m3Client.IncCounter(metrics.ReceiveMessageBatchOutputHostScope, metrics.OutputhostRequests)
	path := request.GetDestinationPath()
	cgName := request.GetConsumerGroupName()
	count := request.GetMaxNumberOfMessages()
	timeoutTime := time.Now().Add(time.Duration(request.GetReceiveTimeout()) * time.Second)
	deadline, hasDeadline := ctx.Deadline()
	if hasDeadline && deadline.Before(timeoutTime) {
		timeoutTime = deadline
	}
	lclLg := h.logger.WithFields(bark.Fields{
		common.TagDstPth: common.FmtDstPth(path),
		common.TagCnsPth: common.FmtCnsPth(cgName),
	})
	cgDesc, errC, err := h.getConsumerGroup(ctx, path, cgName, false  )
	if err != nil || cgDesc == nil || len(cgDesc.GetDestinationUUID()) == 0 || len(cgDesc.GetConsumerGroupUUID()) == 0 {
		lclLg.WithField(common.TagErr, err).Error(`Error translating to UUID with error`)
		h.incFailureCounter(metrics.ReceiveMessageBatchOutputHostScope, errC)
		return nil, err
	}
	lclLg = lclLg.WithField(common.TagCnsm, cgDesc.GetConsumerGroupUUID())
	cgCache, err := h.createAndLoadCGCache(ctx, *cgDesc, path, lclLg)
	defer atomic.AddInt32(&h.loadShutdownRef, -1)
	if err != nil {
		lclLg.WithField(common.TagErr, err).Error(`unable to load consumer group with error`)
		h.m3Client.IncCounter(metrics.ReceiveMessageBatchOutputHostScope, metrics.OutputhostFailures)
		return nil, err
	}
	cgCache.updateLastDisconnectTime()
	cgCache.connsWG.Add(1)  
	defer cgCache.connsWG.Done()
	res := cherami.NewReceiveMessageBatchResult_()
	msgCacheWriteTicker := common.NewTimer(msgCacheWriteTimeout)
	defer msgCacheWriteTicker.Stop()
	deliver := func(msg *cherami.ConsumerMessage, msgCacheCh chan cacheMsg) {
		res.Messages = append(res.Messages, msg)
		msgCacheWriteTicker.Reset(msgCacheWriteTimeout)
		select {
		case msgCacheCh <- cacheMsg{msg: msg, connID: -1}:
		case <-cgCache.closeChannel:
			lclLg.WithField(common.TagAckID, common.FmtAckID(msg.GetAckId())).
				Error("outputhost: Unable to write the message to the cache because cg cache closing")
		case <-msgCacheWriteTicker.C:
			lclLg.WithField(common.TagAckID, common.FmtAckID(msg.GetAckId())).
				Error("outputhost: Unable to write the message to the cache because time out")
			h.m3Client.IncCounter(metrics.ReceiveMessageBatchOutputHostScope, metrics.OutputhostReceiveMsgBatchWriteToMsgCacheTimeout)
		}
	}
	firstResultTimer := common.NewTimer(timeoutTime.Sub(time.Now()))
	defer firstResultTimer.Stop()
	select {
	case msg, ok := <-cgCache.msgsRedeliveryCh:
		if !ok {
			h.m3Client.IncCounter(metrics.ReceiveMessageBatchOutputHostScope, metrics.OutputhostFailures)
			return nil, ErrCgUnloaded
		}
		deliver(msg, cgCache.msgCacheRedeliveredCh)
	case msg, ok := <-cgCache.msgsCh:
		if !ok {
			h.m3Client.IncCounter(metrics.ReceiveMessageBatchOutputHostScope, metrics.OutputhostFailures)
			return nil, ErrCgUnloaded
		}
		deliver(msg, cgCache.msgCacheCh)
	case <-firstResultTimer.C:
		exception := cherami.NewTimeoutError()
		exception.Message = "Timeout getting next message, probably no message available at the moment."
		h.m3Client.IncCounter(metrics.ReceiveMessageBatchOutputHostScope, metrics.OutputhostLongPollingTimeOut)
		return nil, exception
	}
	moreResultTimer := common.NewTimer(timeoutTime.Sub(time.Now()))
	defer moreResultTimer.Stop()
MORERESULTLOOP:
	for remaining := count - 1; remaining > 0; remaining-- {
		select {
		case <-moreResultTimer.C:
			break MORERESULTLOOP
		default:
			select {
			case msg, ok := <-cgCache.msgsRedeliveryCh:
				if !ok {
					break MORERESULTLOOP
				}
				deliver(msg, cgCache.msgCacheRedeliveredCh)
			case msg, ok := <-cgCache.msgsCh:
				if !ok {
					break MORERESULTLOOP
				}
				deliver(msg, cgCache.msgCacheCh)
			case <-moreResultTimer.C:
				break MORERESULTLOOP
			}
		}
	}
	h.m3Client.AddCounter(metrics.ReceiveMessageBatchOutputHostScope, metrics.OutputhostMessageSent, int64(len(res.Messages)))
	cgCache.consumerM3Client.AddCounter(metrics.ReceiveMessageBatchOutputHostCGScope, metrics.OutputhostCGMessageSent, int64(len(res.Messages)))
	return res, nil
}
func (h *OutputHost) ConsumerGroupsUpdated(ctx thrift.Context, request *admin.ConsumerGroupsUpdatedRequest) (err error) {
	defer atomic.AddInt32(&h.loadShutdownRef, -1)
	sw := h.m3Client.StartTimer(metrics.ConsumerGroupsUpdatedScope, metrics.OutputhostLatencyTimer)
	defer sw.Stop()
	h.m3Client.IncCounter(metrics.ConsumerGroupsUpdatedScope, metrics.OutputhostRequests)
	if atomic.AddInt32(&h.loadShutdownRef, 1) <= 0 {
		h.logger.Error("not loading the CG cache outputHost already shutdown")
		h.m3Client.IncCounter(metrics.ConsumerGroupsUpdatedScope, metrics.OutputhostFailures)
		return ErrHostShutdown
	}
	var intErr error
	updateUUID := request.GetUpdateUUID()
	h.logger.WithField(common.TagUpdateUUID, updateUUID).Debug("ConsumerGroupsUpdated: processing notification")
	for _, req := range request.Updates {
		cgUUID := req.GetConsumerGroupUUID()
		h.cgMutex.RLock()
		cg, ok := h.cgCache[cgUUID]
		h.cgMutex.RUnlock()
		if ok {
			notifyType := req.GetType()
			h.logger.WithFields(bark.Fields{
				common.TagCnsm:       common.FmtCnsm(cgUUID),
				common.TagUpdateUUID: updateUUID,
				`notifyType`:         notifyType,
			}).Info("ConsumerGroupsUpdated: processing notification for consumer-group")
			switch notifyType {
			case admin.NotificationType_CLIENT:
				cg.reconfigureClients(updateUUID)
			case admin.NotificationType_HOST:
				intErr = cg.refreshCgCacheNoLock(ctx)
			case admin.NotificationType_ALL:
				intErr = cg.refreshCgCacheNoLock(ctx)
				if intErr == nil {
					cg.reconfigureClients(updateUUID)
				}
			default:
				h.logger.WithFields(bark.Fields{
					common.TagCnsm:       common.FmtCnsm(cgUUID),
					common.TagUpdateUUID: updateUUID,
					`notifyType`:         notifyType,
				}).Error(`ConsumerGroupsUpdated: unknown notification type`)
				intErr = &cherami.BadRequestError{}
			}
		} else {
			h.logger.WithFields(bark.Fields{
				common.TagCnsm:       common.FmtCnsm(cgUUID),
				common.TagUpdateUUID: updateUUID,
				`notifyType`:         req.GetType(),
			}).Error("ConsumerGroupsUpdated: cgCache does not exist")
			intErr = &cherami.EntityNotExistsError{}
		}
		if intErr != nil {
			err = intErr
			h.m3Client.IncCounter(metrics.ConsumerGroupsUpdatedScope, metrics.OutputhostFailures)
		}
	}
	return err
}
func (h *OutputHost) unloadAll() {
	h.cgMutex.Lock()
	for cgUUID, cgCache := range h.cgCache {
		h.logger.WithField(common.TagCnsm, cgUUID).Info("outputhost: closing streams for consumer group.")
		go cgCache.unloadConsumerGroupCache()
	}
	h.cgMutex.Unlock()
}
func (h *OutputHost) Shutdown() {
	atomic.AddInt32(&h.loadShutdownRef, -0x80000000)
	for atomic.LoadInt32(&h.loadShutdownRef) > -0x80000000 {
		time.Sleep(time.Second)
	}
	h.unloadAll()
	h.shutdownWG.Wait()
	close(h.shutdownCh)
}
func (h *OutputHost) Start(thriftService []thrift.TChanServer) {
	h.SCommon.Start(thriftService)
	h.hostIDHeartbeater = common.NewHostIDHeartbeater(h.metaClient, h.GetHostUUID(), h.GetHostPort(), h.GetHostName(), h.logger)
	h.logger = h.logger.WithField(common.TagHostIP, common.FmtHostIP(h.SCommon.GetHostPort()))
	h.loadReporter = h.GetLoadReporterDaemonFactory().CreateReporter(hostLoadReportingInterval, h, h.logger)
	h.loadReporter.Start()
	h.hostIDHeartbeater.Start()
	h.cfgMgr.Start()
	go h.manageCgCache()
}
func (h *OutputHost) manageCgCache() {
	for {
		select {
		case cgUUID := <-h.unloadCacheCh:
			h.cgMutex.Lock()
			h.logger.WithField(common.TagCnsm, common.FmtCnsm(cgUUID)).
				Info("unloading consumer group")
			delete(h.cgCache, cgUUID)
			h.cgMutex.Unlock()
		case ackloadMsg := <-h.ackMgrLoadCh:
			h.ackMgrMutex.Lock()
			h.ackMgrMap[ackloadMsg.ackMgrID] = ackloadMsg.ackMgr
			h.ackMgrMutex.Unlock()
		case ackMgrID := <-h.ackMgrUnloadCh:
			h.ackMgrMutex.Lock()
			delete(h.ackMgrMap, ackMgrID)
			h.ackMgrMutex.Unlock()
		case <-h.shutdownCh:
			return
		}
	}
}
func (h *OutputHost) Report(reporter common.LoadReporter) {
	msgsOut := h.hostMetrics.GetAndReset(load.HostMetricMsgsOut)
	bytesOut := h.hostMetrics.GetAndReset(load.HostMetricBytesOut)
	numConns := h.hostMetrics.Get(load.HostMetricNumOpenConns)
	numExtents := h.hostMetrics.Get(load.HostMetricNumOpenExtents)
	hostMetrics := controller.NodeMetrics{
		NumberOfActiveExtents:   common.Int64Ptr(numExtents),
		NumberOfConnections:     common.Int64Ptr(numConns),
		OutgoingMessagesCounter: common.Int64Ptr(msgsOut),
		OutgoingBytesCounter:    common.Int64Ptr(bytesOut),
	}
	reporter.ReportHostMetric(hostMetrics)
	h.m3Client.UpdateGauge(metrics.ConsConnectionStreamScope, metrics.OutputhostConsConnection, numConns)
}
func (h *OutputHost) Stop() {
	h.hostIDHeartbeater.Stop()
	h.loadReporter.Stop()
	h.cfgMgr.Stop()
	h.SCommon.Stop()
}
func (h *OutputHost) RegisterWSHandler() *http.ServeMux {
	mux := http.NewServeMux()
	mux.HandleFunc(fmt.Sprintf(ccommon.HTTPHandlerPattern, ccommon.EndpointOpenConsumerStream), h.OpenConsumerStreamHandler)
	return mux
}
func (h *OutputHost) SetFrontendClient(frontendClient ccherami.TChanBFrontend) {
	h.frontendClient = frontendClient
}
func NewOutputHost(
	serviceName string,
	sVice common.SCommon,
	metadataClient metadata.TChanMetadataService,
	frontendClient ccherami.TChanBFrontend,
	opts *OutOptions,
	kafkaCfg configure.CommonKafkaConfig,
) (*OutputHost, []thrift.TChanServer) {
	deploymentName := sVice.GetConfig().GetDeploymentName()
	bs := OutputHost{
		logger:         (sVice.GetConfig().GetLogger()).WithFields(bark.Fields{common.TagOut: common.FmtOut(sVice.GetHostUUID()), common.TagDplName: common.FmtDplName(deploymentName)}),
		SCommon:        sVice,
		frontendClient: frontendClient,
		cgCache:        make(map[string]*consumerGroupCache),
		cacheTimeout:   defaultIdleTimeout,
		shutdownCh:     make(chan struct{}),
		unloadCacheCh:  make(chan string, defaultUnloadChSize),
		ackMgrMap:      make(map[uint32]*ackManager),
		ackMgrLoadCh:   make(chan ackMgrLoadMsg, defaultAckMgrMapChSize),
		ackMgrUnloadCh: make(chan uint32, defaultAckMgrMapChSize),
		ackMgrIDGen:    common.NewHostAckIDGenerator(defaultAckMgrIDStartFrom),
		hostMetrics:    load.NewHostMetrics(),
		kafkaCfg:       kafkaCfg,
	}
	sarama.Logger = NewSaramaLoggerFromBark(bs.logger, `sarama`)
	bs.sessionID = common.UUIDToUint16(sVice.GetHostUUID())
	bs.m3Client = metrics.NewClient(sVice.GetMetricsReporter(), metrics.Outputhost)
	if opts != nil {
		if opts.CacheIdleTimeout != 0 {
			bs.cacheTimeout = opts.CacheIdleTimeout
		}
		if opts.KStreamFactory != nil {
			bs.kafkaStreamFactory = opts.KStreamFactory
		}
	}
	bs.metaClient = mm.NewMetadataMetricsMgr(metadataClient, bs.m3Client, bs.logger)
	bs.dClient = sVice.GetDConfigClient()
	bs.dynamicConfigManage()
	bs.cfgMgr = newConfigManager(metadataClient, bs.logger)
	common.StartEKG(bs.logger)
	thisOutputHost = &bs
	return &bs, []thrift.TChanServer{cherami.NewTChanBOutServer(&bs), admin.NewTChanOutputHostAdminServer(&bs)}
}