scheduler: add leader verify in balance-region scheduler

pkg/mock/mockcluster/mockcluster.go

@@ -574,6 +574,7 @@ func (mc *Cluster) RemoveScheduler(name string) error {
 }
 
 // MockRegionInfo returns a mock region
+// If leaderStoreID is zero, the regions would have no leader
 func (mc *Cluster) MockRegionInfo(regionID uint64, leaderStoreID uint64,
 	followerStoreIDs, learnerStoreIDs []uint64, epoch *metapb.RegionEpoch) *core.RegionInfo {
 
@@ -583,8 +584,11 @@ func (mc *Cluster) MockRegionInfo(regionID uint64, leaderStoreID uint64,
 		EndKey:      []byte(fmt.Sprintf("%20d", regionID+1)),
 		RegionEpoch: epoch,
 	}
-	leader, _ := mc.AllocPeer(leaderStoreID)
-	region.Peers = []*metapb.Peer{leader}
+	var leader *metapb.Peer
+	if leaderStoreID != 0 {
+		leader, _ = mc.AllocPeer(leaderStoreID)
+		region.Peers = append(region.Peers, leader)
+	}
 	for _, storeID := range followerStoreIDs {
 		peer, _ := mc.AllocPeer(storeID)
 		region.Peers = append(region.Peers, peer)

server/schedule/filter/filters.go

@@ -16,6 +16,7 @@ package filter
 import (
 	"fmt"
 
+	"github.com/golang/protobuf/proto"
 	"github.com/pingcap/kvproto/pkg/metapb"
 	"github.com/pingcap/log"
 	"github.com/tikv/pd/pkg/slice"
@@ -683,11 +684,7 @@ func (f *isolationFilter) Target(opt *config.PersistOptions, store *core.StoreIn
 // FitRegion in filter
 func createRegionForRuleFit(startKey, endKey []byte,
 	peers []*metapb.Peer, leader *metapb.Peer, opts ...core.RegionCreateOption) *core.RegionInfo {
-	copyLeader := &metapb.Peer{
-		Id:      leader.Id,
-		StoreId: leader.StoreId,
-		Role:    leader.Role,
-	}
+	copyLeader := proto.Clone(leader).(*metapb.Peer)
 	copyPeers := make([]*metapb.Peer, 0, len(peers))
 	for _, p := range peers {
 		peer := &metapb.Peer{

server/schedulers/balance_region.go

@@ -173,6 +173,12 @@ func (s *balanceRegionScheduler) Schedule(cluster opt.Cluster) []*operator.Opera
 				schedulerCounter.WithLabelValues(s.GetName(), "region-hot").Inc()
 				continue
 			}
+			// Check region whether have leader
+			if region.GetLeader() == nil {
+				log.Warn("region have no leader", zap.String("scheduler", s.GetName()), zap.Uint64("region-id", region.GetID()))
+				schedulerCounter.WithLabelValues(s.GetName(), "no-leader").Inc()
+				continue
+			}
 
 			oldPeer := region.GetStorePeer(sourceID)
 			if op := s.transferPeer(cluster, region, oldPeer); op != nil {

server/schedulers/balance_test.go

@@ -900,6 +900,23 @@ func (s *testBalanceRegionSchedulerSuite) checkReplacePendingRegion(c *C, tc *mo
 	testutil.CheckTransferPeer(c, sb.Schedule(tc)[0], operator.OpKind(0), 1, 4)
 }
 
+func (s *testBalanceRegionSchedulerSuite) TestShouldNotBalance(c *C) {
+	opt := config.NewTestOptions()
+	tc := mockcluster.NewCluster(opt)
+	tc.DisableFeature(versioninfo.JointConsensus)
+	oc := schedule.NewOperatorController(s.ctx, nil, nil)
+	sb, err := schedule.CreateScheduler(BalanceRegionType, oc, core.NewStorage(kv.NewMemoryKV()), schedule.ConfigSliceDecoder(BalanceRegionType, []string{"", ""}))
+	c.Assert(err, IsNil)
+	region := tc.MockRegionInfo(1, 0, []uint64{2, 3, 4}, nil, nil)
+	tc.PutRegion(region)
+	operators := sb.Schedule(tc)
+	if operators != nil {
+		c.Assert(len(operators), Equals, 0)
+	} else {
+		c.Assert(operators, IsNil)
+	}
+}
+
 var _ = Suite(&testRandomMergeSchedulerSuite{})
 
 type testRandomMergeSchedulerSuite struct{}