Repository: tikv/client-go
Pull request #979 (2023-09-15 06:40:09)

*: try follower when stale read request timeout (#969)
cherry-pick #969


Patch:
diff --git a/internal/locate/region_request.go b/internal/locate/region_request.go
index d362a12..8ca7d62 100644
--- a/internal/locate/region_request.go
+++ b/internal/locate/region_request.go
@@ -632,17 +632,15 @@ func (state *accessFollower) next(bo *retry.Backoffer, selector *replicaSelector
 				zap.Bool("leader-epoch-stale", leaderEpochStale),
 				zap.Bool("leader-unreachable", leaderUnreachable),
 				zap.Bool("leader-invalid", leaderInvalid),
+				zap.Bool("stale-read", state.isStaleRead),
 				zap.Any("labels", state.option.labels))
 		}
-		// If leader tried and received deadline exceeded error, return nil to upper layer to retry with default timeout.
-		if leader.deadlineErrUsingConfTimeout {
-			return nil, nil
-		}
-		if leaderInvalid {
+		if leaderInvalid || leader.deadlineErrUsingConfTimeout {
 			// In stale-read, the request will fallback to leader after the local follower failure.
 			// If the leader is also unavailable, we can fallback to the follower and use replica-read flag again,
 			// The remote follower not tried yet, and the local follower can retry without stale-read flag.
-			if state.isStaleRead {
+			// If leader tried and received deadline exceeded error, try follower.
+			if state.isStaleRead || leader.deadlineErrUsingConfTimeout {
 				selector.state = &tryFollower{
 					leaderIdx: state.leaderIdx,
 					lastIdx:   state.leaderIdx,
@@ -1316,6 +1314,7 @@ func (s *RegionRequestSender) logSendReqError(bo *retry.Backoffer, msg string, r
 		zap.String("replica-status", strings.Join(replicaStatus, "; ")),
 		zap.Int("total-backoff-ms", bo.GetTotalSleep()),
 		zap.Int("total-backoff-times", bo.GetTotalBackoffTimes()),
+		zap.Uint64("max-exec-timeout-ms", req.Context.MaxExecutionDurationMs),
 		zap.String("total-region-errors", totalErrorStr.String()))
 }
 


Test patch:
diff --git a/internal/locate/region_request3_test.go b/internal/locate/region_request3_test.go
index 711591f..dbcd5b9 100644
--- a/internal/locate/region_request3_test.go
+++ b/internal/locate/region_request3_test.go
@@ -1285,14 +1285,9 @@ func (s *testRegionRequestToThreeStoresSuite) TestSendReqFirstTimeout() {
 			s.Nil(err)
 			s.True(IsFakeRegionError(regionErr))
 			s.Equal(1, len(s.regionRequestSender.Stats))
-			if staleRead {
-				rpcNum := s.regionRequestSender.Stats[tikvrpc.CmdGet].Count
-				s.True(rpcNum == 1 || rpcNum == 2) // 1 rpc or 2 rpc
-			} else {
-				s.Equal(int64(3), s.regionRequestSender.Stats[tikvrpc.CmdGet].Count) // 3 rpc
-				s.Equal(3, len(reqTargetAddrs))                                      // each rpc to a different store.
-			}
-			s.Equal(0, bo.GetTotalBackoffTimes()) // no backoff since fast retry.
+			s.Equal(int64(3), s.regionRequestSender.Stats[tikvrpc.CmdGet].Count) // 3 rpc
+			s.Equal(3, len(reqTargetAddrs))                                      // each rpc to a different store.
+			s.Equal(0, bo.GetTotalBackoffTimes())                                // no backoff since fast retry.
 			// warn: must reset MaxExecutionDurationMs before retry.
 			resetStats()
 			if staleRead {
@@ -1315,6 +1310,46 @@ func (s *testRegionRequestToThreeStoresSuite) TestSendReqFirstTimeout() {
 	}
 }
 
+func (s *testRegionRequestToThreeStoresSuite) TestStaleReadTryFollowerAfterTimeout() {
+	var (
+		leaderAddr  string
+		leaderLabel []*metapb.StoreLabel
+	)
+	bo := retry.NewBackoffer(context.Background(), 10000)
+	mockRPCClient := &fnClient{fn: func(ctx context.Context, addr string, req *tikvrpc.Request, timeout time.Duration) (*tikvrpc.Response, error) {
+		if addr == leaderAddr {
+			return nil, context.DeadlineExceeded
+		}
+		return &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{Value: []byte("value")}}, nil
+	}}
+	s.regionRequestSender = NewRegionRequestSender(s.cache, mockRPCClient)
+	s.regionRequestSender.RegionRequestRuntimeStats = NewRegionRequestRuntimeStats()
+	getLocFn := func() *KeyLocation {
+		loc, err := s.regionRequestSender.regionCache.LocateKey(bo, []byte("a"))
+		s.Nil(err)
+		region := s.regionRequestSender.regionCache.GetCachedRegionWithRLock(loc.Region)
+		leaderStore, _, _, _ := region.WorkStorePeer(region.getStore())
+		leaderAddr, err = s.regionRequestSender.regionCache.getStoreAddr(s.bo, region, leaderStore)
+		s.Nil(err)
+		leaderLabel = []*metapb.StoreLabel{{Key: "id", Value: strconv.FormatUint(leaderStore.StoreID(), 10)}}
+		return loc
+	}
+	req := tikvrpc.NewRequest(tikvrpc.CmdGet, &kvrpcpb.GetRequest{Key: []byte("a")}, kvrpcpb.Context{})
+	req.EnableStaleRead()
+	loc := getLocFn()
+	var ops []StoreSelectorOption
+	ops = append(ops, WithMatchLabels(leaderLabel))
+	resp, _, err := s.regionRequestSender.SendReqCtx(bo, req, loc.Region, time.Second, tikvrpc.TiKV, ops...)
+	s.Nil(err)
+	regionErr, err := resp.GetRegionError()
+	s.Nil(err)
+	s.Nil(regionErr)
+	s.Equal([]byte("value"), resp.Resp.(*kvrpcpb.GetResponse).Value)
+	s.Equal(1, len(s.regionRequestSender.Stats))
+	s.Equal(int64(2), s.regionRequestSender.Stats[tikvrpc.CmdGet].Count) // 2 rpc
+	s.Equal(0, bo.GetTotalBackoffTimes())                                // no backoff since fast retry.
+}
+
 func (s *testRegionRequestToThreeStoresSuite) TestRetryRequestSource() {
 	leaderStore, _ := s.loadAndGetLeaderStore()
 	regionLoc, err := s.cache.LocateRegionByID(s.bo, s.regionID)