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

sink(cdc): handle sink errors more fast and light #8949

Merged
merged 23 commits into from
May 16, 2023
Merged

sink(cdc): handle sink errors more fast and light #8949

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
23 commits
Select commit Hold shift + click to select a range
f0424b2
init
hicqu May 9, 2023
c7dd8cd
handle error in api client
hicqu May 10, 2023
212b4fc
some updates
hicqu May 10, 2023
d6ddf8d
fix tests
hicqu May 11, 2023
12c21d1
make check
hicqu May 11, 2023
88598a4
some changes
hicqu May 12, 2023
e8f215c
fix tests
hicqu May 12, 2023
2d8e70d
fix tests
hicqu May 12, 2023
1343073
address comments
hicqu May 12, 2023
80aa1e7
fix tests
hicqu May 12, 2023
2bce733
fix tests
hicqu May 12, 2023
fc0e86d
make fmt
hicqu May 15, 2023
a2bd341
Merge branch 'master' into sink-error-split-1
ti-chi-bot[bot] May 15, 2023
c098ba5
Merge branch 'master' into sink-error-split-1
ti-chi-bot[bot] May 15, 2023
901cdb2
fix some integration tests
hicqu May 15, 2023
674b0d8
Merge branch 'sink-error-split-1' of github.com:hicqu/ticdc into sink…
hicqu May 15, 2023
8f0bbd9
disable a case temp
hicqu May 15, 2023
4ace4f9
fix some tests
hicqu May 16, 2023
bd16f01
some fixes
hicqu May 16, 2023
cd6d2a1
fix syncpoint
hicqu May 16, 2023
032c0da
Merge branch 'master' into sink-error-split-1
ti-chi-bot[bot] May 16, 2023
19a70d6
fix a bug that table sink doesn't know backend is dead
hicqu May 16, 2023
5ac5fe0
Merge branch 'sink-error-split-1' of github.com:hicqu/ticdc into sink…
hicqu May 16, 2023
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
2 changes: 1 addition & 1 deletion cdc/model/codec/codec.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -216,8 +216,8 @@ func redoLogFromV1(rv1 *codecv1.RedoLog) (r *model.RedoLog) {
TableInfo: rv1.RedoDDL.DDL.TableInfo,
PreTableInfo: rv1.RedoDDL.DDL.PreTableInfo,
Type: rv1.RedoDDL.DDL.Type,
Done: rv1.RedoDDL.DDL.Done,
}
r.RedoDDL.DDL.Done.Store(rv1.RedoDDL.DDL.Done)
}
return
}
Expand Down
3 changes: 2 additions & 1 deletion cdc/model/sink.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ import (
"fmt"
"strconv"
"sync"
"sync/atomic"
"unsafe"

"github.com/pingcap/log"
Expand Down Expand Up @@ -610,7 +611,7 @@ type DDLEvent struct {
TableInfo *TableInfo `msg:"-"`
PreTableInfo *TableInfo `msg:"-"`
Type model.ActionType `msg:"-"`
Done bool `msg:"-"`
Done atomic.Bool `msg:"-"`
Charset string `msg:"-"`
Collate string `msg:"-"`
}
Expand Down
32 changes: 20 additions & 12 deletions cdc/owner/changefeed.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,7 +129,11 @@ type changefeed struct {
filter filter.Filter,
) (puller.DDLPuller, error)

newSink func(changefeedID model.ChangeFeedID, info *model.ChangeFeedInfo, reportErr func(error)) DDLSink
newSink func(
changefeedID model.ChangeFeedID, info *model.ChangeFeedInfo,
reportError func(err error), reportWarning func(err error),
) DDLSink

newScheduler func(
ctx cdcContext.Context, up *upstream.Upstream, epoch uint64, cfg *config.SchedulerConfig,
) (scheduler.Scheduler, error)
Expand Down Expand Up @@ -179,7 +183,10 @@ func newChangefeed4Test(
schemaStorage entry.SchemaStorage,
filter filter.Filter,
) (puller.DDLPuller, error),
newSink func(changefeedID model.ChangeFeedID, info *model.ChangeFeedInfo, reportErr func(err error)) DDLSink,
newSink func(
Copy link
Member

Choose a reason for hiding this comment

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

Maybe we can give it a type, then we don't need to copy and paste it a couple of times.

changefeedID model.ChangeFeedID, info *model.ChangeFeedInfo,
reportError func(err error), reportWarning func(err error),
) DDLSink,
newScheduler func(
ctx cdcContext.Context, up *upstream.Upstream, epoch uint64, cfg *config.SchedulerConfig,
) (scheduler.Scheduler, error),
Expand Down Expand Up @@ -296,11 +303,7 @@ func (c *changefeed) tick(ctx cdcContext.Context, captures map[model.CaptureID]*
return errors.Trace(err)
default:
}
// we need to wait ddl ddlSink to be ready before we do the other things
// otherwise, we may cause a nil pointer panic when we try to write to the ddl ddlSink.
if !c.ddlSink.isInitialized() {
return nil
}

hicqu marked this conversation as resolved.
Show resolved Hide resolved
// TODO: pass table checkpointTs when we support concurrent process ddl
allPhysicalTables, minTableBarrierTs, barrier, err := c.ddlManager.tick(ctx, checkpointTs, nil)
if err != nil {
Expand Down Expand Up @@ -567,7 +570,13 @@ LOOP:
zap.String("changefeed", c.id.ID),
)

c.ddlSink = c.newSink(c.id, c.state.Info, ctx.Throw)
c.ddlSink = c.newSink(c.id, c.state.Info, ctx.Throw, func(err error) {
// TODO(qupeng): report the warning.
log.Warn("ddlSink internal error",
zap.String("namespace", c.id.Namespace),
zap.String("changefeed", c.id.ID),
zap.Error(err))
})
c.ddlSink.run(cancelCtx)

c.ddlPuller, err = c.newDDLPuller(cancelCtx,
Expand All @@ -586,8 +595,7 @@ LOOP:
ctx.Throw(c.ddlPuller.Run(cancelCtx))
}()

c.downstreamObserver, err = c.newDownstreamObserver(
ctx, c.state.Info.SinkURI, c.state.Info.Config)
c.downstreamObserver, err = c.newDownstreamObserver(ctx, c.state.Info.SinkURI, c.state.Info.Config)
if err != nil {
return err
}
Expand Down Expand Up @@ -1013,8 +1021,8 @@ func (c *changefeed) tickDownstreamObserver(ctx context.Context) {
defer cancel()
if err := c.downstreamObserver.Tick(cctx); err != nil {
// Prometheus is not deployed, it happens in non production env.
if strings.Contains(err.Error(),
fmt.Sprintf(":%d", errno.ErrPrometheusAddrIsNotSet)) {
noPrometheusMsg := fmt.Sprintf(":%d", errno.ErrPrometheusAddrIsNotSet)
if strings.Contains(err.Error(), noPrometheusMsg) {
return
}
log.Warn("backend observer tick error", zap.Error(err))
Expand Down
6 changes: 1 addition & 5 deletions cdc/owner/changefeed_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -150,10 +150,6 @@ func (m *mockDDLSink) close(ctx context.Context) error {
return nil
}

func (m *mockDDLSink) isInitialized() bool {
return true
}

func (m *mockDDLSink) Barrier(ctx context.Context) error {
return nil
}
Expand Down Expand Up @@ -219,7 +215,7 @@ func createChangefeed4Test(ctx cdcContext.Context, t *testing.T,
return &mockDDLPuller{resolvedTs: startTs - 1, schemaStorage: schemaStorage}, nil
},
// new ddl ddlSink
func(_ model.ChangeFeedID, _ *model.ChangeFeedInfo, _ func(err error)) DDLSink {
func(_ model.ChangeFeedID, _ *model.ChangeFeedInfo, _ func(error), _ func(error)) DDLSink {
return &mockDDLSink{
resetDDLDone: true,
recordDDLHistory: false,
Expand Down
Loading