func (nlcoord *NsqLookupCoordinator) Start() error {
	if nlcoord.leadership != nil {
		err := nlcoord.leadership.Register(&nlcoord.myNode)
		if err != nil {
			coordLog.Warningf("failed to register nsqlookup coordinator: %v", err)
			return err
		}
	}
	nlcoord.wg.Add(1)
	go nlcoord.handleLeadership()
	go nlcoord.nsqlookupRpcServer.start(nlcoord.myNode.NodeIP, nlcoord.myNode.RpcPort)
	nlcoord.notifyNodesLookup()
	return nil
}
init and register to leader server
package consistence
import (
	"errors"
	"fmt"
	"math/rand"
	"net"
	"runtime"
	"strconv"
	"sync"
	"sync/atomic"
	"time"
	"github.com/cenkalti/backoff"
)
var (
	ErrAlreadyExist         = errors.New("already exist")
	ErrTopicNotCreated      = errors.New("topic is not created")
	ErrWaitingLeaderRelease = errors.New("leader session is still alive")
	ErrNotNsqLookupLeader   = errors.New("Not nsqlookup leader")
	ErrClusterUnstable      = errors.New("the cluster is unstable")
	errLookupExiting        = errors.New("lookup is exiting")
	ErrLeaderNodeLost           = NewCoordErr("leader node is lost", CoordTmpErr)
	ErrNodeNotFound             = NewCoordErr("node not found", CoordCommonErr)
	ErrLeaderElectionFail       = NewCoordErr("Leader election failed.", CoordElectionTmpErr)
	ErrNoLeaderCanBeElected     = NewCoordErr("No leader can be elected", CoordElectionTmpErr)
	ErrNodeUnavailable          = NewCoordErr("No node is available for topic", CoordTmpErr)
	ErrJoinISRInvalid           = NewCoordErr("Join ISR failed", CoordCommonErr)
	ErrJoinISRTimeout           = NewCoordErr("Join ISR timeout", CoordCommonErr)
	ErrWaitingJoinISR           = NewCoordErr("The topic is waiting node to join isr", CoordCommonErr)
	ErrLeaderSessionNotReleased = NewCoordErr("The topic leader session is not released", CoordElectionTmpErr)
	ErrTopicISRCatchupEnough    = NewCoordErr("the topic isr and catchup nodes are enough", CoordTmpErr)
	ErrClusterNodeRemoving      = NewCoordErr("the node is mark as removed", CoordTmpErr)
	ErrTopicNodeConflict        = NewCoordErr("the topic node info is conflicted", CoordElectionErr)
	ErrLeadershipServerUnstable = NewCoordErr("the leadership server is unstable", CoordTmpErr)
)
var (
	waitMigrateInterval          = time.Minute * 10
	waitEmergencyMigrateInterval = time.Second * 10
	waitRemovingNodeInterval     = time.Second * 30
	balanceInterval              = time.Second * 60
	doCheckInterval              = time.Second * 60
)
type JoinISRState struct {
	sync.Mutex
	waitingJoin      bool
	waitingSession   string
	waitingStart     time.Time
	readyNodes       map[string]struct{}
	doneChan         chan struct{}
	isLeadershipWait bool
}
func (s *JoinISRState) String() string {
	return fmt.Sprintf("join %v, session %v, start %s, readys %v, isLeader %v",
		s.waitingJoin, s.waitingSession, s.waitingStart, s.readyNodes, s.isLeadershipWait)
}
type RpcFailedInfo struct {
	nodeID    string
	topic     string
	partition int
	failTime  time.Time
}
func getOthersExceptLeader(topicInfo *TopicPartitionMetaInfo) []string {
	others := make([]string, 0, len(topicInfo.ISR)+len(topicInfo.CatchupList)-1)
	for _, n := range topicInfo.ISR {
		if n == topicInfo.Leader {
			continue
		}
		others = append(others, n)
	}
	others = append(others, topicInfo.CatchupList...)
	return others
}
type TopicNameInfo struct {
	TopicName      string
	TopicPartition int
}
type Options struct {
	BalanceStart int
	BalanceEnd   int
}
type NsqLookupCoordinator struct {
	clusterKey         string
	myNode             NsqLookupdNodeInfo
	leaderMu           sync.RWMutex
	leaderNode         NsqLookupdNodeInfo
	leadership         NSQLookupdLeadership
	nodesMutex         sync.RWMutex
	nsqdNodes          map[string]NsqdNodeInfo
	removingNodes      map[string]string
	nodesEpoch         int64
	rpcMutex           sync.RWMutex
	nsqdRpcClients     map[string]*NsqdRpcClient
	checkTopicFailChan chan TopicNameInfo
	stopChan           chan struct{}
	joinStateMutex     sync.Mutex
	joinISRState       map[string]*JoinISRState
	failedRpcMutex     sync.Mutex
	failedRpcList      []RpcFailedInfo
	nsqlookupRpcServer *NsqLookupCoordRpcServer
	wg                 sync.WaitGroup
	nsqdMonitorChan    chan struct{}
	isClusterUnstable  int32
	isUpgrading        int32
	dpm                *DataPlacement
	balanceWaiting     int32
	doChecking         int32
	interruptChecking  int32
	enableTopNBalance  int32
	cachedNodeStats    map[string]cachedNodeTopicStats
	cachedMutex        sync.Mutex
}
func NewNsqLookupCoordinator(cluster string, n *NsqLookupdNodeInfo, opts *Options) *NsqLookupCoordinator {
	coord := &NsqLookupCoordinator{
		clusterKey:         cluster,
		myNode:             *n,
		leadership:         nil,
		nsqdNodes:          make(map[string]NsqdNodeInfo),
		removingNodes:      make(map[string]string),
		nsqdRpcClients:     make(map[string]*NsqdRpcClient),
		checkTopicFailChan: make(chan TopicNameInfo, 3),
		stopChan:           make(chan struct{}),
		joinISRState:       make(map[string]*JoinISRState),
		failedRpcList:      make([]RpcFailedInfo, 0),
		nsqdMonitorChan:    make(chan struct{}),
		cachedNodeStats:    make(map[string]cachedNodeTopicStats),
	}
	if coord.leadership != nil {
		coord.leadership.InitClusterID(coord.clusterKey)
	}
	coord.nsqlookupRpcServer = NewNsqLookupCoordRpcServer(coord)
	coord.dpm = NewDataPlacement(coord)
	if opts != nil {
		coord.dpm.SetBalanceInterval(opts.BalanceStart, opts.BalanceEnd)
	}
	return coord
}
func (nlcoord *NsqLookupCoordinator) SetLeadershipMgr(l NSQLookupdLeadership) {
	nlcoord.leadership = l
	if nlcoord.leadership != nil {
		nlcoord.leadership.InitClusterID(nlcoord.clusterKey)
	}
}
func RetryWithTimeout(fn func() error) error {
	bo := backoff.NewExponentialBackOff()
	bo.MaxElapsedTime = time.Second * 15
	bo.MaxInterval = time.Second * 5
	return backoff.Retry(fn, bo)
}