Skip to content

Commit

Permalink
remove ctx from the ddl sink interface
Browse files Browse the repository at this point in the history
  • Loading branch information
3AceShowHand committed Sep 26, 2024
1 parent 939f3d2 commit 9b0f471
Show file tree
Hide file tree
Showing 4 changed files with 5 additions and 8 deletions.
6 changes: 1 addition & 5 deletions cdc/owner/changefeed.go
Original file line number Diff line number Diff line change
Expand Up @@ -789,11 +789,7 @@ func (c *changefeed) releaseResources(ctx context.Context) {
c.wg.Wait()

if c.ddlSink != nil {
canceledCtx, cancel := context.WithCancel(context.Background())
cancel()
// TODO(dongmen): remove ctx from func ddlSink.close(), it is useless.
// We don't need to wait ddlSink Close, pass a canceled context is ok
if err := c.ddlSink.close(canceledCtx); err != nil {
if err := c.ddlSink.close(); err != nil {
log.Warn("owner close ddlSink failed",
zap.String("namespace", c.id.Namespace),
zap.String("changefeed", c.id.ID),
Expand Down
2 changes: 1 addition & 1 deletion cdc/owner/changefeed_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -148,7 +148,7 @@ func (m *mockDDLSink) getCheckpointTsAndTableNames() (uint64, []*model.TableInfo
return m.mu.checkpointTs, m.mu.currentTables
}

func (m *mockDDLSink) close(_ context.Context) error {
func (m *mockDDLSink) close() error {
m.wg.Wait()
return nil
}
Expand Down
4 changes: 2 additions & 2 deletions cdc/owner/ddl_sink.go
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@ type DDLSink interface {
// It will return after the bootstrap event is sent.
emitBootstrap(ctx context.Context, bootstrap *model.DDLEvent) error
// close the ddlsink, cancel running goroutine.
close(ctx context.Context) error
close() error
}

type ddlSinkImpl struct {
Expand Down Expand Up @@ -406,7 +406,7 @@ func (s *ddlSinkImpl) emitSyncPoint(ctx context.Context, checkpointTs uint64) (e
}
}

func (s *ddlSinkImpl) close(ctx context.Context) (err error) {
func (s *ddlSinkImpl) close() (err error) {
s.cancel()
s.wg.Wait()

Expand Down
1 change: 1 addition & 0 deletions cdc/owner/owner.go
Original file line number Diff line number Diff line change
Expand Up @@ -483,6 +483,7 @@ func (o *ownerImpl) updateMetrics() {
ownershipCounter.Add(float64(now.Sub(o.lastTickTime)) / float64(time.Second))
o.lastTickTime = now

// todo: move this to each changefeed individually ?
for cfID, cf := range o.changefeeds {
if cf.latestInfo != nil {
changefeedStatusGauge.WithLabelValues(cfID.Namespace, cfID.ID).
Expand Down

0 comments on commit 9b0f471

Please # to comment.