Skip to content

Commit

Permalink
syncer: exit watch leader immediately (#8824)
Browse files Browse the repository at this point in the history
close #8823

Signed-off-by: Ryan Leung <[email protected]>
  • Loading branch information
rleungx authored Nov 20, 2024
1 parent 76e3b43 commit 41ec8dc
Showing 1 changed file with 15 additions and 4 deletions.
19 changes: 15 additions & 4 deletions pkg/syncer/client.go
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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()))
Expand All @@ -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() {
Expand Down Expand Up @@ -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 {
Expand Down

0 comments on commit 41ec8dc

Please sign in to comment.