adjust the maintainer manager handle message

Author: 3AceShowHand

maintainer/maintainer_manager.go

@@ -56,9 +56,6 @@ type Manager struct {
 	pdClock     pdutil.Clock
 	regionCache *tikv.RegionCache
 
-	// msgCh is used to cache messages from coordinator
-	msgCh chan *messaging.TargetMessage
-
 	taskScheduler threadpool.ThreadPool
 }
 
@@ -95,16 +92,11 @@ func NewMaintainerManager(selfNode *node.Info,
 // recvMessages is the message handler for maintainer manager
 func (m *Manager) recvMessages(ctx context.Context, msg *messaging.TargetMessage) error {
 	switch msg.Type {
-	// Coordinator related messages
-	case messaging.TypeAddMaintainerRequest,
-		messaging.TypeRemoveMaintainerRequest,
-		messaging.TypeCoordinatorBootstrapRequest:
-		select {
-		case <-ctx.Done():
-			return ctx.Err()
-		case m.msgCh <- msg:
-		}
-		return nil
+	case messaging.TypeCoordinatorBootstrapRequest:
+		log.Info("received coordinator bootstrap request", zap.String("from", msg.From.String()))
+		m.onCoordinatorBootstrapRequest(msg)
+	case messaging.TypeAddMaintainerRequest, messaging.TypeRemoveMaintainerRequest:
+		m.onDispatchMaintainerRequest(msg)
 	// receive bootstrap response message from the dispatcher manager
 	case messaging.TypeMaintainerBootstrapResponse:
 		req := msg.Message[0].(*heartbeatpb.MaintainerBootstrapResponse)
@@ -140,8 +132,6 @@ func (m *Manager) Run(ctx context.Context) error {
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
-		case msg := <-m.msgCh:
-			m.handleMessage(msg)
 		case <-ticker.C:
 			// 1.  try to send heartbeat to coordinator
 			m.sendHeartbeat()
@@ -274,24 +264,33 @@ func (m *Manager) onRemoveMaintainerRequest(msg *messaging.TargetMessage) *heart
 
 func (m *Manager) onDispatchMaintainerRequest(
 	msg *messaging.TargetMessage,
-) *heartbeatpb.MaintainerStatus {
+) {
+	if !m.isBootstrap() {
+		return
+	}
 	if m.coordinatorID != msg.From {
 		log.Warn("ignore invalid coordinator id",
 			zap.Any("request", msg),
 			zap.Any("coordinatorID", m.coordinatorID),
 			zap.Stringer("from", msg.From))
-		return nil
 	}
+	var status *heartbeatpb.MaintainerStatus
 	switch msg.Type {
 	case messaging.TypeAddMaintainerRequest:
 		req := msg.Message[0].(*heartbeatpb.AddMaintainerRequest)
-		return m.onAddMaintainerRequest(req)
+		status = m.onAddMaintainerRequest(req)
 	case messaging.TypeRemoveMaintainerRequest:
-		return m.onRemoveMaintainerRequest(msg)
+		status = m.onRemoveMaintainerRequest(msg)
 	default:
 		log.Warn("unknown message type", zap.Any("message", msg.Message))
 	}
-	return nil
+	if status == nil {
+		return
+	}
+	response := &heartbeatpb.MaintainerHeartbeat{
+		Statuses: []*heartbeatpb.MaintainerStatus{status},
+	}
+	m.sendMessages(response)
 }
 
 func (m *Manager) sendHeartbeat() {
@@ -314,27 +313,6 @@ func (m *Manager) sendHeartbeat() {
 	}
 }
 
-func (m *Manager) handleMessage(msg *messaging.TargetMessage) {
-	switch msg.Type {
-	case messaging.TypeCoordinatorBootstrapRequest:
-		log.Info("received coordinator bootstrap request", zap.String("from", msg.From.String()))
-		m.onCoordinatorBootstrapRequest(msg)
-	case messaging.TypeAddMaintainerRequest,
-		messaging.TypeRemoveMaintainerRequest:
-		if m.isBootstrap() {
-			status := m.onDispatchMaintainerRequest(msg)
-			if status == nil {
-				return
-			}
-			response := &heartbeatpb.MaintainerHeartbeat{
-				Statuses: []*heartbeatpb.MaintainerStatus{status},
-			}
-			m.sendMessages(response)
-		}
-	default:
-	}
-}
-
 func (m *Manager) dispatcherMaintainerMessage(
 	ctx context.Context, changefeed common.ChangeFeedID, msg *messaging.TargetMessage,
 ) error {