From 4f97ed883e0f1f0c93fc6021d7589cc69d5d2be7 Mon Sep 17 00:00:00 2001 From: Ryan Leung Date: Tue, 19 Nov 2024 15:39:57 +0800 Subject: [PATCH] exit watch leader immediately Signed-off-by: Ryan Leung --- pkg/syncer/client.go | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/pkg/syncer/client.go b/pkg/syncer/client.go index dd6edbd8d73..a094ed0988d 100644 --- a/pkg/syncer/client.go +++ b/pkg/syncer/client.go @@ -43,6 +43,7 @@ const ( keepaliveTime = 10 * time.Second keepaliveTimeout = 3 * time.Second msgSize = 8 * units.MiB + retryInterval = time.Second ) // StopSyncWithLeader stop to sync the region with leader. @@ -153,7 +154,11 @@ 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) + select { + case <-ctx.Done(): + return + case <-time.After(retryInterval): + } continue } log.Info("server starts to synchronize with leader", zap.String("server", s.server.Name()), zap.String("leader", s.server.GetLeader().GetName()), zap.Uint64("request-index", s.history.getNextIndex())) @@ -165,7 +170,11 @@ 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) + select { + case <-ctx.Done(): + return + case <-time.After(retryInterval): + } break } if s.history.getNextIndex() != resp.GetStartIndex() {