Merge branch 'client-go-503' of github.com:sticnarf/tidb into client-…

…go-503

br/pkg/restore/import.go

@@ -401,7 +401,8 @@ func (importer *FileImporter) ImportKVFiles(
 		logutil.Key("startKey", startKey),
 		logutil.Key("endKey", endKey))
 
-	rs := utils.InitialRetryState(32, 100*time.Millisecond, 8*time.Second)
+	// This RetryState will retry 48 time, for 5 min - 6 min.
+	rs := utils.InitialRetryState(48, 100*time.Millisecond, 8*time.Second)
 	ctl := OverRegionsInRange(startKey, endKey, importer.metaClient, &rs)
 	err = ctl.Run(ctx, func(ctx context.Context, r *RegionInfo) RPCResult {
 		return importer.ImportKVFileForRegion(ctx, file, rule, startTS, restoreTS, r)

br/pkg/restore/split.go

@@ -451,7 +451,9 @@ func PaginateScanRegion(
 			var batch []*RegionInfo
 			batch, err = client.ScanRegions(ctx, scanStartKey, endKey, limit)
 			if err != nil {
-				return errors.Trace(err)
+				err = errors.Annotatef(berrors.ErrPDBatchScanRegion, "scan regions from start-key:%s, err: %s",
+					redact.Key(scanStartKey), err.Error())
+				return err
 			}
 			regions = append(regions, batch...)
 			if len(batch) < limit {

br/pkg/restore/split_client.go

@@ -33,6 +33,7 @@ import (
 	"go.uber.org/multierr"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/status"
 )
@@ -499,6 +500,11 @@ func (c *pdClient) GetOperator(ctx context.Context, regionID uint64) (*pdpb.GetO
 }
 
 func (c *pdClient) ScanRegions(ctx context.Context, key, endKey []byte, limit int) ([]*RegionInfo, error) {
+	failpoint.Inject("no-leader-error", func(_ failpoint.Value) {
+		logutil.CL(ctx).Debug("failpoint no-leader-error injected.")
+		failpoint.Return(nil, status.Error(codes.Unavailable, "not leader"))
+	})
+
 	regions, err := c.client.ScanRegions(ctx, key, endKey, limit)
 	if err != nil {
 		return nil, errors.Trace(err)

br/pkg/restore/split_test.go

@@ -38,8 +38,9 @@ type TestClient struct {
 	injectInScatter     func(*restore.RegionInfo) error
 	supportBatchScatter bool
 
-	scattered map[uint64]bool
-	InjectErr bool
+	scattered   map[uint64]bool
+	InjectErr   bool
+	InjectTimes int32
 }
 
 func NewTestClient(
@@ -216,8 +217,9 @@ func (c *TestClient) GetOperator(ctx context.Context, regionID uint64) (*pdpb.Ge
 }
 
 func (c *TestClient) ScanRegions(ctx context.Context, key, endKey []byte, limit int) ([]*restore.RegionInfo, error) {
-	if c.InjectErr {
-		return nil, errors.New("mock scan error")
+	if c.InjectErr && c.InjectTimes > 0 {
+		c.InjectTimes -= 1
+		return nil, status.Error(codes.Unavailable, "not leader")
 	}
 
 	infos := c.regionsInfo.ScanRange(key, endKey, limit)

br/pkg/restore/util_test.go

@@ -233,7 +233,10 @@ func TestPaginateScanRegion(t *testing.T) {
 	require.Regexp(t, ".*scan region return empty result.*", err.Error())
 
 	regionMap, regions = makeRegions(1)
-	batch, err = restore.PaginateScanRegion(ctx, NewTestClient(stores, regionMap, 0), []byte{}, []byte{}, 3)
+	tc := NewTestClient(stores, regionMap, 0)
+	tc.InjectErr = true
+	tc.InjectTimes = 10
+	batch, err = restore.PaginateScanRegion(ctx, tc, []byte{}, []byte{}, 3)
 	require.NoError(t, err)
 	require.Equal(t, regions, batch)
 
@@ -273,11 +276,12 @@ func TestPaginateScanRegion(t *testing.T) {
 	require.True(t, berrors.ErrRestoreInvalidRange.Equal(err))
 	require.Regexp(t, ".*startKey > endKey.*", err.Error())
 
-	tc := NewTestClient(stores, regionMap, 0)
+	tc = NewTestClient(stores, regionMap, 0)
 	tc.InjectErr = true
-	_, err = restore.PaginateScanRegion(ctx, tc, regions[1].Region.EndKey, regions[5].Region.EndKey, 3)
+	tc.InjectTimes = 65
+	_, err = restore.PaginateScanRegion(ctx, tc, []byte{}, []byte{}, 3)
 	require.Error(t, err)
-	require.Regexp(t, ".*mock scan error.*", err.Error())
+	require.True(t, berrors.ErrPDBatchScanRegion.Equal(err))
 
 	// make the regionMap losing some region, this will cause scan region check fails
 	delete(regionMap, uint64(3))

br/pkg/task/restore.go

@@ -265,7 +265,7 @@ func (cfg *RestoreConfig) adjustRestoreConfig() {
 
 func (cfg *RestoreConfig) adjustRestoreConfigForStreamRestore() {
 	if cfg.Config.Concurrency == 0 {
-		cfg.Config.Concurrency = 32
+		cfg.Config.Concurrency = 16
 	}
 }
 

br/tests/br_full/run.sh

@@ -80,6 +80,7 @@ for ct in limit lz4 zstd; do
   # restore full
   echo "restore with $ct backup start..."
   export GO_FAILPOINTS="github.com/pingcap/tidb/br/pkg/restore/restore-storage-error=1*return(\"connection refused\");github.com/pingcap/tidb/br/pkg/restore/restore-gRPC-error=1*return(true)"
+  export GO_FAILPOINTS=$GO_FAILPOINTS";github.com/pingcap/tidb/br/pkg/restore/no-leader-error=3*return(true)"
   run_br restore full -s "local://$TEST_DIR/$DB-$ct" --pd $PD_ADDR --ratelimit 1024
   export GO_FAILPOINTS=""