br: fix restore full fail when restore cluster <= 6.2 using br 6.3 (#…

…38082)

close #38055

br/pkg/pdutil/pd.go

@@ -147,18 +147,26 @@ func pdRequest(
 	ctx context.Context,
 	addr string, prefix string,
 	cli *http.Client, method string, body io.Reader) ([]byte, error) {
+	_, respBody, err := pdRequestWithCode(ctx, addr, prefix, cli, method, body)
+	return respBody, err
+}
+
+func pdRequestWithCode(
+	ctx context.Context,
+	addr string, prefix string,
+	cli *http.Client, method string, body io.Reader) (int, []byte, error) {
 	u, err := url.Parse(addr)
 	if err != nil {
-		return nil, errors.Trace(err)
+		return 0, nil, errors.Trace(err)
 	}
 	reqURL := fmt.Sprintf("%s/%s", u, prefix)
 	req, err := http.NewRequestWithContext(ctx, method, reqURL, body)
 	if err != nil {
-		return nil, errors.Trace(err)
+		return 0, nil, errors.Trace(err)
 	}
 	resp, err := cli.Do(req)
 	if err != nil {
-		return nil, errors.Trace(err)
+		return 0, nil, errors.Trace(err)
 	}
 	count := 0
 	for {
@@ -170,7 +178,7 @@ func pdRequest(
 		time.Sleep(pdRequestRetryInterval())
 		resp, err = cli.Do(req)
 		if err != nil {
-			return nil, errors.Trace(err)
+			return 0, nil, errors.Trace(err)
 		}
 	}
 	defer func() {
@@ -179,14 +187,14 @@ func pdRequest(
 
 	if resp.StatusCode != http.StatusOK {
 		res, _ := io.ReadAll(resp.Body)
-		return nil, errors.Annotatef(berrors.ErrPDInvalidResponse, "[%d] %s %s", resp.StatusCode, res, reqURL)
+		return resp.StatusCode, nil, errors.Annotatef(berrors.ErrPDInvalidResponse, "[%d] %s %s", resp.StatusCode, res, reqURL)
 	}
 
 	r, err := io.ReadAll(resp.Body)
 	if err != nil {
-		return nil, errors.Trace(err)
+		return resp.StatusCode, nil, errors.Trace(err)
 	}
-	return r, nil
+	return resp.StatusCode, r, nil
 }
 
 func pdRequestRetryInterval() time.Duration {
@@ -841,8 +849,14 @@ func (p *PdController) ResetTS(ctx context.Context, ts uint64) error {
 	})
 	var err error
 	for _, addr := range p.addrs {
-		_, e := pdRequest(ctx, addr, resetTSPrefix, p.cli, http.MethodPost, bytes.NewBuffer(reqData))
+		code, _, e := pdRequestWithCode(ctx, addr, resetTSPrefix, p.cli, http.MethodPost, bytes.NewBuffer(reqData))
 		if e != nil {
+			// for pd version <= 6.2, if the given ts < current ts of pd, pd returns StatusForbidden.
+			// it's not an error for br
+			if code == http.StatusForbidden {
+				log.Info("reset-ts returns with status forbidden, ignore")
+				return nil
+			}
 			log.Warn("failed to reset ts", zap.Uint64("ts", ts), zap.String("addr", addr), zap.Error(e))
 			err = e
 			continue

br/pkg/pdutil/pd_serial_test.go

@@ -205,6 +205,25 @@ func TestPDRequestRetry(t *testing.T) {
 	require.Error(t, reqErr)
 }
 
+func TestPDResetTSCompatibility(t *testing.T) {
+	ctx := context.Background()
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusForbidden)
+	}))
+	defer ts.Close()
+	pd := PdController{addrs: []string{ts.URL}, cli: http.DefaultClient}
+	reqErr := pd.ResetTS(ctx, 123)
+	require.NoError(t, reqErr)
+
+	ts2 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer ts2.Close()
+	pd = PdController{addrs: []string{ts2.URL}, cli: http.DefaultClient}
+	reqErr = pd.ResetTS(ctx, 123)
+	require.NoError(t, reqErr)
+}
+
 func TestStoreInfo(t *testing.T) {
 	storeInfo := pdtypes.StoreInfo{
 		Status: &pdtypes.StoreStatus{

br/pkg/task/BUILD.bazel

@@ -40,6 +40,7 @@ go_library(
         "//br/pkg/utils",
         "//br/pkg/version",
         "//config",
+        "//ddl",
         "//kv",
         "//parser/model",
         "//parser/mysql",

executor/kvtest/BUILD.bazel

@@ -7,8 +7,8 @@ go_test(
         "kv_test.go",
         "main_test.go",
     ],
-    race = "on",
     flaky = True,
+    race = "on",
     deps = [
         "//config",
         "//meta/autoid",