fix bugs

pingcap · Aug 16, 2024 · 201e54f · 201e54f
1 parent c7535dd
commit 201e54f
Show file tree

Hide file tree

Showing 4 changed files with 3 additions and 4 deletions.
diff --git a/coordinator/coordinator.go b/coordinator/coordinator.go
@@ -86,6 +86,7 @@ func NewCoordinator(capture *common.NodeInfo,
 		id,
 		c.newChangefeed, c.newBootstrapMessage,
 		scheduler.NewBasicScheduler(id),
+		scheduler.NewBalanceScheduler(time.Minute, 1000),
 	)
 
 	// receive messages

diff --git a/downstreamadapter/dispathermanagermanager/dispatcher_manager_manager.go b/downstreamadapter/dispathermanagermanager/dispatcher_manager_manager.go
@@ -99,7 +99,8 @@ func (m *DispatcherManagerManager) handleAddDispatcherManager(from messaging.Ser
 		eventDispatcherManager.SetMaintainerID(from)
 	}
 	response := &heartbeatpb.MaintainerBootstrapResponse{
-		Statuses: make([]*heartbeatpb.TableSpanStatus, 0, eventDispatcherManager.GetDispatcherMap().Len()),
+		ChangefeedID: maintainerBootstrapRequest.ChangefeedID,
+		Statuses:     make([]*heartbeatpb.TableSpanStatus, 0, eventDispatcherManager.GetDispatcherMap().Len()),
 	}
 	eventDispatcherManager.GetDispatcherMap().ForEach(func(tableSpan *common.TableSpan, tableEventDispatcher *dispatcher.TableEventDispatcher) {
 		response.Statuses = append(response.Statuses, &heartbeatpb.TableSpanStatus{

diff --git a/maintainer/maintainer_manager.go b/maintainer/maintainer_manager.go
@@ -91,8 +91,6 @@ func (m *Manager) RecvMessages(ctx context.Context, msg *messaging.TargetMessage
 		case <-ctx.Done():
 			return ctx.Err()
 		case m.msgCh <- msg:
-		default:
-			log.Warn("msg chan is full")
 		}
 		return nil
 	// receive bootstrap response message from dispatcher manager manager

diff --git a/pkg/messaging/router.go b/pkg/messaging/router.go
@@ -54,7 +54,6 @@ func (r *router) runDispatch(ctx context.Context, wg *sync.WaitGroup, out <-chan
 				err := handler(ctx, msg)
 				if err != nil {
 					log.Error("router: close, since handle message failed", zap.Error(err), zap.Any("msg", msg))
-					return
 				}
 			}
 		}
-Original file line number
+Diff line change
@@ Expand Up @@
     				err := handler(ctx, msg)
     				if err != nil {
     					log.Error("router: close, since handle message failed", zap.Error(err), zap.Any("msg", msg))
-    					return
     				}
     			}
     		}
@@ Expand Down @@