func (ncoord *NsqdCoordinator) forceCleanTopicData(topicName string, partition int) *CoordErr {
	ncoord.forceCleanTopicCommitLogs(topicName, partition)
	localErr := ncoord.localNsqd.ForceDeleteTopicData(topicName, partition)
	if localErr != nil {
		if !os.IsNotExist(localErr) {
			coordLog.Infof("delete topic %v-%v local data failed : %v", topicName, partition, localErr)
			return &CoordErr{localErr.Error(), RpcCommonErr, CoordLocalErr}
		}
	}
	return nil
}
do not hold any coordinator lock outside, since we need delete local topic which will wait pub loop quit.
pub loop may hold the coordinator lock.
package consistence
import (
	"bytes"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"os"
	"path"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/youzan/nsq/internal/protocol"
	"github.com/youzan/nsq/nsqd"
)
const (
	MAX_WRITE_RETRY             = 8
	maxWriteWaitTimeout         = time.Second * 5
	MAX_CATCHUP_RETRY           = 5
	MAX_LOG_PULL                = 10000
	MAX_LOG_PULL_BYTES          = 1024 * 1024 * 32
	MAX_CATCHUP_RUNNING         = 8
	API_BACKUP_DELAYED_QUEUE_DB = "/delayqueue/backupto"
)
var (
	MaxRetryWait                = time.Millisecond * 200
	ForceFixLeaderData          = false
	MaxTopicRetentionSizePerDay = int64(1024 * 1024 * 1024 * 16)
	flushTicker                 = time.Second * 2
	sleepMsBetweenLogSyncPull   = int32(0)
)
func ChangeSleepMsBetweenLogSyncPull(ms int) {
	atomic.StoreInt32(&sleepMsBetweenLogSyncPull, int32(ms))
}
var testCatchupPausedPullLogs int32
func GetTopicPartitionFileName(topic string, partition int, suffix string) string {
	var tmpbuf bytes.Buffer
	tmpbuf.WriteString(topic)
	tmpbuf.WriteString("_")
	tmpbuf.WriteString(strconv.Itoa(partition))
	tmpbuf.WriteString(suffix)
	return tmpbuf.String()
}
func GetTopicPartitionBasePath(rootPath string, topic string, partition int) string {
	return filepath.Join(rootPath, topic)
}
type TopicPartitionID struct {
	TopicName      string
	TopicPartition int
}
func (ncoord *TopicPartitionID) String() string {
	return ncoord.TopicName + "-" + strconv.Itoa(ncoord.TopicPartition)
}
type ILocalLogQueue interface {
	IsDataNeedFix() bool
	SetDataFixState(bool)
	ForceFlush()
	ResetBackendEndNoLock(nsqd.BackendOffset, int64) error
	ResetBackendWithQueueStartNoLock(int64, int64) error
	GetDiskQueueSnapshot(checkCommit bool) *nsqd.DiskQueueSnapshot
	TotalMessageCnt() uint64
	TotalDataSize() int64
	TryFixQueueEnd(nsqd.BackendOffset, int64) error
	CheckDiskQueueReadToEndOK(offset int64, seekCnt int64, endOffset nsqd.BackendOffset) error
	PutRawDataOnReplica(rawData []byte, offset nsqd.BackendOffset, checkSize int64, msgNum int32) (nsqd.BackendQueueEnd, error)
	PutMessageOnReplica(msgs *nsqd.Message, offset nsqd.BackendOffset, checkSize int64) (nsqd.BackendQueueEnd, error)
	TryCleanOldData(retentionSize int64, noRealClean bool, maxCleanOffset nsqd.BackendOffset) (nsqd.BackendQueueEnd, error)
}
func getCommitLogAndLocalLogQ(tcData *coordData, localTopic *nsqd.Topic,
	fromDelayedQueue bool) (ILocalLogQueue, *TopicCommitLogMgr) {
	var localLogQ ILocalLogQueue
	localLogQ = localTopic
	logMgr := tcData.logMgr
	if fromDelayedQueue {
		if tcData.topicInfo.OrderedMulti {
			return nil, nil
		}
		dq := localTopic.GetDelayedQueue()
		if dq == nil {
			return nil, nil
		}
		localLogQ = dq
		logMgr = tcData.delayedLogMgr
	}
	return localLogQ, logMgr
}
func getOrCreateCommitLogAndLocalLogQ(tcData *coordData, localTopic *nsqd.Topic,
	fromDelayedQueue bool) (ILocalLogQueue, *TopicCommitLogMgr, *CoordErr) {
	var localLogQ ILocalLogQueue
	localLogQ = localTopic
	logMgr := tcData.logMgr
	if fromDelayedQueue {
		if tcData.topicInfo.OrderedMulti {
			return nil, nil, ErrLocalDelayedQueueMissing
		}
		dq, _ := localTopic.GetOrCreateDelayedQueueNoLock(tcData.delayedLogMgr)
		if dq == nil {
			return nil, nil, ErrLocalDelayedQueueMissing
		}
		localLogQ = dq
		logMgr = tcData.delayedLogMgr
	}
	return localLogQ, logMgr, nil
}
func maybeInitDelayedQ(tcData *coordData, localTopic *nsqd.Topic) error {
	if atomic.LoadInt32(&nsqd.EnableDelayedQueue) != 1 {
		return nil
	}
	if !tcData.topicInfo.OrderedMulti && tcData.delayedLogMgr != nil {
		localTopic.Lock()
		_, err := localTopic.GetOrCreateDelayedQueueNoLock(tcData.delayedLogMgr)
		localTopic.Unlock()
		return err
	}
	return nil
}
type NsqdCoordinator struct {
	clusterKey             string
	leadership             NSQDLeadership
	lookupMutex            sync.Mutex
	lookupLeader           NsqLookupdNodeInfo
	lookupRemoteCreateFunc nsqlookupRemoteProxyCreateFunc
	lookupRemoteClients    map[string]INsqlookupRemoteProxy
	topicCoords            map[string]map[int]*TopicCoordinator
	coordMutex             sync.RWMutex
	myNode                 NsqdNodeInfo
	rpcClientMutex         sync.Mutex
	nsqdRpcClients         sync.Map
	flushNotifyChan        chan TopicPartitionID
	stopChan               chan struct{}
	dataRootPath           string
	localNsqd              *nsqd.NSQD
	rpcServer              *NsqdCoordRpcServer
	grpcServer             *nsqdCoordGRpcServer
	tryCheckUnsynced       chan bool
	wg                     sync.WaitGroup
	enableBenchCost        bool
	stopping               int32
	catchupRunning         int32
}
func NewNsqdCoordinator(cluster, ip, tcpport, rpcport, httpport, extraID string, rootPath string, nsqd *nsqd.NSQD) *NsqdCoordinator {
	nodeInfo := NsqdNodeInfo{
		NodeIP:   ip,
		TcpPort:  tcpport,
		RpcPort:  rpcport,
		HttpPort: httpport,
	}
	nodeInfo.ID = GenNsqdNodeID(&nodeInfo, extraID)
	nsqdCoord := &NsqdCoordinator{
		clusterKey:             cluster,
		leadership:             nil,
		topicCoords:            make(map[string]map[int]*TopicCoordinator),
		myNode:                 nodeInfo,
		flushNotifyChan:        make(chan TopicPartitionID, 2),
		stopChan:               make(chan struct{}),
		dataRootPath:           rootPath,
		localNsqd:              nsqd,
		tryCheckUnsynced:       make(chan bool, 1),
		lookupRemoteCreateFunc: NewNsqLookupRpcClient,
		lookupRemoteClients:    make(map[string]INsqlookupRemoteProxy),
	}
	if nsqdCoord.leadership != nil {
		nsqdCoord.leadership.InitClusterID(nsqdCoord.clusterKey)
	}
	nsqdCoord.rpcServer = NewNsqdCoordRpcServer(nsqdCoord, rootPath)
	nsqdCoord.grpcServer = NewNsqdCoordGRpcServer(nsqdCoord, rootPath)
	return nsqdCoord
}
func (ncoord *NsqdCoordinator) GetMyID() string {
	return ncoord.myNode.GetID()
}
func (ncoord *NsqdCoordinator) SetLeadershipMgr(l NSQDLeadership) {
	ncoord.leadership = l
	if ncoord.leadership != nil {
		ncoord.leadership.InitClusterID(ncoord.clusterKey)
	}
}
func (ncoord *NsqdCoordinator) getExistRpcClient(nid string) (*NsqdRpcClient, bool) {
	v, ok := ncoord.nsqdRpcClients.Load(nid)
	if !ok {
		return nil, false
	}
	c, ok := v.(*NsqdRpcClient)
	if ok && !c.ShouldRemoved() {
		return c, ok
	}
	return nil, false
}
func (ncoord *NsqdCoordinator) cleanRpcClients(all bool) {
	removed := make([]string, 0)
	ncoord.rpcClientMutex.Lock()
	ncoord.nsqdRpcClients.Range(func(key interface{}, value interface{}) bool {
		if all {
			removed = append(removed, key.(string))
			return true
		}
		c, ok := value.(*NsqdRpcClient)
		if ok && c.ShouldRemoved() {
			removed = append(removed, key.(string))
		}
		return true
	})
	for _, nid := range removed {
		v, ok := ncoord.nsqdRpcClients.Load(nid)
		if ok {
			c := v.(*NsqdRpcClient)
			ncoord.nsqdRpcClients.Delete(nid)
			c.Close()
		}
	}
	ncoord.rpcClientMutex.Unlock()
}
func (ncoord *NsqdCoordinator) acquireRpcClient(nid string) (*NsqdRpcClient, *CoordErr) {
	c, ok := ncoord.getExistRpcClient(nid)
	if ok {
		return c, nil
	}
	ncoord.rpcClientMutex.Lock()
	defer ncoord.rpcClientMutex.Unlock()
	v, ok := ncoord.nsqdRpcClients.Load(nid)
	if ok {
		c, ok = v.(*NsqdRpcClient)
		if ok && c.ShouldRemoved() {
			coordLog.Infof("rpc removing removed client: %v", nid)
			ncoord.nsqdRpcClients.Delete(nid)
			c.Close()
			ok = false
		}
	}
	var err error
	if !ok {
		addr := ExtractRpcAddrFromID(nid)
		c, err = NewNsqdRpcClient(addr, RPC_TIMEOUT_SHORT)
		if err != nil {
			return nil, NewCoordErr(err.Error(), CoordNetErr)
		}
		ncoord.nsqdRpcClients.Store(nid, c)
	}
	return c, nil
}
func (ncoord *NsqdCoordinator) Start() error {
	ncoord.wg.Add(1)
	go ncoord.watchNsqLookupd()
	start := time.Now()
	for {
		ncoord.lookupMutex.Lock()
		l := ncoord.lookupLeader
		ncoord.lookupMutex.Unlock()
		if l.GetID() != "" {
			break
		}
		time.Sleep(time.Second)
		coordLog.Infof("waiting for lookupd ...")
		if time.Now().Sub(start) > time.Second*30 {
			panic("no lookupd found while starting nsqd coordinator")
		}
	}
	err := ncoord.loadLocalTopicData()
	if err != nil {
		close(ncoord.stopChan)
		ncoord.rpcServer.stop()
		return err
	}
	realAddr, err := ncoord.rpcServer.start(ncoord.myNode.NodeIP, ncoord.myNode.RpcPort)
	if err != nil {
		return err
	}
	_, realRpcPort, _ := net.SplitHostPort(realAddr)
	ncoord.myNode.RpcPort = realRpcPort
	if ncoord.leadership != nil {
		err := ncoord.leadership.RegisterNsqd(&ncoord.myNode)
		if err != nil {
			coordLog.Warningf("failed to register nsqd coordinator: %v", err)
			return err
		}
	}
	ncoord.wg.Add(1)
	go ncoord.checkForUnsyncedTopics()
	ncoord.wg.Add(1)
	go ncoord.periodFlushCommitLogs()
	ncoord.wg.Add(1)
	go ncoord.checkAndCleanOldData()
	return nil
}
func (ncoord *NsqdCoordinator) Stop() {
	if atomic.LoadInt32(&ncoord.stopping) == 1 {
		return
	}
	ncoord.prepareLeavingCluster()
	close(ncoord.stopChan)
	ncoord.rpcServer.stop()
	ncoord.rpcServer = nil
	ncoord.cleanRpcClients(true)
	ncoord.wg.Wait()
	ncoord.lookupMutex.Lock()
	for _, c := range ncoord.lookupRemoteClients {
		if c != nil {
			c.Close()
		}
	}
	ncoord.lookupMutex.Unlock()
}
func doLogQClean(tcData *coordData, localTopic *nsqd.Topic, retentionSize int64, fromDelayedQueue bool) {
	localLogQ, logMgr := getCommitLogAndLocalLogQ(tcData, localTopic, fromDelayedQueue)
	if localLogQ == nil || logMgr == nil {
		return
	}
	cleanEndInfo, err := localLogQ.TryCleanOldData(retentionSize, true, 0)
	if err != nil {
		coordLog.Infof("failed to get clean end: %v", err)
	}
	if cleanEndInfo != nil {
		coordLog.Infof("topic %v try clean to : %v", tcData.topicInfo.GetTopicDesp(), cleanEndInfo)
		matchIndex, matchOffset, l, err := logMgr.SearchLogDataByMsgOffset(int64(cleanEndInfo.Offset()))
		if err != nil {
			coordLog.Infof("search log failed: %v", err)
			return
		}
		coordLog.Infof("clean commit log at : %v, %v, %v", matchIndex, matchOffset, l)
		if l.MsgOffset > int64(cleanEndInfo.Offset()) {
			coordLog.Infof("search log clean position exceed the clean end, something wrong")
			return
		}
		maxCleanOffset := cleanEndInfo.Offset()
		if l.MsgOffset < int64(cleanEndInfo.Offset()) {
			maxCleanOffset = nsqd.BackendOffset(l.MsgOffset)
		}
		oldMatchOffset := matchOffset
		matchOffset, err = logMgr.GetMaxAvailableCleanOffset(matchIndex, matchOffset)
		if err != nil {
			coordLog.Infof("clean commit log no more available clean at (%v-%v): %s", matchIndex, matchOffset, err)
			return
		}
		if matchOffset >= oldMatchOffset {
			matchOffset = oldMatchOffset
		} else {
			l, err = logMgr.GetCommitLogFromOffsetV2(matchIndex, matchOffset)
			if err != nil {
				coordLog.Infof("clean commit log failed at (%v-%v): %s", matchIndex, matchOffset, err)
				return
			}
			if l.MsgOffset < int64(maxCleanOffset) {
				maxCleanOffset = nsqd.BackendOffset(l.MsgOffset)
			}
		}
		coordLog.Infof("clean commit log relocated at : %v, %v, %v, %v", matchIndex, matchOffset, maxCleanOffset, l)
		err = logMgr.CleanOldData(matchIndex, matchOffset)
		if err != nil {
			coordLog.Infof("clean commit log err : %v", err)
		} else {
			retentionSize = 1024 * 512
			_, err := localLogQ.TryCleanOldData(retentionSize, false, maxCleanOffset)
			if err != nil {
				coordLog.Infof("failed to clean disk queue: %v", err)
			}
		}
	}
}
func (ncoord *NsqdCoordinator) GreedyCleanTopicOldData(localTopic *nsqd.Topic) error {
	tcData, err := ncoord.getTopicCoordData(localTopic.GetTopicName(), localTopic.GetTopicPart())
	if err != nil {
		return err.ToErrorType()
	}
	retentionDay := tcData.topicInfo.RetentionDay
	if retentionDay == 0 {
		retentionDay = int32(nsqd.DEFAULT_RETENTION_DAYS)
	}
	retentionSize := (MaxTopicRetentionSizePerDay / 16) * int64(retentionDay)
	doLogQClean(tcData, localTopic, retentionSize, false)
	doLogQClean(tcData, localTopic, retentionSize, true)
	return nil
}
func (ncoord *NsqdCoordinator) getAllCoords(tmpCoords map[string]map[int]*TopicCoordinator) {
	ncoord.coordMutex.RLock()
	for name, tc := range ncoord.topicCoords {
		coords, ok := tmpCoords[name]
		if !ok {
			coords = make(map[int]*TopicCoordinator)
			tmpCoords[name] = coords
		}
		for pid, tpc := range tc {
			coords[pid] = tpc
		}
	}
	ncoord.coordMutex.RUnlock()
}
func (ncoord *NsqdCoordinator) checkAndCleanOldData() {
	defer ncoord.wg.Done()
	ticker := time.NewTicker(time.Minute * 10)
	defer ticker.Stop()
	doCheckAndCleanOld := func(checkRetentionDay bool) {
		tmpCoords := make(map[string]map[int]*TopicCoordinator)
		ncoord.getAllCoords(tmpCoords)
		for _, tc := range tmpCoords {
			for _, tpc := range tc {
				tcData := tpc.GetData()
				localTopic, err := ncoord.localNsqd.GetExistingTopic(tcData.topicInfo.Name, tcData.topicInfo.Partition)
				if err != nil {
					coordLog.Infof("failed to get local topic: %v", tcData.topicInfo.GetTopicDesp())
					continue
				}
				retentionDay := tcData.topicInfo.RetentionDay
				if retentionDay == 0 {
					retentionDay = int32(nsqd.DEFAULT_RETENTION_DAYS)
				}
				retentionSize := MaxTopicRetentionSizePerDay * int64(retentionDay)
				if checkRetentionDay {
					retentionSize = 0
				}
				doLogQClean(tcData, localTopic, retentionSize, false)
				doLogQClean(tcData, localTopic, retentionSize, true)
			}
		}
	}
	for {
		select {
		case <-ticker.C:
			now := time.Now()
			if now.Hour() > 2 && now.Hour() < 6 {
				coordLog.Infof("check and clean at time: %v", now)
				doCheckAndCleanOld(true)
			}
			doCheckAndCleanOld(false)
		case <-ncoord.stopChan:
			return
		}
	}
}
func (ncoord *NsqdCoordinator) periodFlushCommitLogs() {
	const FLUSH_DISTANCE = 4
	tmpCoords := make(map[string]map[int]*TopicCoordinator)
	syncCounter := 0
	defer ncoord.wg.Done()
	flushTicker := time.NewTicker(flushTicker)
	defer flushTicker.Stop()
	doFlush := func() {
		syncCounter++
		ncoord.getAllCoords(tmpCoords)
		flushAll := syncCounter%30 == 0
		matchCnt := syncCounter % FLUSH_DISTANCE
		for _, tc := range tmpCoords {
			for pid, tpc := range tc {
				tcData := tpc.GetData()
				if tcData.GetLeader() == ncoord.myNode.GetID() || flushAll {
					localTopic, err := ncoord.localNsqd.GetExistingTopic(tcData.topicInfo.Name, tcData.topicInfo.Partition)
					if err != nil {
						coordLog.Infof("no local topic: %v", tcData.topicInfo.GetTopicDesp())
					} else {
						if flushAll || ((pid+1)%FLUSH_DISTANCE == matchCnt) {
							localTopic.ForceFlush()
							tcData.flushCommitLogs()
						}
					}
				}
				if !tpc.IsExiting() && tcData.GetLeader() == ncoord.myNode.GetID() {
					syncChList := !tpc.IsWriteDisabled() && flushAll
					if ((pid+1)%FLUSH_DISTANCE) == matchCnt || syncChList {
						ncoord.trySyncTopicChannels(tcData, syncChList, false)
					}
				}
				delete(tc, pid)
			}
		}
	}
	for {
		select {
		case <-flushTicker.C:
			doFlush()
		case <-ncoord.stopChan:
			time.Sleep(time.Second)
			doFlush()
			return
		}
	}
}
func (ncoord *NsqdCoordinator) getLookupRemoteProxy() (INsqlookupRemoteProxy, *CoordErr) {
	ncoord.lookupMutex.Lock()
	l := ncoord.lookupLeader
	addr := net.JoinHostPort(l.NodeIP, l.RpcPort)
	c, ok := ncoord.lookupRemoteClients[addr]
	ncoord.lookupMutex.Unlock()
	if l.NodeIP == "" {
		return nil, NewCoordErr("missing lookup leader", CoordNetErr)
	}
	if ok && c != nil {
		return c, nil
	}
	c, err := ncoord.lookupRemoteCreateFunc(addr, RPC_TIMEOUT_FOR_LOOKUP)
	if err == nil {
		ncoord.lookupMutex.Lock()
		ncoord.lookupRemoteClients[addr] = c
		if len(ncoord.lookupRemoteClients) > 3 {
			for k := range ncoord.lookupRemoteClients {
				if k == addr {
					continue
				}
				delete(ncoord.lookupRemoteClients, k)
			}
		}
		ncoord.lookupMutex.Unlock()
		return c, nil
	}
	coordLog.Infof("get lookup remote %v failed: %v", l, err)
	return c, NewCoordErr(err.Error(), CoordNetErr)
}
func (ncoord *NsqdCoordinator) GetCurrentLookupd() NsqLookupdNodeInfo {
	ncoord.lookupMutex.Lock()
	defer ncoord.lookupMutex.Unlock()
	return ncoord.lookupLeader
}
func (ncoord *NsqdCoordinator) GetAllLookupdNodes() ([]NsqLookupdNodeInfo, error) {
	return ncoord.leadership.GetAllLookupdNodes()
}
func (ncoord *NsqdCoordinator) watchNsqLookupd() {
	nsqlookupLeaderChan := make(chan *NsqLookupdNodeInfo, 1)
	if ncoord.leadership != nil {
		go ncoord.leadership.WatchLookupdLeader(nsqlookupLeaderChan, ncoord.stopChan)
	}
	defer ncoord.wg.Done()
	for {
		select {
		case n, ok := <-nsqlookupLeaderChan:
			if !ok {
				return
			}
			ncoord.lookupMutex.Lock()
			if n.GetID() != ncoord.lookupLeader.GetID() {
				coordLog.Infof("on nsqd: lookup leader changed from %v to %v", ncoord.lookupLeader, *n)
				ncoord.lookupLeader = *n
			}
			ncoord.lookupMutex.Unlock()
		}
	}
}
func (ncoord *NsqdCoordinator) checkLocalTopicMagicCode(topicInfo *TopicPartitionMetaInfo, tryFix bool) error {
	removedPath, err := ncoord.localNsqd.CheckMagicCode(topicInfo.Name, topicInfo.Partition, topicInfo.MagicCode, topicInfo.Ext, tryFix)
	if err != nil {
		coordLog.Infof("check magic code error: %v", err)
		return err
	}
	if removedPath != "" {
		basepath := GetTopicPartitionBasePath(ncoord.dataRootPath, topicInfo.Name, topicInfo.Partition)
		tmpLogMgr, err := InitTopicCommitLogMgr(topicInfo.Name, topicInfo.Partition, basepath, 1)
		if err != nil {
			coordLog.Warningf("topic %v failed to init tmp log manager: %v", topicInfo.GetTopicDesp(), err)
		} else {
			tmpLogMgr.MoveTo(removedPath)
			tmpLogMgr.Delete()
		}
	}
	return nil
}