From b4ce5d00fbcfde842d71763c6bd07fc48e6c9233 Mon Sep 17 00:00:00 2001
From: tangenta <tangenta@126.com>
Date: Thu, 11 Jan 2024 11:32:01 +0800
Subject: [PATCH] ddl: check owner status before commit DDL changes (#50251)

close pingcap/tidb#47105, close pingcap/tidb#50055
---
 pkg/ddl/ddl_worker.go | 17 +++++++++++++++++
 1 file changed, 17 insertions(+)

diff --git a/pkg/ddl/ddl_worker.go b/pkg/ddl/ddl_worker.go
index d1d3c80a42..d3dea0240c 100644
--- a/pkg/ddl/ddl_worker.go
+++ b/pkg/ddl/ddl_worker.go
@@ -849,6 +849,9 @@ func (w *JobContext) setDDLLabelForDiagnosis(jobType model.ActionType) {
 }
 
 func (w *worker) HandleJobDone(d *ddlCtx, job *model.Job, t *meta.Meta) error {
+	if err := w.checkOwnerBeforeCommit(); err != nil {
+		return err
+	}
 	err := w.finishDDLJob(t, job)
 	if err != nil {
 		w.sess.Rollback()
@@ -942,6 +945,10 @@ func (w *worker) HandleDDLJobTable(d *ddlCtx, job *model.Job) (int64, error) {
 		return 0, err
 	}
 
+	if err = w.checkOwnerBeforeCommit(); err != nil {
+		return 0, err
+	}
+
 	if runJobErr != nil && !job.IsRollingback() && !job.IsRollbackDone() {
 		// If the running job meets an error
 		// and the job state is rolling back, it means that we have already handled this error.
@@ -989,6 +996,16 @@ func (w *worker) HandleDDLJobTable(d *ddlCtx, job *model.Job) (int64, error) {
 	return schemaVer, nil
 }
 
+func (w *worker) checkOwnerBeforeCommit() error {
+	if !w.ddlCtx.isOwner() && w.tp != localWorker {
+		// Since this TiDB instance is not a DDL owner anymore,
+		// it should not commit any transaction.
+		w.sess.Rollback()
+		return dbterror.ErrNotOwner
+	}
+	return nil
+}
+
 // HandleDDLJobV2 handles v2 ddl job.
 // Compare with v1:
 // 1. directly insert the job to history job table(incompatible with CDC).