*: force the new upgrading tidb to be owner

pkg/owner/manager.go

@@ -499,3 +499,13 @@ func init() {
 		logutil.BgLogger().Warn("set manager session TTL failed", zap.Error(err))
 	}
 }
+
+// DeleteLeader deletes the leader key.
+func DeleteLeader(ctx context.Context, cli *clientv3.Client, key string) error {
+	ownerKey, _, _, _, _, err := getOwnerInfo(ctx, ctx, cli, key)
+	if err != nil {
+		return errors.Trace(err)
+	}
+	_, err = cli.Delete(ctx, ownerKey)
+	return err
+}

pkg/session/bootstrap.go

@@ -31,6 +31,7 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tidb/pkg/bindinfo"
 	"github.com/pingcap/tidb/pkg/config"
+	"github.com/pingcap/tidb/pkg/ddl"
 	"github.com/pingcap/tidb/pkg/disttask/framework/proto"
 	"github.com/pingcap/tidb/pkg/domain"
 	"github.com/pingcap/tidb/pkg/domain/infosync"
@@ -62,6 +63,8 @@ import (
 	"go.uber.org/zap"
 )
 
+var bootstrapOwnerKey = "/tidb/distributeLock/"
+
 const (
 	// CreateUserTable is the SQL statement creates User table in system db.
 	// WARNING: There are some limitations on altering the schema of mysql.user table.
@@ -1369,6 +1372,77 @@ var (
 	SupportUpgradeHTTPOpVer int64 = version174
 )
 
+func acquireLock(s sessiontypes.Session) bool {
+	dom := domain.GetDomain(s)
+	if dom == nil {
+		logutil.BgLogger().Warn("domain is nil")
+		return false
+	}
+	cli := dom.GetEtcdClient()
+	if cli == nil {
+		logutil.BgLogger().Warn("etcd client is nil")
+		return false
+	}
+	// The lock is used to make sure only one TiDB server is bootstrapping the system.
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	lease, err := cli.Grant(ctx, 30)
+	cancel()
+	if err != nil {
+		return false
+	}
+	etcdSession, err := concurrency.NewSession(cli, concurrency.WithLease(lease.ID))
+	if err != nil {
+		return false
+	}
+	mu := concurrency.NewMutex(etcdSession, bootstrapOwnerKey)
+	err = mu.Lock(context.Background())
+	return err == nil
+}
+
+func releaseLock(s sessiontypes.Session) {
+	dom := domain.GetDomain(s)
+	if dom == nil {
+		logutil.BgLogger().Warn("domain is nil")
+		return
+	}
+	cli := dom.GetEtcdClient()
+	if cli == nil {
+		logutil.BgLogger().Warn("etcd client is nil")
+		return
+	}
+	etcdSession, err := concurrency.NewSession(cli)
+	if err != nil {
+		return
+	}
+	mu := concurrency.NewMutex(etcdSession, bootstrapOwnerKey)
+	err = mu.Unlock(context.Background())
+	if err != nil {
+		logutil.BgLogger().Error("release lock failed", zap.Error(err))
+		return
+	}
+}
+
+func forceToLeader(ctx context.Context, s sessiontypes.Session) error {
+	dom := domain.GetDomain(s)
+	for !dom.DDL().OwnerManager().IsOwner() {
+		ownerID, err := dom.DDL().OwnerManager().GetOwnerID(ctx)
+		if err != nil && (errors.ErrorEqual(err, concurrency.ErrElectionNoLeader) || strings.Contains(err.Error(), "no owner")) {
+			time.Sleep(50 * time.Millisecond)
+			continue
+		} else if err != nil {
+			logutil.BgLogger().Error("unexpected error", zap.Error(err))
+			return err
+		}
+		err = owner.DeleteLeader(ctx, dom.EtcdClient(), ddl.DDLOwnerKey)
+		if err != nil {
+			logutil.BgLogger().Error("unexpected error", zap.Error(err), zap.String("ownerID", ownerID))
+			return err
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+	return nil
+}
+
 func checkDistTask(s sessiontypes.Session, ver int64) {
 	if ver > version195 {
 		// since version195 we enable dist task by default, no need to check
@@ -1418,35 +1492,31 @@ func checkDistTask(s sessiontypes.Session, ver int64) {
 // upgrade function  will do some upgrade works, when the system is bootstrapped by low version TiDB server
 // For example, add new system variables into mysql.global_variables table.
 func upgrade(s sessiontypes.Session) {
-	ver, err := getBootstrapVersion(s)
+	// Do upgrade works then update bootstrap version.
+	isNull, err := InitMDLVariableForUpgrade(s.GetStore())
+	if err != nil {
+		logutil.BgLogger().Fatal("[upgrade] init metadata lock failed", zap.Error(err))
+	}
+
+	var ver int64
+	acquireLock(s)
+	ver, err = getBootstrapVersion(s)
 	terror.MustNil(err)
 	if ver >= currentBootstrapVersion {
 		// It is already bootstrapped/upgraded by a higher version TiDB server.
+		releaseLock(s)
 		return
 	}
+	defer releaseLock(s)
 
-	checkDistTask(s, ver)
-	printClusterState(s, ver)
-
-	// Only upgrade from under version92 and this TiDB is not owner set.
-	// The owner in older tidb does not support concurrent DDL, we should add the internal DDL to job queue.
-	if ver < version92 {
-		useConcurrentDDL, err := checkOwnerVersion(context.Background(), domain.GetDomain(s))
-		if err != nil {
-			logutil.BgLogger().Fatal("[upgrade] upgrade failed", zap.Error(err))
-		}
-		if !useConcurrentDDL {
-			// Use another variable DDLForce2Queue but not EnableConcurrentDDL since in upgrade it may set global variable, the initial step will
-			// overwrite variable EnableConcurrentDDL.
-			variable.DDLForce2Queue.Store(true)
-		}
-	}
-	// Do upgrade works then update bootstrap version.
-	isNull, err := InitMDLVariableForUpgrade(s.GetStore())
+	err = forceToLeader(context.Background(), s)
 	if err != nil {
-		logutil.BgLogger().Fatal("[upgrade] init metadata lock failed", zap.Error(err))
+		logutil.BgLogger().Fatal("[upgrade] force to owner failed", zap.Error(err))
 	}
 
+	checkDistTask(s, ver)
+	printClusterState(s, ver)
+
 	// when upgrade from v6.4.0 or earlier, enables metadata lock automatically,
 	// but during upgrade we disable it.
 	if isNull {