syncer: exit watch leader immediately (#8824) (#8827)

close #8823

Signed-off-by: ti-chi-bot <[email protected]>
Signed-off-by: Ryan Leung <[email protected]>

Co-authored-by: Ryan Leung <[email protected]>

server/region_syncer/client.go

@@ -38,6 +38,7 @@ import (
 const (
 	keepaliveTime    = 10 * time.Second
 	keepaliveTimeout = 3 * time.Second
+	retryInterval    = time.Second
 )
 
 // StopSyncWithLeader stop to sync the region with leader.
@@ -163,7 +164,15 @@ func (s *RegionSyncer) StartSyncWithLeader(addr string) {
 					}
 				}
 				log.Error("server failed to establish sync stream with leader", zap.String("server", s.server.Name()), zap.String("leader", s.server.GetLeader().GetName()), errs.ZapError(err))
-				time.Sleep(time.Second)
+				timer := time.NewTimer(retryInterval)
+				select {
+				case <-ctx.Done():
+					log.Info("stop synchronizing with leader due to context canceled")
+					timer.Stop()
+					return
+				case <-timer.C:
+					timer.Stop()
+				}
 				continue
 			}
 
@@ -175,7 +184,15 @@ func (s *RegionSyncer) StartSyncWithLeader(addr string) {
 					if err = stream.CloseSend(); err != nil {
 						log.Error("failed to terminate client stream", errs.ZapError(errs.ErrGRPCCloseSend, err))
 					}
-					time.Sleep(time.Second)
+					timer := time.NewTimer(retryInterval)
+					select {
+					case <-ctx.Done():
+						log.Info("stop synchronizing with leader due to context canceled")
+						timer.Stop()
+						return
+					case <-timer.C:
+						timer.Stop()
+					}
 					break
 				}
 				if s.history.GetNextIndex() != resp.GetStartIndex() {