@@ -174,7 +174,7 @@ func NewController(
174174 // detect the capture changes
175175 c .nodeManager .RegisterNodeChangeHandler (
176176 nodeChangeHandlerID ,
177- func (allNodes map [node.ID ]* node.Info ) {
177+ func (_ map [node.ID ]* node.Info ) {
178178 c .nodeChanged .Lock ()
179179 defer c .nodeChanged .Unlock ()
180180 c .nodeChanged .changed = true
@@ -355,9 +355,14 @@ func (c *Controller) RequestResolvedTsFromLogCoordinator(ctx context.Context, ch
355355 changefeedID := c .changefeedDB .GetChangefeedIDByName (changefeedDisplayName )
356356 ids := c .nodeManager .GetAliveNodeIDs ()
357357 for _ , id := range ids {
358- c .messageCenter .SendEvent (messaging .NewSingleTargetMessage (id , messaging .LogCoordinatorTopic , & heartbeatpb.LogCoordinatorResolvedTsRequest {
358+ if err := c .messageCenter .SendEvent (messaging .NewSingleTargetMessage (id , messaging .LogCoordinatorTopic , & heartbeatpb.LogCoordinatorResolvedTsRequest {
359359 ChangefeedID : changefeedID .ToPB (),
360- }))
360+ })); err != nil {
361+ log .Warn ("failed to request resolved ts from log coordinator" ,
362+ zap .Stringer ("target" , id ),
363+ zap .String ("changefeed" , changefeedID .DisplayName .String ()),
364+ zap .Error (err ))
365+ }
361366 }
362367
363368 // wait for some time to get the resolved ts
@@ -686,7 +691,7 @@ func (c *Controller) CreateChangefeed(ctx context.Context, info *config.ChangeFe
686691 return errors .Trace (ctx .Err ())
687692 case <- ticker .C :
688693 log .Warn ("changefeed is in scheduling, wait a moment" , zap .String ("changefeed" , info .ChangefeedID .DisplayName .String ()))
689- count += 1
694+ count ++
690695 }
691696 }
692697
@@ -728,7 +733,7 @@ func (c *Controller) RemoveChangefeed(ctx context.Context, id common.ChangeFeedI
728733 case <- ctx .Done ():
729734 return 0 , errors .Trace (ctx .Err ())
730735 case <- ticker .C :
731- count += 1
736+ count ++
732737 log .Info ("wait for stop changefeed operator finished" , zap .Int ("count" , count ), zap .Any ("id" , id ))
733738 }
734739 }
@@ -766,7 +771,7 @@ func (c *Controller) PauseChangefeed(ctx context.Context, id common.ChangeFeedID
766771 case <- ctx .Done ():
767772 return errors .Trace (ctx .Err ())
768773 case <- ticker .C :
769- count += 1
774+ count ++
770775 log .Info ("wait for stop changefeed operator finished" , zap .Int ("count" , count ), zap .Any ("id" , id ))
771776 }
772777 }
0 commit comments