fix(*): merge operator inconsistent after maintainer move

downstreamadapter/dispatchermanager/dispatcher_manager.go

@@ -97,6 +97,8 @@ type DispatcherManager struct {
 	// Entries must be deleted on completion (create -> after creation; remove -> on cleanup), otherwise
 	// future maintainer requests for the same dispatcherID will be ignored.
 	currentOperatorMap sync.Map // map[common.DispatcherID]SchedulerDispatcherRequest (in dispatcher manager, not heartbeatpb)
+	// mergeOperatorMap keeps in-flight merge requests so bootstrap can reconstruct merge operators after maintainer failover.
+	mergeOperatorMap sync.Map // map[mergedDispatcherID.String()]*heartbeatpb.MergeDispatcherRequest
 	// schemaIDToDispatchers is shared in the DispatcherManager,
 	// it store all the infos about schemaID->Dispatchers
 	// Dispatchers may change the schemaID when meets some special events, such as rename ddl
@@ -191,6 +193,7 @@ func NewDispatcherManager(
 		ctx:                   ctx,
 		dispatcherMap:         newDispatcherMap[*dispatcher.EventDispatcher](),
 		currentOperatorMap:    sync.Map{},
+		mergeOperatorMap:      sync.Map{},
 		changefeedID:          changefeedID,
 		keyspaceID:            keyspaceID,
 		pdClock:               pdClock,

downstreamadapter/dispatchermanager/dispatcher_manager_merge.go

@@ -0,0 +1,105 @@
+// Copyright 2025 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package dispatchermanager
+
+import (
+	"github.com/pingcap/log"
+	"github.com/pingcap/ticdc/heartbeatpb"
+	"github.com/pingcap/ticdc/pkg/common"
+	"go.uber.org/zap"
+)
+
+// TrackMergeOperator records an in-flight merge request so bootstrap can restore it after maintainer failover.
+func (e *DispatcherManager) TrackMergeOperator(req *heartbeatpb.MergeDispatcherRequest) {
+	if req == nil || req.MergedDispatcherID == nil {
+		return
+	}
+	mergedID := common.NewDispatcherIDFromPB(req.MergedDispatcherID)
+	e.mergeOperatorMap.Store(mergedID.String(), cloneMergeDispatcherRequest(req))
+}
+
+// RemoveMergeOperator drops the persisted merge request once the merged dispatcher has converged.
+func (e *DispatcherManager) RemoveMergeOperator(mergedDispatcherID common.DispatcherID) {
+	e.mergeOperatorMap.Delete(mergedDispatcherID.String())
+}
+
+// MaybeCleanupMergeOperator removes a persisted merge request when the merged dispatcher is already complete or gone.
+func (e *DispatcherManager) MaybeCleanupMergeOperator(req *heartbeatpb.MergeDispatcherRequest) {
+	if req == nil || req.MergedDispatcherID == nil {
+		return
+	}
+	mergedID := common.NewDispatcherIDFromPB(req.MergedDispatcherID)
+	if common.IsRedoMode(req.Mode) {
+		if dispatcherItem, ok := e.redoDispatcherMap.Get(mergedID); ok {
+			if dispatcherItem.GetComponentStatus() == heartbeatpb.ComponentState_Working {
+				e.RemoveMergeOperator(mergedID)
+			}
+			return
+		}
+	} else {
+		if dispatcherItem, ok := e.dispatcherMap.Get(mergedID); ok {
+			if dispatcherItem.GetComponentStatus() == heartbeatpb.ComponentState_Working {
+				e.RemoveMergeOperator(mergedID)
+			}
+			return
+		}
+	}
+	log.Info("cleanup merge operator because merged dispatcher not found",
+		zap.Stringer("changefeedID", e.changefeedID),
+		zap.String("dispatcherID", mergedID.String()),
+		zap.Int64("mode", req.Mode))
+	e.RemoveMergeOperator(mergedID)
+}
+
+// GetMergeOperators returns cloned in-flight merge requests for maintainer bootstrap.
+func (e *DispatcherManager) GetMergeOperators() []*heartbeatpb.MergeDispatcherRequest {
+	operators := make([]*heartbeatpb.MergeDispatcherRequest, 0)
+	e.mergeOperatorMap.Range(func(_, value any) bool {
+		req, ok := value.(*heartbeatpb.MergeDispatcherRequest)
+		if !ok || req == nil {
+			return true
+		}
+		operators = append(operators, cloneMergeDispatcherRequest(req))
+		return true
+	})
+	return operators
+}
+
+func cloneMergeDispatcherRequest(req *heartbeatpb.MergeDispatcherRequest) *heartbeatpb.MergeDispatcherRequest {
+	if req == nil {
+		return nil
+	}
+	clone := &heartbeatpb.MergeDispatcherRequest{
+		Mode: req.Mode,
+	}
+	if req.ChangefeedID != nil {
+		id := *req.ChangefeedID
+		clone.ChangefeedID = &id
+	}
+	if req.MergedDispatcherID != nil {
+		mergedID := *req.MergedDispatcherID
+		clone.MergedDispatcherID = &mergedID
+	}
+	if len(req.DispatcherIDs) > 0 {
+		clone.DispatcherIDs = make([]*heartbeatpb.DispatcherID, 0, len(req.DispatcherIDs))
+		for _, dispatcherID := range req.DispatcherIDs {
+			if dispatcherID == nil {
+				continue
+			}
+			id := *dispatcherID
+			clone.DispatcherIDs = append(clone.DispatcherIDs, &id)
+		}
+	}
+	return clone
+}

downstreamadapter/dispatchermanager/heartbeat_collector.go

@@ -60,6 +60,8 @@ type HeartBeatCollector struct {
 	mergeDispatcherRequestDynamicStream       dynstream.DynamicStream[int, common.GID, MergeDispatcherRequest, *DispatcherManager, *MergeDispatcherRequestHandler]
 	mc                                        messaging.MessageCenter
 
+	dispatcherManagers sync.Map // map[common.GID]*DispatcherManager
+
 	wg       sync.WaitGroup
 	cancel   context.CancelFunc
 	isClosed atomic.Bool
@@ -128,6 +130,7 @@ func (c *HeartBeatCollector) RegisterDispatcherManager(m *DispatcherManager) err
 	if err != nil {
 		return errors.Trace(err)
 	}
+	c.dispatcherManagers.Store(m.changefeedID.Id, m)
 	return nil
 }
 
@@ -170,6 +173,7 @@ func (c *HeartBeatCollector) RemoveDispatcherManager(id common.ChangeFeedID) err
 	if err != nil {
 		return errors.Trace(err)
 	}
+	c.dispatcherManagers.Delete(id.Id)
 	return nil
 }
 
@@ -287,6 +291,9 @@ func (c *HeartBeatCollector) RecvMessages(_ context.Context, msg *messaging.Targ
 			NewRedoMetaMessage(redoMessage))
 	case messaging.TypeMergeDispatcherRequest:
 		mergeDispatcherRequest := msg.Message[0].(*heartbeatpb.MergeDispatcherRequest)
+		if manager, ok := c.dispatcherManagers.Load(common.NewChangefeedGIDFromPB(mergeDispatcherRequest.ChangefeedID)); ok {
+			manager.(*DispatcherManager).TrackMergeOperator(mergeDispatcherRequest)
+		}
 		c.mergeDispatcherRequestDynamicStream.Push(
 			common.NewChangefeedGIDFromPB(mergeDispatcherRequest.ChangefeedID),
 			NewMergeDispatcherRequest(mergeDispatcherRequest))

downstreamadapter/dispatchermanager/helper.go

@@ -785,11 +785,19 @@ func (h *MergeDispatcherRequestHandler) Handle(dispatcherManager *DispatcherMana
 	}
 
 	mergeDispatcherRequest := reqs[0]
+	dispatcherManager.TrackMergeOperator(mergeDispatcherRequest.MergeDispatcherRequest)
 	dispatcherIDs := make([]common.DispatcherID, 0, len(mergeDispatcherRequest.DispatcherIDs))
 	for _, id := range mergeDispatcherRequest.DispatcherIDs {
 		dispatcherIDs = append(dispatcherIDs, common.NewDispatcherIDFromPB(id))
 	}
-	dispatcherManager.MergeDispatcher(dispatcherIDs, common.NewDispatcherIDFromPB(mergeDispatcherRequest.MergedDispatcherID), mergeDispatcherRequest.Mode)
+	task := dispatcherManager.MergeDispatcher(
+		dispatcherIDs,
+		common.NewDispatcherIDFromPB(mergeDispatcherRequest.MergedDispatcherID),
+		mergeDispatcherRequest.Mode,
+	)
+	if task == nil {
+		dispatcherManager.MaybeCleanupMergeOperator(mergeDispatcherRequest.MergeDispatcherRequest)
+	}
 	return false
 }
 

downstreamadapter/dispatchermanager/task.go

@@ -232,6 +232,7 @@ func abortMerge[T dispatcher.Dispatcher](t *MergeCheckTask, dispatcherMap *Dispa
 	}
 
 	removeDispatcher(t.manager, t.mergedDispatcher.GetId(), dispatcherMap, sinkType)
+	t.manager.RemoveMergeOperator(t.mergedDispatcher.GetId())
 }
 
 func doMerge[T dispatcher.Dispatcher](t *MergeCheckTask, dispatcherMap *DispatcherMap[T]) {
@@ -330,6 +331,7 @@ func doMerge[T dispatcher.Dispatcher](t *MergeCheckTask, dispatcherMap *Dispatch
 		}
 		dispatcher.Remove()
 	}
+	t.manager.RemoveMergeOperator(t.mergedDispatcher.GetId())
 }
 
 type mergedStartTsCandidate struct {

downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator.go

@@ -526,4 +526,6 @@ func retrieveOperatorsForBootstrapResponse(
 		})
 		return true
 	})
+
+	response.MergeOperators = append(response.MergeOperators, manager.GetMergeOperators()...)
 }