ddl, owner: clean processing ddl jobs when owner retired (#50101)

close #50073

Author: tangenta

pkg/ddl/ddl.go

@@ -792,6 +792,12 @@ func (d *ddl) Start(ctxPool *pools.ResourcePool) error {
 	defer d.sessPool.Put(ctx)
 
 	ingest.InitGlobalLightningEnv()
+	d.ownerManager.SetRetireOwnerHook(func() {
+		// Since this instance is not DDL owner anymore, we clean up the processing job info.
+		if ingest.LitBackCtxMgr != nil {
+			ingest.LitBackCtxMgr.MarkJobFinish()
+		}
+	})
 
 	return nil
 }

pkg/ddl/job_table.go

@@ -225,6 +225,7 @@ func (d *ddl) getReorgJob(sess *sess.Session) (*model.Job, error) {
 			return false, nil
 		}
 		if (job.Type == model.ActionAddIndex || job.Type == model.ActionAddPrimaryKey) &&
+			job.State == model.JobStateQueueing &&
 			job.ReorgMeta != nil &&
 			job.ReorgMeta.IsFastReorg &&
 			ingest.LitBackCtxMgr != nil {

pkg/owner/manager.go

@@ -65,6 +65,8 @@ type Manager interface {
 
 	// SetBeOwnerHook sets a hook. The hook is called before becoming an owner.
 	SetBeOwnerHook(hook func())
+	// SetRetireOwnerHook will be called after retiring the owner.
+	SetRetireOwnerHook(hook func())
 }
 
 const (
@@ -114,8 +116,10 @@ type ownerManager struct {
 	elec           unsafe.Pointer
 	sessionLease   *atomicutil.Int64
 	wg             sync.WaitGroup
-	beOwnerHook    func()
 	campaignCancel context.CancelFunc
+
+	beOwnerHook     func()
+	retireOwnerHook func()
 }
 
 // NewOwnerManager creates a new Manager.
@@ -160,6 +164,10 @@ func (m *ownerManager) SetBeOwnerHook(hook func()) {
 	m.beOwnerHook = hook
 }
 
+func (m *ownerManager) SetRetireOwnerHook(hook func()) {
+	m.retireOwnerHook = hook
+}
+
 // ManagerSessionTTL is the etcd session's TTL in seconds. It's exported for testing.
 var ManagerSessionTTL = 60
 
@@ -222,6 +230,9 @@ func (m *ownerManager) toBeOwner(elec *concurrency.Election) {
 
 // RetireOwner make the manager to be a not owner.
 func (m *ownerManager) RetireOwner() {
+	if m.retireOwnerHook != nil {
+		m.retireOwnerHook()
+	}
 	atomic.StorePointer(&m.elec, nil)
 }
 

pkg/owner/mock.go

@@ -42,6 +42,7 @@ type mockManager struct {
 	wg           sync.WaitGroup
 	cancel       context.CancelFunc
 	beOwnerHook  func()
+	retireHook   func()
 	campaignDone chan struct{}
 	resignDone   chan struct{}
 }
@@ -168,10 +169,16 @@ func (*mockManager) RequireOwner(context.Context) error {
 	return nil
 }
 
+// SetBeOwnerHook implements Manager.SetBeOwnerHook interface.
 func (m *mockManager) SetBeOwnerHook(hook func()) {
 	m.beOwnerHook = hook
 }
 
+// SetRetireOwnerHook implements Manager.SetRetireOwnerHook interface.
+func (m *mockManager) SetRetireOwnerHook(hook func()) {
+	m.retireHook = hook
+}
+
 // CampaignCancel implements Manager.CampaignCancel interface
 func (m *mockManager) CampaignCancel() {
 	m.campaignDone <- struct{}{}