From c4e4907b791b0e8dbd133bcc3f9401763b6591c5 Mon Sep 17 00:00:00 2001
From: rharding6373 <harding@cockroachlabs.com>
Date: Thu, 5 Sep 2024 21:20:57 -0700
Subject: [PATCH] changefeedccl: add testing knob for async flush
 synchronization

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
---
 pkg/ccl/changefeedccl/sink_cloudstorage.go | 19 +++++++++++++++++++
 pkg/ccl/changefeedccl/testing_knobs.go     |  3 +++
 2 files changed, 22 insertions(+)

diff --git a/pkg/ccl/changefeedccl/sink_cloudstorage.go b/pkg/ccl/changefeedccl/sink_cloudstorage.go
index f152f59f9058..26abbc72257f 100644
--- a/pkg/ccl/changefeedccl/sink_cloudstorage.go
+++ b/pkg/ccl/changefeedccl/sink_cloudstorage.go
@@ -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
 }
 
@@ -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)
 }
@@ -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)
diff --git a/pkg/ccl/changefeedccl/testing_knobs.go b/pkg/ccl/changefeedccl/testing_knobs.go
index 6f1bbfb5d7fa..89caf6f173c1 100644
--- a/pkg/ccl/changefeedccl/testing_knobs.go
+++ b/pkg/ccl/changefeedccl/testing_knobs.go
@@ -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.