func (mgr *placementManager) HasReplacementInstance() (bool, error) {
	_, placement, err := mgr.Placement()
	if err != nil {
		return false, err
	}
	currInstance, err := mgr.instanceFrom(placement)
	if err != nil {
		return false, err
	}
	currShardSetID := currInstance.ShardSetID()
	currShards := currInstance.Shards().All()
	for _, currShard := range currShards {
		if currShard.State() != shard.Leaving {
			return false, nil
		}
	}
	allInstances := placement.Instances()
	for _, instance := range allInstances {
		if instance.ShardSetID() != currShardSetID || instance.ID() == mgr.instanceID {
			continue
		}
		otherShards := instance.Shards().All()
		if len(otherShards) != len(currShards) {
			continue
		}
		match := true
		for i := 0; i < len(otherShards); i++ {
			if otherShards[i].State() != shard.Initializing ||
				otherShards[i].ID() != currShards[i].ID() ||
				otherShards[i].CutoverNanos() != currShards[i].CutoffNanos() {
				match = false
				break
			}
		}
		if match {
			return true, nil
		}
	}
	return false, nil
}
TODO(xichen): move the method to placement interface.
package aggregator
import (
	"errors"
	"sync"
	"github.com/m3db/m3cluster/placement"
	"github.com/m3db/m3cluster/shard"
	"github.com/m3db/m3x/clock"
	"github.com/uber-go/tally"
)
var (
	ErrInstanceNotFoundInPlacement = errors.New("instance not found in placement")
	errPlacementManagerNotOpenOrClosed = errors.New("placement manager not open or closed")
	errPlacementManagerOpenOrClosed    = errors.New("placement manager already open or closed")
)
type PlacementManager interface {
	Open() error
	Placement() (placement.ActiveStagedPlacement, placement.Placement, error)
	Instance() (placement.Instance, error)
	InstanceFrom(placement placement.Placement) (placement.Instance, error)
	HasReplacementInstance() (bool, error)
	Shards() (shard.Shards, error)
	Close() error
}
type placementManagerMetrics struct {
	activeStagedPlacementErrors tally.Counter
	activePlacementErrors       tally.Counter
	instanceNotFound            tally.Counter
}
func newPlacementManagerMetrics(scope tally.Scope) placementManagerMetrics {
	return placementManagerMetrics{
		activeStagedPlacementErrors: scope.Counter("active-staged-placement-errors"),
		activePlacementErrors:       scope.Counter("active-placement-errors"),
		instanceNotFound:            scope.Counter("instance-not-found"),
	}
}
type placementManagerState int
const (
	placementManagerNotOpen placementManagerState = iota
	placementManagerOpen
	placementManagerClosed
)
type placementManager struct {
	sync.RWMutex
	nowFn            clock.NowFn
	instanceID       string
	placementWatcher placement.StagedPlacementWatcher
	state   placementManagerState
	metrics placementManagerMetrics
}
func NewPlacementManager(opts PlacementManagerOptions) PlacementManager {
	instrumentOpts := opts.InstrumentOptions()
	return &placementManager{
		nowFn:            opts.ClockOptions().NowFn(),
		instanceID:       opts.InstanceID(),
		placementWatcher: opts.StagedPlacementWatcher(),
		metrics:          newPlacementManagerMetrics(instrumentOpts.MetricsScope()),
	}
}
func (mgr *placementManager) Open() error {
	mgr.Lock()
	defer mgr.Unlock()
	if mgr.state != placementManagerNotOpen {
		return errPlacementManagerOpenOrClosed
	}
	if err := mgr.placementWatcher.Watch(); err != nil {
		return err
	}
	mgr.state = placementManagerOpen
	return nil
}
func (mgr *placementManager) Placement() (placement.ActiveStagedPlacement, placement.Placement, error) {
	mgr.RLock()
	stagedPlacement, placement, err := mgr.placementWithLock()
	mgr.RUnlock()
	return stagedPlacement, placement, err
}
func (mgr *placementManager) Instance() (placement.Instance, error) {
	mgr.RLock()
	instance, err := mgr.instanceWithLock()
	mgr.RUnlock()
	return instance, err
}
func (mgr *placementManager) InstanceFrom(placement placement.Placement) (placement.Instance, error) {
	return mgr.instanceFrom(placement)
}