Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

kv/client: fix region loss in single region handler #3281

Merged
merged 4 commits into from
Nov 5, 2021
Merged

kv/client: fix region loss in single region handler #3281

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
39101af
kv/client: fix region loss in single region handler
amyangfei Nov 4, 2021
079187c
fix make check
amyangfei Nov 4, 2021
dadd865
address comment
amyangfei Nov 5, 2021
649cc62
Merge branch 'master' into fix-region-loss
ti-chi-bot Nov 5, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 7 additions & 6 deletions cdc/kv/region_worker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -235,7 +235,7 @@ func (w *regionWorker) checkShouldExit() error {
return nil
}

func (w *regionWorker) handleSingleRegionError(ctx context.Context, err error, state *regionFeedState) error {
func (w *regionWorker) handleSingleRegionError(err error, state *regionFeedState) error {
if state.lastResolvedTs > state.sri.ts {
state.sri.ts = state.lastResolvedTs
}
Expand Down Expand Up @@ -273,7 +273,9 @@ func (w *regionWorker) handleSingleRegionError(ctx context.Context, err error, s
}

revokeToken := !state.initialized
err2 := w.session.onRegionFail(ctx, regionErrorInfo{
// since the context used in region worker will be cancelled after region
// worker exits, we must use the parent context to prevent regionErrorInfo loss.
err2 := w.session.onRegionFail(w.parentCtx, regionErrorInfo{
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we need to add some comments here? I see another place where it is used added a comment why we use parentCtx instead of ctx.

See: https://github.com/pingcap/ticdc/blob/37bac66f0673103892c71e585ab3d9d4658c1f74/cdc/kv/region_worker.go#L792
It looks like the changes here are also for this purpose, sorry I'm not very familiar with this piece of code.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes, the same reason

singleRegionInfo: state.sri,
err: err,
}, revokeToken)
Expand Down Expand Up @@ -388,26 +390,25 @@ func (w *regionWorker) processEvent(ctx context.Context, event *regionStatefulEv
case *cdcpb.Event_Entries_:
err = w.handleEventEntry(ctx, x, event.state)
if err != nil {
err = w.handleSingleRegionError(ctx, err, event.state)
err = w.handleSingleRegionError(err, event.state)
}
case *cdcpb.Event_Admin_:
log.Info("receive admin event", zap.Stringer("event", event.changeEvent))
case *cdcpb.Event_Error:
err = w.handleSingleRegionError(
ctx,
cerror.WrapError(cerror.ErrEventFeedEventError, &eventError{err: x.Error}),
event.state,
)
case *cdcpb.Event_ResolvedTs:
if err = w.handleResolvedTs(ctx, x.ResolvedTs, event.state); err != nil {
err = w.handleSingleRegionError(ctx, err, event.state)
err = w.handleSingleRegionError(err, event.state)
}
}
}

if event.resolvedTs != nil {
if err = w.handleResolvedTs(ctx, event.resolvedTs.Ts, event.state); err != nil {
err = w.handleSingleRegionError(ctx, err, event.state)
err = w.handleSingleRegionError(err, event.state)
}
}
event.state.lock.Unlock()
Expand Down