redo(ticdc): use multi part s3 uploader in redo (#10227)

cdc/api/v2/model.go

@@ -272,6 +272,7 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 			Storage:               c.Consistent.Storage,
 			UseFileBackend:        c.Consistent.UseFileBackend,
 			Compression:           c.Consistent.Compression,
+			FlushConcurrency:      c.Consistent.FlushConcurrency,
 		}
 	}
 	if c.Sink != nil {
@@ -502,6 +503,7 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 			Storage:               cloned.Consistent.Storage,
 			UseFileBackend:        cloned.Consistent.UseFileBackend,
 			Compression:           cloned.Consistent.Compression,
+			FlushConcurrency:      cloned.Consistent.FlushConcurrency,
 		}
 	}
 	if cloned.Mounter != nil {
@@ -690,6 +692,7 @@ type ConsistentConfig struct {
 	Storage               string `json:"storage,omitempty"`
 	UseFileBackend        bool   `json:"use_file_backend"`
 	Compression           string `json:"compression,omitempty"`
+	FlushConcurrency      int    `json:"flush_concurrency,omitempty"`
 }
 
 // ChangefeedSchedulerConfig is per changefeed scheduler settings.

cdc/redo/writer/memory/file_worker.go

@@ -190,7 +190,12 @@ func (f *fileWorkerGroup) bgFlushFileCache(egCtx context.Context) error {
 			if err := file.writer.Close(); err != nil {
 				return errors.Trace(err)
 			}
-			err := f.extStorage.WriteFile(egCtx, file.filename, file.writer.buf.Bytes())
+			var err error
+			if f.cfg.FlushConcurrency <= 1 {
+				err = f.extStorage.WriteFile(egCtx, file.filename, file.writer.buf.Bytes())
+			} else {
+				err = f.multiPartUpload(egCtx, file)
+			}
 			f.metricFlushAllDuration.Observe(time.Since(start).Seconds())
 			if err != nil {
 				return errors.Trace(err)
@@ -204,6 +209,19 @@ func (f *fileWorkerGroup) bgFlushFileCache(egCtx context.Context) error {
 	}
 }
 
+func (f *fileWorkerGroup) multiPartUpload(ctx context.Context, file *fileCache) error {
+	multipartWrite, err := f.extStorage.Create(ctx, file.filename, &storage.WriterOption{
+		Concurrency: f.cfg.FlushConcurrency,
+	})
+	if err != nil {
+		return errors.Trace(err)
+	}
+	if _, err = multipartWrite.Write(ctx, file.writer.buf.Bytes()); err != nil {
+		return errors.Trace(err)
+	}
+	return errors.Trace(multipartWrite.Close(ctx))
+}
+
 func (f *fileWorkerGroup) bgWriteLogs(
 	egCtx context.Context, inputCh <-chan *polymorphicRedoEvent,
 ) (err error) {

pkg/config/consistent.go

@@ -34,6 +34,7 @@ type ConsistentConfig struct {
 	Storage               string `toml:"storage" json:"storage"`
 	UseFileBackend        bool   `toml:"use-file-backend" json:"use-file-backend"`
 	Compression           string `toml:"compression" json:"compression"`
+	FlushConcurrency      int    `toml:"flush-concurrency" json:"flush-concurrency,omitempty"`
 }
 
 // ValidateAndAdjust validates the consistency config and adjusts it if necessary.

pkg/util/external_storage.go

@@ -195,8 +195,12 @@ func (s *extStorageWithTimeout) WalkDir(
 func (s *extStorageWithTimeout) Create(
 	ctx context.Context, path string, option *storage.WriterOption,
 ) (storage.ExternalFileWriter, error) {
-	ctx, cancel := context.WithTimeout(ctx, s.timeout)
-	defer cancel()
+	if option.Concurrency <= 1 {
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeout(ctx, s.timeout)
+		defer cancel()
+	}
+	// multipart uploading spawns a background goroutine, can't set timeout
 	return s.ExternalStorage.Create(ctx, path, option)
 }
 

tests/integration_tests/consistent_replicate_storage_s3/conf/changefeed.toml

@@ -1,3 +1,4 @@
 [consistent]
 level = "eventual"
 storage = "s3://logbucket/test-changefeed?endpoint=http://127.0.0.1:24927/"
+flush-concurrency = 2