fix a backoff bug & enable pause resume test

.github/workflows/check_and_build.yaml

@@ -88,6 +88,11 @@ jobs:
           pwd && ls -l bin/ && ls -l tools/bin/
           export TICDC_NEWARCH=true && make integration_test CASE=charset_gbk 
 
+      - name: Test changefeed_pause_resume
+        if: ${{ success() }}
+        run: |
+          export TICDC_NEWARCH=true && make integration_test CASE=changefeed_pause_resume
+
       # only upload logs of the last case
       - name: Copy logs to hack permission
         if: ${{ always() }}
@@ -99,6 +104,7 @@ jobs:
           tail -n 10 $DIR/cdc.log
           sudo cp -r -L $DIR/{cdc.log,stdout.log,sync_diff} ./logs/$CASE/
           sudo chown -R runner ./logs
+          sudo tar -czvf ./logs.tar.gz ./logs
 
       # Update logs as artifact seems not stable, so we set `continue-on-error: true` here.
       - name: Upload logs
@@ -107,4 +113,4 @@ jobs:
         with:
           name: upstream-switch-logs
           path: |
-            ./logs
+            ./logs.tar.gz

coordinator/changefeed/backoff.go

@@ -84,10 +84,7 @@ func (m *Backoff) ShouldRun() bool {
 func (m *Backoff) shouldFailWhenRetry() bool {
 	// NextBackOff() will return -1 once the MaxElapsedTime has elapsed,
 	// set the changefeed to failed state.
-	if m.backoffInterval == m.errBackoff.Stop {
-		return true
-	}
-	return false
+	return m.backoffInterval == m.errBackoff.Stop
 }
 
 // resetErrRetry reset the error retry related fields
@@ -154,6 +151,11 @@ func (m *Backoff) HandleError(errs []*heartbeatpb.RunningError) (bool, *heartbea
 	log.Warn("changefeed meets an error, will be stopped",
 		zap.Any("error", errs))
 
+	if !m.isRestarting.Load() {
+		// errBackoff may be stopped, reset it before the first retry.
+		m.resetErrRetry()
+		m.isRestarting.Store(true)
+	}
 	// set the next retry time
 	m.backoffInterval = m.errBackoff.NextBackOff()
 	m.nextRetryTime = atomic.NewTime(time.Now().Add(m.backoffInterval))
@@ -169,7 +171,6 @@ func (m *Backoff) HandleError(errs []*heartbeatpb.RunningError) (bool, *heartbea
 		)
 		return true, lastError
 	}
-	m.isRestarting.Store(true)
 	// patch the last error to changefeed info
 	return false, lastError
 }

coordinator/changefeed/backoff_test.go

@@ -17,6 +17,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/benbjohnson/clock"
 	"github.com/pingcap/ticdc/heartbeatpb"
 	"github.com/pingcap/ticdc/pkg/common"
 	"github.com/pingcap/tiflow/cdc/model"
