From 3e55a4f7fc7060c9f69021a8a32a1be93b6fb8d9 Mon Sep 17 00:00:00 2001 From: Ti Chi Robot Date: Fri, 20 Dec 2024 10:31:18 +0800 Subject: [PATCH] syncer: exit watch leader immediately (#8824) (#8827) close tikv/pd#8823 Signed-off-by: ti-chi-bot Signed-off-by: Ryan Leung Co-authored-by: Ryan Leung --- server/region_syncer/client.go | 21 +++++++++++++++++++-- 1 file changed, 19 insertions(+), 2 deletions(-) diff --git a/server/region_syncer/client.go b/server/region_syncer/client.go index 8c8cfba3b5a..9401011ec5b 100644 --- a/server/region_syncer/client.go +++ b/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() {