disttask: fix subtask cannot resume after upgrade (#51072)

close #50894

Author: tangenta

pkg/disttask/framework/storage/task_table.go

@@ -746,3 +746,19 @@ func (mgr *TaskManager) GetAllSubtasks(ctx context.Context) ([]*proto.Subtask, e
 	}
 	return subtasks, nil
 }
+
+// AdjustTaskOverflowConcurrency change the task concurrency to a max value supported by current cluster.
+// This is a workaround for an upgrade bug: in v7.5.x, the task concurrency is hard-coded to 16, resulting in
+// a stuck issue if the new version TiDB has less than 16 CPU count.
+// We don't adjust the concurrency in subtask table because this field does not exist in v7.5.0.
+// For details, see https://github.com/pingcap/tidb/issues/50894.
+// For the following versions, there is a check when submiting a new task. This function should be a no-op.
+func (mgr *TaskManager) AdjustTaskOverflowConcurrency(ctx context.Context, se sessionctx.Context) error {
+	cpuCount, err := mgr.getCPUCountOfManagedNode(ctx, se)
+	if err != nil {
+		return err
+	}
+	sql := "update mysql.tidb_global_task set concurrency = %? where concurrency > %?;"
+	_, err = sqlexec.ExecSQL(ctx, se, sql, cpuCount, cpuCount)
+	return err
+}

pkg/session/BUILD.bazel

@@ -23,6 +23,7 @@ go_library(
         "//pkg/ddl/placement",
         "//pkg/ddl/schematracker",
         "//pkg/ddl/syncer",
+        "//pkg/disttask/framework/storage",
         "//pkg/domain",
         "//pkg/domain/infosync",
         "//pkg/errno",
@@ -106,6 +107,7 @@ go_library(
         "@com_github_pingcap_errors//:errors",
         "@com_github_pingcap_failpoint//:failpoint",
         "@com_github_pingcap_kvproto//pkg/kvrpcpb",
+        "@com_github_pingcap_log//:log",
         "@com_github_pingcap_tipb//go-binlog",
         "@com_github_stretchr_testify//require",
         "@com_github_tikv_client_go_v2//error",

pkg/session/sync_upgrade.go

@@ -19,9 +19,12 @@ import (
 	"time"
 
 	"github.com/pingcap/failpoint"
+	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/pkg/ddl"
 	"github.com/pingcap/tidb/pkg/ddl/syncer"
+	dist_store "github.com/pingcap/tidb/pkg/disttask/framework/storage"
 	"github.com/pingcap/tidb/pkg/domain"
+	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/owner"
 	sessiontypes "github.com/pingcap/tidb/pkg/session/types"
 	"github.com/pingcap/tidb/pkg/sessionctx"
@@ -77,11 +80,12 @@ func SyncUpgradeState(s sessionctx.Context, timeout time.Duration) error {
 
 // SyncNormalRunning syncs normal state to etcd.
 func SyncNormalRunning(s sessionctx.Context) error {
+	bgCtx := context.Background()
 	failpoint.Inject("mockResumeAllJobsFailed", func(val failpoint.Value) {
 		if val.(bool) {
 			dom := domain.GetDomain(s)
 			//nolint: errcheck
-			dom.DDL().StateSyncer().UpdateGlobalState(context.Background(), syncer.NewStateInfo(syncer.StateNormalRunning))
+			dom.DDL().StateSyncer().UpdateGlobalState(bgCtx, syncer.NewStateInfo(syncer.StateNormalRunning))
 			failpoint.Return(nil)
 		}
 	})
@@ -95,7 +99,15 @@ func SyncNormalRunning(s sessionctx.Context) error {
 		logger.Warn("resume the job failed", zap.Error(e))
 	}
 
-	ctx, cancelFunc := context.WithTimeout(context.Background(), 3*time.Second)
+	if mgr, _ := dist_store.GetTaskManager(); mgr != nil {
+		ctx := kv.WithInternalSourceType(bgCtx, kv.InternalDistTask)
+		err := mgr.AdjustTaskOverflowConcurrency(ctx, s)
+		if err != nil {
+			log.Warn("cannot adjust task overflow concurrency", zap.Error(err))
+		}
+	}
+
+	ctx, cancelFunc := context.WithTimeout(bgCtx, 3*time.Second)
 	defer cancelFunc()
 	dom := domain.GetDomain(s)
 	err = dom.DDL().StateSyncer().UpdateGlobalState(ctx, syncer.NewStateInfo(syncer.StateNormalRunning))