coordinator: fix test assert and refactor StopByChangefeedID (#1898)

close #1897

Author: River2000i

coordinator/changefeed/backoff.go

@@ -106,7 +106,7 @@ func (m *Backoff) shouldFailWhenRetry() bool {
 // resetErrRetry reset the error retry related fields
 func (m *Backoff) resetErrRetry() {
 	m.errBackoff.Reset()
-	m.nextRetryTime = atomic.NewTime(time.Time{})
+	m.nextRetryTime.Store(time.Time{})
 	m.failed.Store(false)
 	m.retrying.Store(false)
 }
@@ -203,7 +203,7 @@ func (m *Backoff) HandleError(errs []*heartbeatpb.RunningError) (bool, *heartbea
 	}
 	// set the next retry time
 	m.backoffInterval = m.errBackoff.NextBackOff()
-	m.nextRetryTime = atomic.NewTime(time.Now().Add(m.backoffInterval))
+	m.nextRetryTime.Store(time.Now().Add(m.backoffInterval))
 
 	// check if we exceed the maxElapsedTime
 	if m.shouldFailWhenRetry() {

coordinator/changefeed/backoff_test.go

@@ -78,6 +78,7 @@ func TestRetry(t *testing.T) {
 	})
 	require.False(t, changefeed)
 	require.Equal(t, config.StateFailed, state)
+	require.Nil(t, err)
 	require.Equal(t, uint64(2), backoff.checkpointTs)
 
 	backoff.resetErrRetry()

coordinator/changefeed/changefeed_db.go

@@ -107,30 +107,27 @@ func (db *ChangefeedDB) StopByChangefeedID(cfID common.ChangeFeedID, remove bool
 	defer db.lock.Unlock()
 
 	cf, ok := db.changefeeds[cfID]
-	if ok {
-		// remove the changefeed
-		delete(db.changefeeds, cfID)
-		delete(db.stopped, cf.ID)
-		db.RemoveReplicaWithoutLock(cf)
-
-		if remove {
-			log.Info("remove changefeed", zap.String("changefeed", cf.ID.String()))
-		} else {
-			log.Info("stop changefeed", zap.String("changefeed", cfID.String()))
-			// push back to stopped
-			db.changefeeds[cfID] = cf
-			db.stopped[cfID] = cf
-		}
-
-		nodeID := cf.GetNodeID()
-		if cf.GetNodeID() == "" {
-			log.Info("changefeed is not scheduled, delete directly")
-			return ""
-		}
+	if !ok {
+		return ""
+	}
+	nodeID := cf.GetNodeID()
+	if nodeID != "" {
 		cf.SetNodeID("")
-		return nodeID
 	}
-	return ""
+
+	// Remove from replication tracking
+	db.RemoveReplicaWithoutLock(cf)
+
+	if remove {
+		log.Info("remove changefeed", zap.String("changefeed", cf.ID.String()))
+		delete(db.changefeeds, cfID)
+		delete(db.stopped, cfID)
+	} else {
+		log.Info("stop changefeed", zap.String("changefeed", cfID.String()))
+		db.stopped[cfID] = cf
+	}
+
+	return nodeID
 }
 
 // GetSize returns the size of the all chagnefeeds
@@ -186,8 +183,8 @@ func (db *ChangefeedDB) MarkMaintainerReplicating(task *Changefeed) {
 }
 
 // GetWaitingSchedulingChangefeeds returns the absent maintainers and the working state of each node
-func (db *ChangefeedDB) GetWaitingSchedulingChangefeeds(absent []*Changefeed, maxSize int) ([]*Changefeed, map[node.ID]int) {
-	absent = db.GetAbsent()
+func (db *ChangefeedDB) GetWaitingSchedulingChangefeeds(maxSize int) ([]*Changefeed, map[node.ID]int) {
+	absent := db.GetAbsent()
 	if len(absent) > maxSize {
 		absent = absent[:maxSize]
 	}

coordinator/changefeed/changefeed_db_test.go

@@ -69,6 +69,7 @@ func TestStopByChangefeedID(t *testing.T) {
 	require.Contains(t, db.stopped, cf.ID)
 	require.Contains(t, db.changefeeds, cf.ID)
 	require.Equal(t, node.ID("node-1"), nodeID)
+	require.Lenf(t, db.GetReplicating(), 0, "")
 
 	require.Equal(t, "", db.StopByChangefeedID(common.NewChangeFeedIDWithName("a"), false).String())
 }
@@ -96,12 +97,14 @@ func TestRemoveChangefeed(t *testing.T) {
 	db.StopByChangefeedID(cf.ID, false)
 	require.NotContains(t, db.GetAbsent(), cf)
 	require.Contains(t, db.changefeeds, cf.ID)
+	require.Contains(t, db.stopped, cf.ID)
 
 	cf2 := &Changefeed{ID: common.NewChangeFeedIDWithName("test2")}
 	db.AddReplicatingMaintainer(cf2, "node1")
 	require.Equal(t, node.ID("node1"), db.StopByChangefeedID(cf2.ID, true))
 	require.NotContains(t, db.GetAbsent(), cf2)
 	require.NotContains(t, db.changefeeds, cf2.ID)
+	require.NotContains(t, db.stopped, cf2.ID)
 	require.Equal(t, "", cf2.nodeID.String())
 }
 
@@ -140,13 +143,13 @@ func TestGetWaitingSchedulingChangefeeds(t *testing.T) {
 	cf3.backoff = NewBackoff(cf3.ID, 0, 0)
 	db.AddAbsentChangefeed(cf3)
 
-	result, nMap := db.GetWaitingSchedulingChangefeeds(nil, 3)
+	result, nMap := db.GetWaitingSchedulingChangefeeds(3)
 	require.Equal(t, 1, nMap["node1"])
 	require.NotContains(t, result, cf1)
 	require.NotContains(t, result, cf2)
 	require.Contains(t, result, cf3)
 
-	result, nMap = db.GetWaitingSchedulingChangefeeds(nil, 1)
+	result, nMap = db.GetWaitingSchedulingChangefeeds(1)
 	require.Equal(t, 1, nMap["node1"])
 	require.Len(t, result, 1)
 }