scheduler(ticdc): revert 3b8d55 and do not return error when resolvedTs less than checkpoint (#9953)

cdc/owner/changefeed.go

@@ -447,19 +447,27 @@
 		}
 	}
 
+<<<<<<< HEAD
 	checkpointTs := c.state.Status.CheckpointTs
 	if c.resolvedTs == 0 {
 		c.resolvedTs = checkpointTs
 	}
 
 	minTableBarrierTs := c.state.Status.MinTableBarrierTs
+=======
+	checkpointTs := c.latestStatus.CheckpointTs
+	if c.resolvedTs == 0 {
+		c.resolvedTs = checkpointTs
+	}
+	minTableBarrierTs := c.latestStatus.MinTableBarrierTs
+>>>>>>> 0c29040814 (scheduler(ticdc): revert 3b8d55 and do not return error when resolvedTs less than checkpoint (#9953))
 
 	failpoint.Inject("NewChangefeedNoRetryError", func() {
 		failpoint.Return(cerror.ErrStartTsBeforeGC.GenWithStackByArgs(checkpointTs-300, checkpointTs))
 	})
 	failpoint.Inject("NewChangefeedRetryError", func() {
 		failpoint.Return(errors.New("failpoint injected retriable error"))
 	})
 
 	if c.state.Info.Config.CheckGCSafePoint {
 		// Check TiDB GC safepoint does not exceed the checkpoint.
@@ -508,7 +516,7 @@
 	// For example, if the ddl job is `add index`, TiCDC will execute the ddl
 	// job again and cause the index to be added twice. We need to fix this
 	// problem in the future. See:https://github.com/pingcap/tiflow/issues/2543
 	if checkpointTs == minTableBarrierTs {
 		ddlStartTs = checkpointTs - 1
 	} else {
 		ddlStartTs = checkpointTs
@@ -548,7 +556,7 @@
 		zap.String("changefeed", c.id.ID),
 	)
 
 	c.ddlSink = c.newSink(c.id, c.state.Info, ctx.Throw, func(err error) {
 		select {
 		case <-ctx.Done():
 		case c.warningCh <- err: