config, changefeed (ticdc): add changefeed error stuck duration config (#9872)

cdc/owner/feed_state_manager.go

@@ -63,10 +63,18 @@
 
 	// resolvedTs and initCheckpointTs is for checking whether resolved timestamp
 	// has been advanced or not.
+<<<<<<< HEAD
 	resolvedTs           model.Ts
 	checkpointTs         model.Ts
 	checkpointTsAdvanced time.Time
 
+=======
+	resolvedTs       model.Ts
+	initCheckpointTs model.Ts
+
+	checkpointTsAdvanced         time.Time
+	lastCheckpointTs             model.Ts
+>>>>>>> 797dc6dd3a (config, changefeed (ticdc): add changefeed error stuck duration config (#9872))
 	changefeedErrorStuckDuration time.Duration
 }
 
@@ -85,6 +93,7 @@
 	m.changefeedErrorStuckDuration = *cfg.ChangefeedErrorStuckDuration
 
 	m.resetErrRetry()
+<<<<<<< HEAD
 	m.isRetrying = false
 	return m
 }
@@ -105,6 +114,9 @@
 
 	m.lastErrorRetryTime = time.Now()
 	return false
+=======
+	return m
+>>>>>>> 797dc6dd3a (config, changefeed (ticdc): add changefeed error stuck duration config (#9872))
 }
 
 // resetErrRetry reset the error retry related fields
@@ -582,9 +594,19 @@
 		currTime := m.upstream.PDClock.CurrentTime()
 		ckptTime := oracle.GetTimeFromTS(m.state.Status.CheckpointTs)
 		m.lastWarningReportCheckpointTs = m.state.Status.CheckpointTs
+<<<<<<< HEAD
 
 		checkpointTsStuck := time.Since(m.checkpointTsAdvanced) > m.changefeedErrorStuckDuration
 		if checkpointTsStuck {
+=======
+		// Conditions:
+		// 1. checkpoint lag is large enough;
+		// 2. checkpoint hasn't been advanced for a long while;
+		// 3. the changefeed has been initialized.
+		if currTime.Sub(ckptTime) > m.changefeedErrorStuckDuration &&
+			time.Since(m.checkpointTsAdvanced) > m.changefeedErrorStuckDuration &&
+			m.resolvedTs > m.initCheckpointTs {
+>>>>>>> 797dc6dd3a (config, changefeed (ticdc): add changefeed error stuck duration config (#9872))
 			log.Info("changefeed retry on warning for a very long time and does not resume, "+
 				"it will be failed", zap.String("changefeed", m.state.ID.ID),
 				zap.Uint64("checkpointTs", m.state.Status.CheckpointTs),

pkg/orchestrator/reactor_state.go

@@ -262,6 +262,7 @@ func (s *ChangefeedReactorState) UpdateCDCKey(key *etcd.CDCKey, value []byte) er
 		return errors.Trace(err)
 	}
 	if key.Tp == etcd.CDCKeyTypeChangefeedInfo {
+		log.Info("update changefeed info", zap.Any("info", s.Info))
 		s.Info.VerifyAndComplete()
 	}
 	return nil