store/tikv: drop the unreachable store's regions from cache. (#2792)

store/tikv/region_cache.go

@@ -326,6 +326,14 @@ func (c *RegionCache) OnRequestFail(ctx *RPCContext) {
 	c.storeMu.Lock()
 	delete(c.storeMu.stores, storeID)
 	c.storeMu.Unlock()
+
+	c.mu.Lock()
+	for id, r := range c.mu.regions {
+		if r.peer.GetStoreId() == storeID {
+			c.dropRegionFromCache(id)
+		}
+	}
+	c.mu.Unlock()
 }
 
 // OnRegionStale removes the old region and inserts new regions into the cache.

store/tikv/region_cache_test.go

@@ -202,7 +202,7 @@ func (s *testRegionCacheSuite) TestSplit(c *C) {
 }
 
 func (s *testRegionCacheSuite) TestMerge(c *C) {
-	// ['' - 'm' - 'z']
+	// key range: ['' - 'm' - 'z']
 	region2 := s.cluster.AllocID()
 	newPeers := s.cluster.AllocIDs(2)
 	s.cluster.Split(s.region1, region2, []byte("m"), newPeers, newPeers[0])
@@ -254,6 +254,31 @@ func (s *testRegionCacheSuite) TestRequestFail(c *C) {
 	c.Assert(region.unreachableStores, HasLen, 0)
 }
 
+func (s *testRegionCacheSuite) TestRequestFail2(c *C) {
+	// key range: ['' - 'm' - 'z']
+	region2 := s.cluster.AllocID()
+	newPeers := s.cluster.AllocIDs(2)
+	s.cluster.Split(s.region1, region2, []byte("m"), newPeers, newPeers[0])
+
+	// Check the two regions.
+	loc1, err := s.cache.LocateKey(s.bo, []byte("a"))
+	c.Assert(err, IsNil)
+	c.Assert(loc1.Region.id, Equals, s.region1)
+	loc2, err := s.cache.LocateKey(s.bo, []byte("x"))
+	c.Assert(err, IsNil)
+	c.Assert(loc2.Region.id, Equals, region2)
+
+	// Request should fail on region1.
+	ctx, _ := s.cache.GetRPCContext(s.bo, loc1.Region)
+	c.Assert(s.cache.storeMu.stores, HasLen, 1)
+	s.checkCache(c, 2)
+	s.cache.OnRequestFail(ctx)
+	// Both region2 and store should be dropped from cache.
+	c.Assert(s.cache.storeMu.stores, HasLen, 0)
+	c.Assert(s.cache.getRegionFromCache([]byte("x")), IsNil)
+	s.checkCache(c, 1)
+}
+
 func (s *testRegionCacheSuite) TestUpdateStoreAddr(c *C) {
 	client := &RawKVClient{
 		clusterID:   0,