Make UT TestCloudStorageWriteEventsWithDateSeparator more stable (#10767

)

close #10764

cdc/sink/dmlsink/cloudstorage/cloud_storage_dml_sink_test.go

@@ -22,6 +22,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/pingcap/failpoint"
 	timodel "github.com/pingcap/tidb/pkg/parser/model"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/pingcap/tidb/pkg/parser/types"
@@ -190,7 +191,7 @@ func TestCloudStorageWriteEventsWithDateSeparator(t *testing.T) {
 
 	ctx, cancel := context.WithCancel(context.Background())
 	parentDir := t.TempDir()
-	uri := fmt.Sprintf("file:///%s?flush-interval=2s", parentDir)
+	uri := fmt.Sprintf("file:///%s?flush-interval=4s", parentDir)
 	sinkURI, err := url.Parse(uri)
 	require.Nil(t, err)
 
@@ -216,7 +217,7 @@ func TestCloudStorageWriteEventsWithDateSeparator(t *testing.T) {
 	tableDir := path.Join(parentDir, "test/table1/33/2023-03-08")
 	err = s.WriteEvents(txns...)
 	require.Nil(t, err)
-	time.Sleep(3 * time.Second)
+	time.Sleep(5 * time.Second)
 
 	fileNames := getTableFiles(t, tableDir)
 	require.Len(t, fileNames, 2)
@@ -236,7 +237,7 @@ func TestCloudStorageWriteEventsWithDateSeparator(t *testing.T) {
 
 	err = s.WriteEvents(txns...)
 	require.Nil(t, err)
-	time.Sleep(3 * time.Second)
+	time.Sleep(5 * time.Second)
 
 	fileNames = getTableFiles(t, tableDir)
 	require.Len(t, fileNames, 3)
@@ -254,9 +255,14 @@ func TestCloudStorageWriteEventsWithDateSeparator(t *testing.T) {
 	mockClock.Set(time.Date(2023, 3, 9, 0, 0, 10, 0, time.UTC))
 	setClock(s, mockClock)
 
+	failpoint.Enable("github.com/pingcap/tiflow/cdc/sink/dmlsink/cloudstorage/passTickerOnce", "1*return")
+	defer func() {
+		_ = failpoint.Disable("github.com/pingcap/tiflow/cdc/sink/dmlsink/cloudstorage/passTickerOnce")
+	}()
+
 	err = s.WriteEvents(txns...)
 	require.Nil(t, err)
-	time.Sleep(3 * time.Second)
+	time.Sleep(10 * time.Second)
 
 	tableDir = path.Join(parentDir, "test/table1/33/2023-03-09")
 	fileNames = getTableFiles(t, tableDir)
@@ -287,7 +293,7 @@ func TestCloudStorageWriteEventsWithDateSeparator(t *testing.T) {
 
 	err = s.WriteEvents(txns...)
 	require.Nil(t, err)
-	time.Sleep(3 * time.Second)
+	time.Sleep(5 * time.Second)
 
 	fileNames = getTableFiles(t, tableDir)
 	require.Len(t, fileNames, 3)

cdc/sink/dmlsink/cloudstorage/dml_worker.go

@@ -20,6 +20,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/br/pkg/storage"
 	"github.com/pingcap/tiflow/cdc/model"
@@ -321,6 +322,12 @@ func (d *dmlWorker) genAndDispatchTask(ctx context.Context,
 	ticker := time.NewTicker(d.config.FlushInterval)
 
 	for {
+		// this failpoint is use to pass this ticker once
+		// to make writeEvent in the test case can write into the same file
+		failpoint.Inject("passTickerOnce", func() {
+			<-ticker.C
+		})
+
 		select {
 		case <-ctx.Done():
 			return errors.Trace(ctx.Err())