Skip to content

Commit

Permalink
changefeedccl: add testing knob for async flush synchronization
Browse files Browse the repository at this point in the history
This commit adds a new changefeed testing knob, AsyncFlushSync, which
can be used to introduce a synchronization point between goroutines
during an async flush. It's currently only used in the cloud storage
sink.

Epic: none

Release note: none
  • Loading branch information
rharding6373 committed Sep 12, 2024
1 parent 30871f7 commit c4e4907
Show file tree
Hide file tree
Showing 2 changed files with 22 additions and 0 deletions.
19 changes: 19 additions & 0 deletions pkg/ccl/changefeedccl/sink_cloudstorage.go
Original file line number Diff line number Diff line change
Expand Up @@ -659,9 +659,17 @@ func (s *cloudStorageSink) flushTopicVersions(
}
return err == nil
})

// Files need to be cleared after the flush completes, otherwise file resources
for _, v := range toRemove {
s.files.Delete(cloudStorageSinkKey{topic: topic, schemaID: v})
}

// Allow synchronization with the async flusher to happen.
if s.testingKnobs != nil && s.testingKnobs.AsyncFlushSync != nil {
s.testingKnobs.AsyncFlushSync()
}

return err
}

Expand All @@ -682,6 +690,11 @@ func (s *cloudStorageSink) Flush(ctx context.Context) error {
return err
}
s.files.Clear(true /* addNodesToFreeList */)
// Allow synchronization with the async flusher to happen.
if s.testingKnobs != nil && s.testingKnobs.AsyncFlushSync != nil {
s.testingKnobs.AsyncFlushSync()
}

s.setDataFileTimestamp()
return s.waitAsyncFlush(ctx)
}
Expand Down Expand Up @@ -816,6 +829,12 @@ func (s *cloudStorageSink) asyncFlusher(ctx context.Context) error {
continue
}

// Allow synchronization with the flushing routine to happen between getting
// the flush request from the channel and completing the flush.
if s.testingKnobs != nil && s.testingKnobs.AsyncFlushSync != nil {
s.testingKnobs.AsyncFlushSync()
}

// flush file to storage.
flushDone := s.metrics.recordFlushRequestCallback()
err := req.file.flushToStorage(ctx, s.es, req.dest, s.metrics)
Expand Down
3 changes: 3 additions & 0 deletions pkg/ccl/changefeedccl/testing_knobs.go
Original file line number Diff line number Diff line change
Expand Up @@ -96,6 +96,9 @@ type TestingKnobs struct {

// OverrideExecCfg returns a modified ExecutorConfig to use under tests.
OverrideExecCfg func(actual *sql.ExecutorConfig) *sql.ExecutorConfig

// AsyncFlushSync is called in async flush goroutines as a way to provide synchronization between them.
AsyncFlushSync func()
}

// ModuleTestingKnobs is part of the base.ModuleTestingKnobs interface.
Expand Down

0 comments on commit c4e4907

Please sign in to comment.