diff --git a/cdc/api/v2/model.go b/cdc/api/v2/model.go index ce68469636c..0564fb0d1ee 100644 --- a/cdc/api/v2/model.go +++ b/cdc/api/v2/model.go @@ -271,6 +271,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 { @@ -757,6 +758,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 { @@ -953,6 +955,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. diff --git a/cdc/redo/writer/memory/file_worker.go b/cdc/redo/writer/memory/file_worker.go index 3571a102c9e..0816f2b0d4c 100644 --- a/cdc/redo/writer/memory/file_worker.go +++ b/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) { diff --git a/pkg/config/consistent.go b/pkg/config/consistent.go index 26f4950b4b3..c8f93c6eac8 100644 --- a/pkg/config/consistent.go +++ b/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. diff --git a/pkg/util/external_storage.go b/pkg/util/external_storage.go index 19be69c0126..d4786fb240a 100644 --- a/pkg/util/external_storage.go +++ b/pkg/util/external_storage.go @@ -196,8 +196,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) } diff --git a/tests/integration_tests/consistent_replicate_storage_s3/conf/changefeed.toml b/tests/integration_tests/consistent_replicate_storage_s3/conf/changefeed.toml index 7edf7be7d69..ff99477c9da 100644 --- a/tests/integration_tests/consistent_replicate_storage_s3/conf/changefeed.toml +++ b/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