ddl: handle cluster state retrieval failure (#53816)

close #53808

pkg/ddl/job_table.go

@@ -352,9 +352,29 @@ func (d *ddl) startLocalWorkerLoop() {
 }
 
 func (s *jobScheduler) startDispatchLoop() {
+	const retryInterval = 3 * time.Second
+	for {
+		err := s.startDispatch()
+		if err == context.Canceled {
+			logutil.DDLLogger().Info("startDispatchLoop quit due to context canceled")
+			return
+		}
+		logutil.DDLLogger().Warn("startDispatchLoop failed, retrying",
+			zap.Error(err))
+
+		select {
+		case <-s.schCtx.Done():
+			logutil.DDLLogger().Info("startDispatchLoop quit due to context done")
+			return
+		case <-time.After(retryInterval):
+		}
+	}
+}
+
+func (s *jobScheduler) startDispatch() error {
 	sessCtx, err := s.sessPool.Get()
 	if err != nil {
-		logutil.DDLLogger().Fatal("dispatch loop get session failed, it should not happen, please try restart TiDB", zap.Error(err))
+		return errors.Trace(err)
 	}
 	defer s.sessPool.Put(sessCtx)
 	se := sess.NewSession(sessCtx)
@@ -363,15 +383,15 @@ func (s *jobScheduler) startDispatchLoop() {
 		notifyDDLJobByEtcdCh = s.etcdCli.Watch(s.schCtx, addingDDLJobNotifyKey)
 	}
 	if err := s.checkAndUpdateClusterState(true); err != nil {
-		logutil.DDLLogger().Fatal("dispatch loop get cluster state failed, it should not happen, please try restart TiDB", zap.Error(err))
+		return errors.Trace(err)
 	}
 	ticker := time.NewTicker(dispatchLoopWaitingDuration)
 	defer ticker.Stop()
 	// TODO move waitSchemaSyncedController out of ddlCtx.
 	s.clearOnceMap()
 	for {
-		if s.schCtx.Err() != nil {
-			return
+		if err := s.schCtx.Err(); err != nil {
+			return err
 		}
 		failpoint.Inject("ownerResignAfterDispatchLoopCheck", func() {
 			if ingest.ResignOwnerForTest.Load() {
@@ -393,7 +413,7 @@ func (s *jobScheduler) startDispatchLoop() {
 				continue
 			}
 		case <-s.schCtx.Done():
-			return
+			return s.schCtx.Err()
 		}
 		if err := s.checkAndUpdateClusterState(false); err != nil {
 			continue