maintainer: Fix Data Race (#3494)

close #3481

Author: hongyunyan

coordinator/operator/operator_controller.go

@@ -163,7 +163,7 @@ func (oc *Controller) pushStopChangefeedOperator(keyspaceID uint32, cfID common.
 			zap.String("operator", old.OP.String()))
 		old.OP.OnTaskRemoved()
 		old.OP.PostFinish()
-		old.IsRemoved = true
+		old.IsRemoved.Store(true)
 		delete(oc.operators, old.OP.ID())
 	}
 	oc.pushOperator(op)
@@ -242,15 +242,15 @@ func (oc *Controller) pollQueueingOperator() (operator.Operator[common.ChangeFee
 		return nil, false
 	}
 	item := heap.Pop(&oc.runningQueue).(*operator.OperatorWithTime[common.ChangeFeedID, *heartbeatpb.MaintainerStatus])
-	if item.IsRemoved {
+	if item.IsRemoved.Load() {
 		return nil, true
 	}
 	op := item.OP
 	opID := item.OP.ID()
 	// always call the PostFinish method to ensure the operator is cleaned up by itself.
 	if op.IsFinished() {
 		op.PostFinish()
-		item.IsRemoved = true
+		item.IsRemoved.Store(true)
 		delete(oc.operators, opID)
 		metrics.CoordinatorFinishedOperatorCount.WithLabelValues(op.Type()).Inc()
 		metrics.CoordinatorOperatorDuration.WithLabelValues(op.Type()).Observe(time.Since(item.CreatedAt).Seconds())

maintainer/operator/operator_controller.go

@@ -245,13 +245,13 @@ func (oc *Controller) pollQueueingOperator() (
 	op := item.OP
 	opID := op.ID()
 	oc.mu.Unlock()
-	if item.IsRemoved {
+	if item.IsRemoved.Load() {
 		return nil, true
 	}
 	// always call the PostFinish method to ensure the operator is cleaned up by itself.
 	if op.IsFinished() {
 		op.PostFinish()
-		item.IsRemoved = true
+		item.IsRemoved.Store(true)
 
 		oc.mu.Lock()
 		delete(oc.operators, opID)
@@ -311,7 +311,7 @@ func (oc *Controller) removeReplicaSet(op *removeDispatcherOperator) {
 			zap.String("operator", old.OP.String()))
 		old.OP.OnTaskRemoved()
 		old.OP.PostFinish()
-		old.IsRemoved = true
+		old.IsRemoved.Store(true)
 
 		oc.mu.Lock()
 		delete(oc.operators, op.ID())

pkg/scheduler/operator/operator_queue.go

@@ -14,6 +14,7 @@
 package operator
 
 import (
+	"sync/atomic"
 	"time"
 
 	"github.com/pingcap/ticdc/pkg/scheduler/replica"
@@ -28,7 +29,7 @@ type OperatorWithTime[T replica.ReplicationID, S replica.ReplicationStatus] stru
 	// CreatedAt records when this operator was created
 	CreatedAt time.Time
 	// IsRemoved indicates whether this operator has been marked for removal
-	IsRemoved bool
+	IsRemoved atomic.Bool
 }
 
 func NewOperatorWithTime[T replica.ReplicationID, S replica.ReplicationStatus](op Operator[T, S], time time.Time) *OperatorWithTime[T, S] {