Skip to content

Commit

Permalink
Fix panic of adjacent-region-scheduler after pd transfer leader (#1250)
Browse files Browse the repository at this point in the history
* scheduler: fix panic of adjacent-region-scheduler after pd leader transferred
  • Loading branch information
nanne007 authored and nolouch committed Sep 20, 2018
1 parent fa5f08c commit de4e6d5
Showing 1 changed file with 7 additions and 2 deletions.
9 changes: 7 additions & 2 deletions server/schedulers/adjacent_region.go
Original file line number Diff line number Diff line change
Expand Up @@ -144,8 +144,8 @@ func (l *balanceAdjacentRegionScheduler) Schedule(cluster schedule.Cluster, opIn
regions := cluster.ScanRegions(l.lastKey, scanLimit)
// scan to the end
if len(regions) <= 1 {
l.adjacentRegionsCount = 0
schedulerStatus.WithLabelValues(l.GetName(), "adjacent_count").Set(float64(l.adjacentRegionsCount))
l.adjacentRegionsCount = 0
l.lastKey = []byte("")
return nil
}
Expand Down Expand Up @@ -199,7 +199,9 @@ func (l *balanceAdjacentRegionScheduler) process(cluster schedule.Cluster) []*sc
l.cacheRegions.head = head + 1
l.lastKey = r2.GetStartKey()
}()
if l.unsafeToBalance(cluster, r1) {
// after the cluster is prepared, there is a gap that some regions heartbeats are not received.
// Leader of those region is nil, and we should skip them.
if r1.GetLeader() == nil || r2.GetLeader() == nil || l.unsafeToBalance(cluster, r1) {
schedulerCounter.WithLabelValues(l.GetName(), "skip").Inc()
return nil
}
Expand All @@ -221,6 +223,9 @@ func (l *balanceAdjacentRegionScheduler) unsafeToBalance(cluster schedule.Cluste
return true
}
store := cluster.GetStore(region.GetLeader().GetStoreId())
if store == nil {
return true
}
s := l.selector.SelectSource(cluster, []*core.StoreInfo{store})
if s == nil {
return true
Expand Down

0 comments on commit de4e6d5

Please sign in to comment.