From 39759ab345e9b27ee80cc1f2c51ade9396be1d73 Mon Sep 17 00:00:00 2001 From: Ti Chi Robot Date: Thu, 28 Mar 2024 19:53:18 +0800 Subject: [PATCH] sink(ticdc): use pd clock in storage sink (#10351) (#10519) close pingcap/tiflow#10352, ref pingcap/tiflow#10374 --- cdc/api/v1/validator.go | 4 +- cdc/api/v2/api_helpers.go | 4 +- cdc/processor/processor.go | 2 +- cdc/processor/sinkmanager/manager.go | 2 +- cdc/sink/sink_test.go | 6 +- cdc/sink/validator.go | 5 +- .../cloudstorage/cloud_storage_dml_sink.go | 6 +- .../cloud_storage_dml_sink_test.go | 16 ++--- .../eventsink/cloudstorage/dml_worker.go | 6 +- .../eventsink/cloudstorage/dml_worker_test.go | 4 +- cdc/sinkv2/eventsink/factory/factory.go | 3 +- cmd/storage-consumer/main.go | 1 + errors.toml | 5 ++ pkg/applier/redo.go | 2 +- pkg/errors/cdc_errors.go | 4 ++ pkg/pdutil/clock.go | 22 +++++++ pkg/sink/cloudstorage/path.go | 59 ++++++++++++------- pkg/sink/cloudstorage/path_test.go | 14 +++-- 18 files changed, 112 insertions(+), 53 deletions(-) diff --git a/cdc/api/v1/validator.go b/cdc/api/v1/validator.go index 42e2095b874..3d1d5b4dd63 100644 --- a/cdc/api/v1/validator.go +++ b/cdc/api/v1/validator.go @@ -170,7 +170,7 @@ func verifyCreateChangefeedConfig( return nil, cerror.ErrAPIInvalidParam.Wrap(errors.Annotatef(err, "invalid timezone:%s", changefeedConfig.TimeZone)) } ctx = contextutil.PutTimezoneInCtx(ctx, tz) - if err := sink.Validate(ctx, info.SinkURI, info.Config); err != nil { + if err := sink.Validate(ctx, info.SinkURI, info.Config, up.PDClock); err != nil { return nil, err } @@ -228,7 +228,7 @@ func VerifyUpdateChangefeedConfig(ctx context.Context, return nil, cerror.ErrChangefeedUpdateRefused.GenWithStackByCause(err) } - if err := sink.Validate(ctx, newInfo.SinkURI, newInfo.Config); err != nil { + if err := sink.Validate(ctx, newInfo.SinkURI, newInfo.Config, nil); err != nil { return nil, cerror.ErrChangefeedUpdateRefused.GenWithStackByCause(err) } } diff --git a/cdc/api/v2/api_helpers.go b/cdc/api/v2/api_helpers.go index c1f5978f903..fcbbd2cb2d0 100644 --- a/cdc/api/v2/api_helpers.go +++ b/cdc/api/v2/api_helpers.go @@ -212,7 +212,7 @@ func (APIV2HelpersImpl) verifyCreateChangefeedConfig( } // verify sink - if err := sink.Validate(ctx, cfg.SinkURI, replicaCfg); err != nil { + if err := sink.Validate(ctx, cfg.SinkURI, replicaCfg, nil); err != nil { return nil, err } @@ -321,7 +321,7 @@ func (APIV2HelpersImpl) verifyUpdateChangefeedConfig( return nil, nil, cerror.ErrChangefeedUpdateRefused.GenWithStackByCause(err) } - if err := sink.Validate(ctx, newInfo.SinkURI, newInfo.Config); err != nil { + if err := sink.Validate(ctx, newInfo.SinkURI, newInfo.Config, nil); err != nil { return nil, nil, cerror.ErrChangefeedUpdateRefused.GenWithStackByCause(err) } } diff --git a/cdc/processor/processor.go b/cdc/processor/processor.go index ea1158b7fed..462dccaed5a 100644 --- a/cdc/processor/processor.go +++ b/cdc/processor/processor.go @@ -878,7 +878,7 @@ func (p *processor) lazyInitImpl(ctx cdcContext.Context) error { log.Info("Try to create sinkV2") sinkV2Factory, err := factory.New(stdCtx, p.changefeed.Info.SinkURI, p.changefeed.Info.Config, - errCh) + errCh, p.upstream.PDClock) if err != nil { log.Error("processor creates sink failed", zap.String("namespace", p.changefeedID.Namespace), diff --git a/cdc/processor/sinkmanager/manager.go b/cdc/processor/sinkmanager/manager.go index 0dd6294704a..f2dac81d76e 100644 --- a/cdc/processor/sinkmanager/manager.go +++ b/cdc/processor/sinkmanager/manager.go @@ -367,7 +367,7 @@ func (m *SinkManager) initSinkFactory() (chan error, uint64) { return m.sinkFactory.errors, m.sinkFactory.version } - m.sinkFactory.f, err = factory.New(m.managerCtx, uri, cfg, m.sinkFactory.errors) + m.sinkFactory.f, err = factory.New(m.managerCtx, uri, cfg, m.sinkFactory.errors, m.up.PDClock) if err != nil { emitError(err) return m.sinkFactory.errors, m.sinkFactory.version diff --git a/cdc/sink/sink_test.go b/cdc/sink/sink_test.go index 1be87c50972..e2888237628 100644 --- a/cdc/sink/sink_test.go +++ b/cdc/sink/sink_test.go @@ -30,19 +30,19 @@ func TestValidateSink(t *testing.T) { // test sink uri error sinkURI := "mysql://root:111@127.0.0.1:3306/" - err := Validate(ctx, sinkURI, replicateConfig) + err := Validate(ctx, sinkURI, replicateConfig, nil) require.NotNil(t, err) require.Contains(t, err.Error(), "fail to open MySQL connection") // test sink uri right sinkURI = "blackhole://" - err = Validate(ctx, sinkURI, replicateConfig) + err = Validate(ctx, sinkURI, replicateConfig, nil) require.Nil(t, err) // test bdr mode error replicateConfig.BDRMode = true sinkURI = "blackhole://" - err = Validate(ctx, sinkURI, replicateConfig) + err = Validate(ctx, sinkURI, replicateConfig, nil) require.NotNil(t, err) require.Contains(t, err.Error(), "sink uri scheme is not supported in BDR mode") } diff --git a/cdc/sink/validator.go b/cdc/sink/validator.go index a13fc2921ca..f9c028b715a 100644 --- a/cdc/sink/validator.go +++ b/cdc/sink/validator.go @@ -22,6 +22,7 @@ import ( "github.com/pingcap/tiflow/cdc/sinkv2/eventsink/factory" "github.com/pingcap/tiflow/pkg/config" cerror "github.com/pingcap/tiflow/pkg/errors" + "github.com/pingcap/tiflow/pkg/pdutil" "github.com/pingcap/tiflow/pkg/sink" pmysql "github.com/pingcap/tiflow/pkg/sink/mysql" "github.com/pingcap/tiflow/pkg/util" @@ -30,7 +31,7 @@ import ( // Validate sink if given valid parameters. // TODO: For now, we create a real sink instance and validate it. // Maybe we should support the dry-run mode to validate sink. -func Validate(ctx context.Context, sinkURI string, cfg *config.ReplicaConfig) error { +func Validate(ctx context.Context, sinkURI string, cfg *config.ReplicaConfig, pdClock pdutil.Clock) error { var err error var uri *url.URL if uri, err = preCheckSinkURI(sinkURI); err != nil { @@ -60,7 +61,7 @@ func Validate(ctx context.Context, sinkURI string, cfg *config.ReplicaConfig) er err = s.Close(ctx) } else { var s *factory.SinkFactory - s, err = factory.New(ctx, sinkURI, cfg, errCh) + s, err = factory.New(ctx, sinkURI, cfg, errCh, pdClock) if err != nil { cancel() return err diff --git a/cdc/sinkv2/eventsink/cloudstorage/cloud_storage_dml_sink.go b/cdc/sinkv2/eventsink/cloudstorage/cloud_storage_dml_sink.go index e86ccf2a008..3024d8bd700 100644 --- a/cdc/sinkv2/eventsink/cloudstorage/cloud_storage_dml_sink.go +++ b/cdc/sinkv2/eventsink/cloudstorage/cloud_storage_dml_sink.go @@ -29,10 +29,10 @@ import ( "github.com/pingcap/tiflow/cdc/sinkv2/metrics" "github.com/pingcap/tiflow/cdc/sinkv2/tablesink/state" "github.com/pingcap/tiflow/cdc/sinkv2/util" - "github.com/pingcap/tiflow/engine/pkg/clock" "github.com/pingcap/tiflow/pkg/chann" "github.com/pingcap/tiflow/pkg/config" cerror "github.com/pingcap/tiflow/pkg/errors" + "github.com/pingcap/tiflow/pkg/pdutil" "github.com/pingcap/tiflow/pkg/sink" "github.com/pingcap/tiflow/pkg/sink/cloudstorage" putil "github.com/pingcap/tiflow/pkg/util" @@ -94,6 +94,7 @@ type dmlSink struct { // NewCloudStorageSink creates a cloud storage sink. func NewCloudStorageSink( ctx context.Context, + pdClock pdutil.Clock, sinkURI *url.URL, replicaConfig *config.ReplicaConfig, errCh chan error, @@ -153,11 +154,10 @@ func NewCloudStorageSink( // create defragmenter. s.defragmenter = newDefragmenter(encodedCh, workerChannels) // create a group of dml workers. - clock := clock.New() for i := 0; i < cfg.WorkerCount; i++ { inputCh := chann.NewDrainableChann[eventFragment]() s.workers[i] = newDMLWorker(i, s.changefeedID, storage, cfg, ext, - inputCh, clock, s.statistics) + inputCh, pdClock, s.statistics) workerChannels[i] = inputCh } diff --git a/cdc/sinkv2/eventsink/cloudstorage/cloud_storage_dml_sink_test.go b/cdc/sinkv2/eventsink/cloudstorage/cloud_storage_dml_sink_test.go index 3639eefbfe7..77b84e6c5d7 100644 --- a/cdc/sinkv2/eventsink/cloudstorage/cloud_storage_dml_sink_test.go +++ b/cdc/sinkv2/eventsink/cloudstorage/cloud_storage_dml_sink_test.go @@ -31,12 +31,13 @@ import ( "github.com/pingcap/tiflow/cdc/sinkv2/tablesink/state" "github.com/pingcap/tiflow/engine/pkg/clock" "github.com/pingcap/tiflow/pkg/config" + "github.com/pingcap/tiflow/pkg/pdutil" "github.com/stretchr/testify/require" ) func setClock(s *dmlSink, clock clock.Clock) { for _, w := range s.workers { - w.filePathGenerator.SetClock(clock) + w.filePathGenerator.SetClock(pdutil.NewMonotonicClock(clock)) } } @@ -126,7 +127,7 @@ func TestCloudStorageWriteEventsWithoutDateSeparator(t *testing.T) { replicaConfig.Sink.Protocol = config.ProtocolCsv.String() replicaConfig.Sink.FileIndexWidth = 6 errCh := make(chan error, 5) - s, err := NewCloudStorageSink(ctx, sinkURI, replicaConfig, errCh) + s, err := NewCloudStorageSink(ctx, pdutil.NewMonotonicClock(clock.New()), sinkURI, replicaConfig, errCh) require.Nil(t, err) var cnt uint64 = 0 batch := 100 @@ -194,10 +195,9 @@ func TestCloudStorageWriteEventsWithDateSeparator(t *testing.T) { replicaConfig.Sink.FileIndexWidth = 6 errCh := make(chan error, 5) - s, err := NewCloudStorageSink(ctx, sinkURI, replicaConfig, errCh) - require.Nil(t, err) mockClock := clock.NewMock() - setClock(s, mockClock) + s, err := NewCloudStorageSink(ctx, pdutil.NewMonotonicClock(mockClock), sinkURI, replicaConfig, errCh) + require.Nil(t, err) var cnt uint64 = 0 batch := 100 @@ -268,11 +268,11 @@ func TestCloudStorageWriteEventsWithDateSeparator(t *testing.T) { // test table is scheduled from one node to another cnt = 0 ctx, cancel = context.WithCancel(context.Background()) - s, err = NewCloudStorageSink(ctx, sinkURI, replicaConfig, errCh) - require.Nil(t, err) mockClock = clock.NewMock() + + s, err = NewCloudStorageSink(ctx, pdutil.NewMonotonicClock(mockClock), sinkURI, replicaConfig, errCh) + require.Nil(t, err) mockClock.Set(time.Date(2023, 3, 9, 0, 1, 10, 0, time.UTC)) - setClock(s, mockClock) err = s.WriteEvents(txns...) require.Nil(t, err) diff --git a/cdc/sinkv2/eventsink/cloudstorage/dml_worker.go b/cdc/sinkv2/eventsink/cloudstorage/dml_worker.go index f700170f38d..cdf1565d58e 100644 --- a/cdc/sinkv2/eventsink/cloudstorage/dml_worker.go +++ b/cdc/sinkv2/eventsink/cloudstorage/dml_worker.go @@ -25,9 +25,9 @@ import ( "github.com/pingcap/tiflow/cdc/sink/codec/common" "github.com/pingcap/tiflow/cdc/sinkv2/metrics" mcloudstorage "github.com/pingcap/tiflow/cdc/sinkv2/metrics/cloudstorage" - "github.com/pingcap/tiflow/engine/pkg/clock" "github.com/pingcap/tiflow/pkg/chann" "github.com/pingcap/tiflow/pkg/errors" + "github.com/pingcap/tiflow/pkg/pdutil" "github.com/pingcap/tiflow/pkg/sink/cloudstorage" "github.com/prometheus/client_golang/prometheus" "go.uber.org/zap" @@ -103,7 +103,7 @@ func newDMLWorker( config *cloudstorage.Config, extension string, inputCh *chann.DrainableChann[eventFragment], - clock clock.Clock, + pdClock pdutil.Clock, statistics *metrics.Statistics, ) *dmlWorker { d := &dmlWorker{ @@ -114,7 +114,7 @@ func newDMLWorker( inputCh: inputCh, flushNotifyCh: make(chan dmlTask, 64), statistics: statistics, - filePathGenerator: cloudstorage.NewFilePathGenerator(config, storage, extension, clock), + filePathGenerator: cloudstorage.NewFilePathGenerator(changefeedID, config, storage, extension, pdClock), metricWriteBytes: mcloudstorage.CloudStorageWriteBytesGauge. WithLabelValues(changefeedID.Namespace, changefeedID.ID), metricFileCount: mcloudstorage.CloudStorageFileCountGauge. diff --git a/cdc/sinkv2/eventsink/cloudstorage/dml_worker_test.go b/cdc/sinkv2/eventsink/cloudstorage/dml_worker_test.go index cb09c78060c..c7e42882f0e 100644 --- a/cdc/sinkv2/eventsink/cloudstorage/dml_worker_test.go +++ b/cdc/sinkv2/eventsink/cloudstorage/dml_worker_test.go @@ -29,6 +29,7 @@ import ( "github.com/pingcap/tiflow/engine/pkg/clock" "github.com/pingcap/tiflow/pkg/chann" "github.com/pingcap/tiflow/pkg/config" + "github.com/pingcap/tiflow/pkg/pdutil" "github.com/pingcap/tiflow/pkg/sink" "github.com/pingcap/tiflow/pkg/sink/cloudstorage" "github.com/pingcap/tiflow/pkg/util" @@ -49,8 +50,9 @@ func testDMLWorker(ctx context.Context, t *testing.T, dir string) *dmlWorker { require.Nil(t, err) statistics := metrics.NewStatistics(ctx, sink.TxnSink) + pdlock := pdutil.NewMonotonicClock(clock.New()) d := newDMLWorker(1, model.DefaultChangeFeedID("dml-worker-test"), storage, - cfg, ".json", chann.NewDrainableChann[eventFragment](), clock.New(), statistics) + cfg, ".json", chann.NewDrainableChann[eventFragment](), pdlock, statistics) return d } diff --git a/cdc/sinkv2/eventsink/factory/factory.go b/cdc/sinkv2/eventsink/factory/factory.go index a0bbc4544ad..edc9450b792 100644 --- a/cdc/sinkv2/eventsink/factory/factory.go +++ b/cdc/sinkv2/eventsink/factory/factory.go @@ -64,6 +64,7 @@ func New(ctx context.Context, sinkURIStr string, cfg *config.ReplicaConfig, errCh chan error, + pdClock pdutil.Clock, ) (*SinkFactory, error) { sinkURI, err := config.GetSinkURIAndAdjustConfigWithSinkURI(sinkURIStr, cfg) if err != nil { @@ -91,7 +92,7 @@ func New(ctx context.Context, s.sinkType = sink.TxnSink s.category = CategoryMQ case sink.S3Scheme, sink.FileScheme, sink.GCSScheme, sink.GSScheme, sink.AzblobScheme, sink.AzureScheme, sink.CloudStorageNoopScheme: - storageSink, err := cloudstorage.NewCloudStorageSink(ctx, sinkURI, cfg, errCh) + storageSink, err := cloudstorage.NewCloudStorageSink(ctx, pdClock, sinkURI, cfg, errCh) if err != nil { return nil, err } diff --git a/cmd/storage-consumer/main.go b/cmd/storage-consumer/main.go index 554a180ad02..80d83d2c765 100644 --- a/cmd/storage-consumer/main.go +++ b/cmd/storage-consumer/main.go @@ -189,6 +189,7 @@ func newConsumer(ctx context.Context) (*consumer, error) { downstreamURIStr, config.GetDefaultReplicaConfig(), errCh, + nil, ) if err != nil { log.Error("failed to create event sink factory", zap.Error(err)) diff --git a/errors.toml b/errors.toml index 1a0a9aa3212..789613a1874 100755 --- a/errors.toml +++ b/errors.toml @@ -456,6 +456,11 @@ error = ''' table not found with index ID %d in index kv ''' +["CDC:ErrInternalCheckFailed"] +error = ''' +internal check failed, %s +''' + ["CDC:ErrInternalServerError"] error = ''' internal server error diff --git a/pkg/applier/redo.go b/pkg/applier/redo.go index 9f6e50936fa..bab52f842f1 100644 --- a/pkg/applier/redo.go +++ b/pkg/applier/redo.go @@ -122,7 +122,7 @@ func (ra *RedoApplier) catchError(ctx context.Context) error { func (ra *RedoApplier) initSink(ctx context.Context) (err error) { replicaConfig := config.GetDefaultReplicaConfig() - ra.sinkFactory, err = dmlfactory.New(ctx, ra.cfg.SinkURI, replicaConfig, ra.errCh) + ra.sinkFactory, err = dmlfactory.New(ctx, ra.cfg.SinkURI, replicaConfig, ra.errCh, nil) if err != nil { return err } diff --git a/pkg/errors/cdc_errors.go b/pkg/errors/cdc_errors.go index c6b28935910..a0b19e4cce5 100644 --- a/pkg/errors/cdc_errors.go +++ b/pkg/errors/cdc_errors.go @@ -1166,6 +1166,10 @@ var ( "invalid replica config, %s", errors.RFCCodeText("CDC:ErrInvalidReplicaConfig"), ) + ErrInternalCheckFailed = errors.Normalize( + "internal check failed, %s", + errors.RFCCodeText("CDC:ErrInternalCheckFailed"), + ) ErrHandleDDLFailed = errors.Normalize( "handle ddl failed, job: %s, query: %s, startTs: %d. "+ diff --git a/pkg/pdutil/clock.go b/pkg/pdutil/clock.go index fa79ec4074b..317e0564b44 100644 --- a/pkg/pdutil/clock.go +++ b/pkg/pdutil/clock.go @@ -20,6 +20,7 @@ import ( "github.com/pingcap/errors" "github.com/pingcap/log" + pclock "github.com/pingcap/tiflow/engine/pkg/clock" "github.com/pingcap/tiflow/pkg/retry" "github.com/tikv/client-go/v2/oracle" pd "github.com/tikv/pd/client" @@ -136,3 +137,24 @@ func (c *clock4Test) Run(ctx context.Context) { func (c *clock4Test) Stop() { } + +type monotonicClock struct { + clock pclock.Clock +} + +// NewMonotonicClock return a new monotonic clock. +func NewMonotonicClock(pClock pclock.Clock) Clock { + return &monotonicClock{ + clock: pClock, + } +} + +func (c *monotonicClock) CurrentTime() time.Time { + return c.clock.Now() +} + +func (c *monotonicClock) Run(ctx context.Context) { +} + +func (c *monotonicClock) Stop() { +} diff --git a/pkg/sink/cloudstorage/path.go b/pkg/sink/cloudstorage/path.go index dae3f5a8a0c..5a01bb9d6cf 100644 --- a/pkg/sink/cloudstorage/path.go +++ b/pkg/sink/cloudstorage/path.go @@ -33,6 +33,7 @@ import ( "github.com/pingcap/tiflow/pkg/config" "github.com/pingcap/tiflow/pkg/errors" "github.com/pingcap/tiflow/pkg/hash" + "github.com/pingcap/tiflow/pkg/pdutil" "github.com/pingcap/tiflow/pkg/util" "github.com/tikv/client-go/v2/oracle" "go.uber.org/zap" @@ -133,11 +134,12 @@ type VersionedTableName struct { // FilePathGenerator is used to generate data file path and index file path. type FilePathGenerator struct { - extension string - config *Config - clock clock.Clock - storage storage.ExternalStorage - fileIndex map[VersionedTableName]*indexWithDate + changefeedID model.ChangeFeedID + extension string + config *Config + pdClock pdutil.Clock + storage storage.ExternalStorage + fileIndex map[VersionedTableName]*indexWithDate hasher *hash.PositionInertia versionMap map[VersionedTableName]uint64 @@ -145,19 +147,27 @@ type FilePathGenerator struct { // NewFilePathGenerator creates a FilePathGenerator. func NewFilePathGenerator( + changefeedID model.ChangeFeedID, config *Config, storage storage.ExternalStorage, extension string, - clock clock.Clock, + pdclock pdutil.Clock, ) *FilePathGenerator { + if pdclock == nil { + pdclock = pdutil.NewMonotonicClock(clock.New()) + log.Warn("pd clock is not set in storage sink, use local clock instead", + zap.String("namespace", changefeedID.Namespace), + zap.String("changefeedID", changefeedID.ID)) + } return &FilePathGenerator{ - config: config, - extension: extension, - storage: storage, - clock: clock, - fileIndex: make(map[VersionedTableName]*indexWithDate), - hasher: hash.NewPositionInertia(), - versionMap: make(map[VersionedTableName]uint64), + changefeedID: changefeedID, + config: config, + extension: extension, + storage: storage, + pdClock: pdclock, + fileIndex: make(map[VersionedTableName]*indexWithDate), + hasher: hash.NewPositionInertia(), + versionMap: make(map[VersionedTableName]uint64), } } @@ -176,8 +186,12 @@ func (f *FilePathGenerator) CheckOrWriteSchema( def.FromTableInfo(tableInfo, table.TableInfoVersion) if !def.IsTableSchema() { // only check schema for table - log.Panic("invalid table schema", zap.Any("versionedTableName", table), + log.Error("invalid table schema", + zap.String("namespace", f.changefeedID.Namespace), + zap.String("changefeedID", f.changefeedID.ID), + zap.Any("versionedTableName", table), zap.Any("tableInfo", tableInfo)) + return errors.ErrInternalCheckFailed.GenWithStackByArgs("invalid table schema in FilePathGenerator") } // Case 1: point check if the schema file exists. @@ -210,10 +224,13 @@ func (f *FilePathGenerator) CheckOrWriteSchema( } version, parsedChecksum := mustParseSchemaName(path) if parsedChecksum != checksum { - // TODO: parsedChecksum should be ignored, remove this panic - // after the new path protocol is verified. - log.Panic("invalid schema file name", + log.Error("invalid schema file name", + zap.String("namespace", f.changefeedID.Namespace), + zap.String("changefeedID", f.changefeedID.ID), zap.String("path", path), zap.Any("checksum", checksum)) + errMsg := fmt.Sprintf("invalid schema filename in storage sink, "+ + "expected checksum: %d, actual checksum: %d", checksum, parsedChecksum) + return errors.ErrInternalCheckFailed.GenWithStackByArgs(errMsg) } if version > lastVersion { lastVersion = version @@ -235,6 +252,8 @@ func (f *FilePathGenerator) CheckOrWriteSchema( // b. the schema file is deleted by the consumer. We write schema file to external storage too. if schemaFileCnt != 0 && lastVersion == 0 { log.Warn("no table schema file found in an non-empty meta path", + zap.String("namespace", f.changefeedID.Namespace), + zap.String("changefeedID", f.changefeedID.ID), zap.Any("versionedTableName", table), zap.Uint32("checksum", checksum)) } @@ -247,8 +266,8 @@ func (f *FilePathGenerator) CheckOrWriteSchema( } // SetClock is used for unit test -func (f *FilePathGenerator) SetClock(clock clock.Clock) { - f.clock = clock +func (f *FilePathGenerator) SetClock(pdClock pdutil.Clock) { + f.pdClock = pdClock } // GenerateDateStr generates a date string base on current time @@ -256,7 +275,7 @@ func (f *FilePathGenerator) SetClock(clock clock.Clock) { func (f *FilePathGenerator) GenerateDateStr() string { var dateStr string - currTime := f.clock.Now() + currTime := f.pdClock.CurrentTime() switch f.config.DateSeparator { case config.DateSeparatorYear.String(): dateStr = currTime.Format("2006") diff --git a/pkg/sink/cloudstorage/path_test.go b/pkg/sink/cloudstorage/path_test.go index f8650441626..9ad360aa527 100644 --- a/pkg/sink/cloudstorage/path_test.go +++ b/pkg/sink/cloudstorage/path_test.go @@ -29,6 +29,7 @@ import ( "github.com/pingcap/tiflow/cdc/model" "github.com/pingcap/tiflow/engine/pkg/clock" "github.com/pingcap/tiflow/pkg/config" + "github.com/pingcap/tiflow/pkg/pdutil" "github.com/pingcap/tiflow/pkg/util" "github.com/stretchr/testify/require" "github.com/tikv/client-go/v2/oracle" @@ -49,7 +50,7 @@ func testFilePathGenerator(ctx context.Context, t *testing.T, dir string) *FileP err = cfg.Apply(ctx, sinkURI, replicaConfig) require.NoError(t, err) - f := NewFilePathGenerator(cfg, storage, ".json", clock.New()) + f := NewFilePathGenerator(model.ChangeFeedID{}, cfg, storage, ".json", pdutil.NewMonotonicClock(clock.New())) return f } @@ -84,7 +85,7 @@ func TestGenerateDataFilePath(t *testing.T) { f = testFilePathGenerator(ctx, t, dir) f.versionMap[table] = table.TableInfoVersion f.config.DateSeparator = config.DateSeparatorYear.String() - f.clock = mockClock + f.SetClock(pdutil.NewMonotonicClock(mockClock)) mockClock.Set(time.Date(2022, 12, 31, 23, 59, 59, 0, time.UTC)) date = f.GenerateDateStr() path, err = f.GenerateDataFilePath(ctx, table, date) @@ -108,7 +109,8 @@ func TestGenerateDataFilePath(t *testing.T) { f = testFilePathGenerator(ctx, t, dir) f.versionMap[table] = table.TableInfoVersion f.config.DateSeparator = config.DateSeparatorMonth.String() - f.clock = mockClock + f.SetClock(pdutil.NewMonotonicClock(mockClock)) + mockClock.Set(time.Date(2022, 12, 31, 23, 59, 59, 0, time.UTC)) date = f.GenerateDateStr() path, err = f.GenerateDataFilePath(ctx, table, date) @@ -132,7 +134,8 @@ func TestGenerateDataFilePath(t *testing.T) { f = testFilePathGenerator(ctx, t, dir) f.versionMap[table] = table.TableInfoVersion f.config.DateSeparator = config.DateSeparatorDay.String() - f.clock = mockClock + f.SetClock(pdutil.NewMonotonicClock(mockClock)) + mockClock.Set(time.Date(2022, 12, 31, 23, 59, 59, 0, time.UTC)) date = f.GenerateDateStr() path, err = f.GenerateDataFilePath(ctx, table, date) @@ -210,7 +213,8 @@ func TestGenerateDataFilePathWithIndexFile(t *testing.T) { f := testFilePathGenerator(ctx, t, dir) mockClock := clock.NewMock() f.config.DateSeparator = config.DateSeparatorDay.String() - f.clock = mockClock + f.SetClock(pdutil.NewMonotonicClock(mockClock)) + mockClock.Set(time.Date(2023, 3, 9, 23, 59, 59, 0, time.UTC)) table := VersionedTableName{ TableNameWithPhysicTableID: model.TableName{