diff --git a/pkg/syncer/client.go b/pkg/syncer/client.go index dd6edbd8d73..3d7f36d9114 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,12 @@ 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(): + log.Info("stop synchronizing with leader due to context canceled") + 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 +171,12 @@ 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(): + log.Info("stop synchronizing with leader due to context canceled") + return + case <-time.After(retryInterval): + } break } if s.history.getNextIndex() != resp.GetStartIndex() { @@ -208,13 +219,13 @@ func (s *RegionSyncer) StartSyncWithLeader(addr string) { log.Debug("region is stale", zap.Stringer("origin", origin.GetMeta()), errs.ZapError(err)) continue } - ctx := &core.MetaProcessContext{ + cctx := &core.MetaProcessContext{ Context: ctx, TaskRunner: ratelimit.NewSyncRunner(), Tracer: core.NewNoopHeartbeatProcessTracer(), // no limit for followers. } - saveKV, _, _, _ := regionGuide(ctx, region, origin) + saveKV, _, _, _ := regionGuide(cctx, region, origin) overlaps := bc.PutRegion(region) if hasBuckets {