Potential bad case for tryIdleReplica

[zyguan]

Here is the case, if the leader is always busy and the followers are busy for a few times, tryIdleReplica will keep trying busy leader until backoff timeout.

func (s *testRegionRequestToThreeStoresSuite) TestDebugLeaderReadBusyThreshold() {
	// load region into cache
	key := []byte("key")
	loc, err := s.cache.LocateKey(s.bo, key)
	s.Require().NoError(err)

	// load region and leader store
	region := s.cache.GetCachedRegionWithRLock(loc.Region)
	leader, _, _, _ := region.WorkStorePeer(region.getStore())

	// make leader read request
	req := tikvrpc.NewReplicaReadRequest(tikvrpc.CmdGet, &kvrpcpb.GetRequest{Key: key}, kv.ReplicaReadLeader, nil)
	req.ReadReplicaScope = oracle.GlobalTxnScope
	req.TxnScope = oracle.GlobalTxnScope
	req.BusyThresholdMs = 100

	// setup mock client: 1) leader always returns server busy; 2) followers return server busy twice.
	followerBusyCount := 2
	s.regionRequestSender.client = &fnClient{fn: func(ctx context.Context, addr string, req *tikvrpc.Request, timeout time.Duration) (response *tikvrpc.Response, err error) {
		if addr == leader.addr || followerBusyCount > 0 {
			logutil.BgLogger().Sugar().Infof(">>> rpc to %s is busy", addr)
			if addr != leader.addr {
				followerBusyCount--
			}
			return &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{RegionError: &errorpb.Error{ServerIsBusy: &errorpb.ServerIsBusy{Reason: "mock busy", EstimatedWaitMs: 60000}}}}, nil
		}
		logutil.BgLogger().Sugar().Infof(">>> rpc to %s", addr)
		return &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{Value: []byte(addr)}}, nil
	}}

	bo := retry.NewBackoffer(context.Background(), -1)
	logutil.BgLogger().Sugar().Info(s.regionRequestSender.SendReqCtx(bo, req, loc.Region, time.Second, tikvrpc.TiKV))
}