move the remove logic to operator_controller.go

Author: sdojjy

maintainer/operator/operator_add.go

@@ -94,6 +94,6 @@ func (m *AddDispatcherOperator) PostFinish() {
 }
 
 func (m *AddDispatcherOperator) String() string {
-	return fmt.Sprintf("add dispatcher operator: %s, dest:%s ",
+	return fmt.Sprintf("add dispatcher operator: %s, dest:%s",
 		m.replicaSet.ID, m.dest)
 }

maintainer/operator/operator_controller.go

@@ -126,7 +126,7 @@ func (oc *Controller) AddOperator(op Operator) bool {
 	}
 	span := oc.replicationDB.GetTaskByID(op.ID())
 	if span == nil {
-		log.Warn("span not found",
+		log.Warn("add operator failed, span not found",
 			zap.String("changefeed", oc.changefeedID),
 			zap.String("operator", op.String()))
 		return false

maintainer/operator/operator_move.go

@@ -154,6 +154,6 @@ func (m *MoveDispatcherOperator) String() string {
 	m.lck.Lock()
 	defer m.lck.Unlock()
 
-	return fmt.Sprintf("move dispatcher operator: %s, origin:%s, dest:%s ",
+	return fmt.Sprintf("move dispatcher operator: %s, origin:%s, dest:%s",
 		m.replicaSet.ID, m.origin, m.dest)
 }

maintainer/operator/operator_split.go

@@ -83,6 +83,6 @@ func (m *SplitDispatcherOperator) Start() {
 
 func (m *SplitDispatcherOperator) String() string {
 	// todo add split region span
-	return fmt.Sprintf("move dispatcher operator: %s, dest:%v ",
+	return fmt.Sprintf("move dispatcher operator: %s, splitSpans:%v",
 		m.replicaSet.ID, m.splitSpans)
 }

maintainer/replica/span_replication.go

@@ -92,9 +92,8 @@ func NewWorkingReplicaSet(
 	return r
 }
 
-func (r *SpanReplication) UpdateStatus(status any) {
-	if status != nil {
-		newStatus := status.(*heartbeatpb.TableSpanStatus)
+func (r *SpanReplication) UpdateStatus(newStatus *heartbeatpb.TableSpanStatus) {
+	if newStatus != nil {
 		if newStatus.CheckpointTs >= r.status.CheckpointTs {
 			r.status = newStatus
 		}
@@ -124,12 +123,8 @@ func (r *SpanReplication) NewAddInferiorMessage(server node.ID) *messaging.Targe
 			ChangefeedID: r.ChangefeedID.ID,
 			Config: &heartbeatpb.DispatcherConfig{
 				DispatcherID: r.ID.ToPB(),
-				Span: &heartbeatpb.TableSpan{
-					TableID:  r.Span.TableID,
-					StartKey: r.Span.StartKey,
-					EndKey:   r.Span.EndKey,
-				},
-				StartTs: r.status.CheckpointTs,
+				Span:         r.Span,
+				StartTs:      r.status.CheckpointTs,
 			},
 			ScheduleAction: heartbeatpb.ScheduleAction_Create,
 		})