*: Fix a bug when have a lot of jobs in upgrading state

ddl/BUILD.bazel

@@ -228,6 +228,7 @@ go_test(
         "//ddl/internal/session",
         "//ddl/placement",
         "//ddl/schematracker",
+        "//ddl/syncer",
         "//ddl/testutil",
         "//ddl/util",
         "//disttask/framework/proto",

ddl/job_table.go

@@ -39,6 +39,7 @@ import (
 	"github.com/pingcap/tidb/sessionctx/variable"
 	"github.com/pingcap/tidb/table"
 	tidb_util "github.com/pingcap/tidb/util"
+	"github.com/pingcap/tidb/util/dbterror"
 	"github.com/pingcap/tidb/util/intest"
 	"github.com/pingcap/tidb/util/logutil"
 	clientv3 "go.etcd.io/etcd/client/v3"
@@ -186,7 +187,12 @@ func (d *ddl) processJobDuringUpgrade(sess *sess.Session, job *model.Job) (isRun
 		}
 
 		if err != nil {
-			logutil.BgLogger().Warn("[ddl-upgrading] pause the job failed", zap.Stringer("job", job), zap.Bool("has job err", len(errs) > 0), zap.Error(err))
+			isCannotPauseDDLJobErr := dbterror.ErrCannotPauseDDLJob.Equal(err)
+			logutil.BgLogger().Warn("[ddl-upgrading] pause the job failed", zap.Stringer("job", job),
+				zap.Bool("isRunnable", isCannotPauseDDLJobErr), zap.Error(err))
+			if isCannotPauseDDLJobErr {
+				return true, nil
+			}
 		} else {
 			logutil.BgLogger().Warn("[ddl-upgrading] pause the job successfully", zap.Stringer("job", job))
 		}

ddl/job_table_test.go

@@ -15,13 +15,17 @@
 package ddl_test
 
 import (
+	"context"
+	"fmt"
 	"sync"
 	"testing"
 	"time"
 
+	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/ddl"
 	"github.com/pingcap/tidb/ddl/internal/callback"
+	"github.com/pingcap/tidb/ddl/syncer"
 	"github.com/pingcap/tidb/parser/model"
 	"github.com/pingcap/tidb/testkit"
 	"github.com/pingcap/tidb/util"
@@ -174,3 +178,67 @@ func check(t *testing.T, record []int64, ids ...int64) {
 		}
 	}
 }
+
+func TestUpgradingRelatedJobState(t *testing.T) {
+	store, dom := testkit.CreateMockStoreAndDomain(t)
+
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("CREATE TABLE e2 (id INT NOT NULL);")
+
+	d := dom.DDL()
+
+	testCases := []struct {
+		sql      string
+		jobState model.JobState
+		err      error
+	}{
+		{"alter table e2 add index idx(id)", model.JobStateDone, nil},
+		{"alter table e2 add index idx1(id)", model.JobStateCancelling, errors.New("[ddl:8214]Cancelled DDL job")},
+		{"alter table e2 add index idx2(id)", model.JobStateRollingback, errors.New("[ddl:8214]Cancelled DDL job")},
+		{"alter table e2 add index idx3(id)", model.JobStateRollbackDone, errors.New("[ddl:8214]Cancelled DDL job")},
+	}
+
+	require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/ddl/mockUpgradingState", `return(true)`))
+	defer func() {
+		require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/mockUpgradingState"))
+	}()
+
+	num := 0
+	hook := &callback.TestDDLCallback{}
+	hook.OnGetJobBeforeExported = func(jobType string) {
+		for i := 0; i < 100; i++ {
+			time.Sleep(time.Millisecond * 100)
+			jobs, err := ddl.GetAllDDLJobs(testkit.NewTestKit(t, store).Session(), nil)
+			require.NoError(t, err)
+			if len(jobs) < 1 || jobs[0].Query != testCases[num].sql {
+				continue
+			}
+
+			if testCases[num].err != nil && jobs[0].SchemaState == model.StateWriteOnly {
+				tk.MustExec("use test")
+				tk.MustExec(fmt.Sprintf("admin cancel ddl jobs %d", jobs[0].ID))
+			}
+			if jobs[0].State == testCases[num].jobState {
+				dom.DDL().StateSyncer().UpdateGlobalState(context.Background(), &syncer.StateInfo{State: syncer.StateUpgrading})
+			}
+			break
+		}
+	}
+	hook.OnGetJobAfterExported = func(jobType string, getJob *model.Job) {
+		if getJob.Query == testCases[num].sql && getJob.State == testCases[num].jobState {
+			dom.DDL().StateSyncer().UpdateGlobalState(context.Background(), &syncer.StateInfo{State: syncer.StateNormalRunning})
+		}
+	}
+	d.SetHook(hook)
+
+	for i, tc := range testCases {
+		num = i
+		if tc.err == nil {
+			tk.MustExec(tc.sql)
+		} else {
+			_, err := tk.Exec(tc.sql)
+			require.Equal(t, tc.err.Error(), err.Error())
+		}
+	}
+}

ddl/mock.go

@@ -161,6 +161,12 @@ func (s *MockStateSyncer) Init(context.Context) error {
 
 // UpdateGlobalState implements StateSyncer.UpdateGlobalState interface.
 func (s *MockStateSyncer) UpdateGlobalState(_ context.Context, stateInfo *syncer.StateInfo) error {
+	failpoint.Inject("mockUpgradingState", func(val failpoint.Value) {
+		if val.(bool) {
+			s.clusterState.Store(stateInfo)
+			failpoint.Return(nil)
+		}
+	})
 	s.globalVerCh <- clientv3.WatchResponse{}
 	s.clusterState.Store(stateInfo)
 	return nil

session/bootstrap.go

@@ -1188,21 +1188,21 @@ func syncUpgradeState(s Session) {
 		if err == nil && len(jobErrs) == 0 {
 			break
 		}
-		isAllFinished := true
+		jobErrStrs := make([]string, 0, len(jobErrs))
 		for _, jobErr := range jobErrs {
 			if dbterror.ErrPausedDDLJob.Equal(jobErr) {
 				continue
 			}
-			isAllFinished = false
+			jobErrStrs = append(jobErrStrs, jobErr.Error())
 		}
-		if isAllFinished {
+		if err == nil && len(jobErrs) == 0 {
 			break
 		}
 
 		if i == retryTimes-1 {
-			logutil.BgLogger().Fatal("[upgrading] pause all jobs failed", zap.Error(err))
+			logutil.BgLogger().Fatal("[upgrading] pause all jobs failed", zap.Strings("errs", jobErrStrs), zap.Error(err))
 		}
-		logutil.BgLogger().Warn("[upgrading] pause all jobs failed", zap.Error(err))
+		logutil.BgLogger().Warn("[upgrading] pause all jobs failed", zap.Strings("errs", jobErrStrs), zap.Error(err))
 		time.Sleep(interval)
 	}
 	logutil.BgLogger().Info("[upgrading] update global state to upgrading", zap.String("state", syncer.StateUpgrading))