@@ -26,6 +27,13 @@ import (
 func TestShouldFailWhenRetry(t *testing.T) {
 	backoff := NewBackoff(common.NewChangeFeedIDWithName("test"), time.Minute*30, 1)
 	require.True(t, backoff.ShouldRun())
+
+	// stop the backoff
+	mc := clock.NewMock()
+	mc.Set(time.Now())
+	backoff.errBackoff.Clock = mc
+	mc.Add(backoff.errBackoff.MaxElapsedTime + 1)
+
 	changefeed, state, err := backoff.CheckStatus(&heartbeatpb.MaintainerStatus{
 		CheckpointTs: 1,
 		Err: []*heartbeatpb.RunningError{

coordinator/changefeed/changefeed_db.go

@@ -105,6 +105,7 @@ func (db *ChangefeedDB) StopByChangefeedID(cfID common.ChangeFeedID, remove bool
 			db.changefeeds[cfID] = cf
 			db.stopped[cfID] = cf
 		}
+		log.Info("stop changefeed", zap.String("changefeed", cfID.String()))
 
 		nodeID := cf.GetNodeID()
 		if cf.GetNodeID() == "" {

coordinator/controller.go

@@ -57,11 +57,10 @@ type Controller struct {
 
 	bootstrapper *bootstrap.Bootstrapper[heartbeatpb.CoordinatorBootstrapResponse]
 
-	stream                   dynstream.DynamicStream[int, string, *Event, *Controller, *StreamHandler]
-	taskScheduler            threadpool.ThreadPool
-	operatorControllerHandle *threadpool.TaskHandle
-	schedulerHandle          *threadpool.TaskHandle
-	backend                  changefeed.Backend
+	stream        dynstream.DynamicStream[int, string, *Event, *Controller, *StreamHandler]
+	taskScheduler threadpool.ThreadPool
+	taskHandlers  []*threadpool.TaskHandle
+	backend       changefeed.Backend
 
 	updatedChangefeedCh chan map[common.ChangeFeedID]*changefeed.Changefeed
 	stateChangedCh      chan *ChangefeedStateChangeEvent
@@ -125,9 +124,7 @@ func NewController(
 	for _, msg := range c.bootstrapper.HandleNewNodes(newNodes) {
 		_ = c.messageCenter.SendCommand(msg)
 	}
-	submitScheduledEvent(c.taskScheduler, c.stream, &Event{
-		eventType: EventPeriod,
-	}, time.Now().Add(time.Millisecond*500))
+	c.submitPeriodTask()
 	return c
 }
 
@@ -160,9 +157,6 @@ func (c *Controller) onPeriodTask() {
 	// resend bootstrap message
 	c.sendMessages(c.bootstrapper.ResendBootstrapMessage())
 	c.collectMetrics()
-	submitScheduledEvent(c.taskScheduler, c.stream, &Event{
-		eventType: EventPeriod,
-	}, time.Now().Add(time.Millisecond*500))
 }
 
 func (c *Controller) onMessage(msg *messaging.TargetMessage) {
@@ -361,17 +355,15 @@ func (c *Controller) FinishBootstrap(workingMap map[common.ChangeFeedID]remoteMa
 	}
 
 	// start operator and scheduler
-	c.operatorControllerHandle = c.taskScheduler.Submit(c.cfScheduller, time.Now())
-	c.schedulerHandle = c.taskScheduler.Submit(c.operatorController, time.Now())
+	operatorControllerHandle := c.taskScheduler.Submit(c.cfScheduller, time.Now())
+	schedulerHandle := c.taskScheduler.Submit(c.operatorController, time.Now())
+	c.taskHandlers = append(c.taskHandlers, operatorControllerHandle, schedulerHandle)
 	c.bootstrapped.Store(true)
 }
 
 func (c *Controller) Stop() {
-	if c.operatorControllerHandle != nil {
-		c.operatorControllerHandle.Cancel()
-	}
-	if c.schedulerHandle != nil {
-		c.schedulerHandle.Cancel()
+	for _, h := range c.taskHandlers {
+		h.Cancel()
 	}
 }
 
@@ -505,17 +497,13 @@ func (c *Controller) RemoveNode(id node.ID) {
 	c.operatorController.OnNodeRemoved(id)
 }
 
-// submitScheduledEvent submits a task to controller pool to send a future event
-func submitScheduledEvent(
-	scheduler threadpool.ThreadPool,
-	stream dynstream.DynamicStream[int, string, *Event, *Controller, *StreamHandler],
-	event *Event,
-	scheduleTime time.Time) {
+func (c *Controller) submitPeriodTask() {
 	task := func() time.Time {
-		stream.In() <- event
-		return time.Time{}
+		c.stream.In() <- &Event{eventType: EventPeriod}
+		return time.Now().Add(time.Millisecond * 500)
 	}
-	scheduler.SubmitFunc(task, scheduleTime)
+	periodTaskhandler := c.taskScheduler.SubmitFunc(task, time.Now().Add(time.Millisecond*500))
+	c.taskHandlers = append(c.taskHandlers, periodTaskhandler)
 }
 
 func (c *Controller) newBootstrapMessage(id node.ID) *messaging.TargetMessage {

coordinator/coordinator.go

@@ -20,7 +20,6 @@ import (
 
 	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/coordinator/changefeed"
-	"github.com/pingcap/ticdc/heartbeatpb"
 	"github.com/pingcap/ticdc/pkg/common"
 	appcontext "github.com/pingcap/ticdc/pkg/common/context"
 	"github.com/pingcap/ticdc/pkg/config"
@@ -247,14 +246,6 @@ func (c *coordinator) sendMessages(msgs []*messaging.TargetMessage) {
 	}
 }
 
-func (c *coordinator) newBootstrapMessage(id node.ID) *messaging.TargetMessage {
-	log.Info("send coordinator bootstrap request", zap.Any("to", id))
-	return messaging.NewSingleTargetMessage(
-		id,
-		messaging.MaintainerManagerTopic,
-		&heartbeatpb.CoordinatorBootstrapRequest{Version: c.version})
-}
-
 func (c *coordinator) updateGCSafepoint(
 	ctx context.Context,
 ) error {

coordinator/operator/operator_controller.go

@@ -95,9 +95,9 @@ func (oc *Controller) AddOperator(op operator.Operator[common.ChangeFeedID, *hea
 	oc.lock.Lock()
 	defer oc.lock.Unlock()
 
-	if _, ok := oc.operators[op.ID()]; ok {
+	if pre, ok := oc.operators[op.ID()]; ok {
 		log.Info("add operator failed, operator already exists",
-			zap.String("operator", op.String()))
+			zap.Stringer("operator", op), zap.Stringer("previousOperator", pre.OP))
 		return false
 	}
 	cf := oc.changefeedDB.GetByID(op.ID())

coordinator/scheduler/scheduler.go

@@ -72,7 +72,7 @@ func (s *Scheduler) Execute() time.Time {
 		absent, nodeSize := s.changefeedDB.GetWaitingSchedulingChangefeeds(s.absent, availableSize)
 		// add the absent node to the node size map
 		// todo: use the bootstrap nodes
-		for id, _ := range s.nodeManager.GetAliveNodes() {
+		for id := range s.nodeManager.GetAliveNodes() {
 			if _, ok := nodeSize[id]; !ok {
 				nodeSize[id] = 0
 			}

tests/integration_tests/changefeed_pause_resume/run.sh

@@ -46,6 +46,7 @@ function run() {
 	done
 
 	for i in $(seq 1 10); do
+		echo "Run $i test" # && read
 		cdc cli changefeed pause --changefeed-id=$changefeed_id --pd=$pd_addr
 
 		for j in $(seq 1 $TABLE_COUNT); do
@@ -66,6 +67,11 @@ function run() {
 		cdc cli changefeed resume --changefeed-id=$changefeed_id --pd=$pd_addr
 
 		check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
+
+		# 1. wait checkpoint ts updated to etcd
+		# 2. wait dispatch closed
+		# NOTICE: remove this sleep after safemode is supported in dispatcher
+		sleep 10
 	done
 
 	cleanup_process $CDC_BINARY