From 1b62946e1f893fbaf8d9be906f15227dd45f2356 Mon Sep 17 00:00:00 2001 From: Jianyuan Jiang Date: Fri, 3 Nov 2023 22:36:39 +0800 Subject: [PATCH] redo(ticdc): add meta flush interval configuration (#9959) (#9971) (#10000) close pingcap/tiflow#9960 --- cdc/api/v2/model.go | 33 +-- cdc/api/v2/model_test.go | 11 +- cdc/owner/changefeed_test.go | 7 +- cdc/owner/feed_state_manager.go | 111 +++++---- cdc/owner/feed_state_manager_test.go | 116 ++++++++++ cdc/redo/manager.go | 28 ++- cdc/redo/manager_test.go | 287 +----------------------- cdc/redo/meta_manager.go | 8 +- cdc/redo/meta_manager_test.go | 27 ++- docs/swagger/docs.go | 3 + docs/swagger/swagger.json | 3 + docs/swagger/swagger.yaml | 2 + pkg/config/config_test_data.go | 3 + pkg/config/consistent.go | 20 +- pkg/config/replica_config.go | 11 +- pkg/redo/config.go | 2 + tests/integration_tests/api_v2/cases.go | 11 +- tests/integration_tests/api_v2/model.go | 11 +- 18 files changed, 306 insertions(+), 388 deletions(-) diff --git a/cdc/api/v2/model.go b/cdc/api/v2/model.go index 33a2b885c2d..1887709830d 100644 --- a/cdc/api/v2/model.go +++ b/cdc/api/v2/model.go @@ -255,11 +255,12 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig( } if c.Consistent != nil { res.Consistent = &config.ConsistentConfig{ - Level: c.Consistent.Level, - MaxLogSize: c.Consistent.MaxLogSize, - FlushIntervalInMs: c.Consistent.FlushIntervalInMs, - Storage: c.Consistent.Storage, - UseFileBackend: c.Consistent.UseFileBackend, + Level: c.Consistent.Level, + MaxLogSize: c.Consistent.MaxLogSize, + FlushIntervalInMs: c.Consistent.FlushIntervalInMs, + MetaFlushIntervalInMs: c.Consistent.MetaFlushIntervalInMs, + Storage: c.Consistent.Storage, + UseFileBackend: c.Consistent.UseFileBackend, } } if c.Sink != nil { @@ -455,11 +456,12 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig { } if cloned.Consistent != nil { res.Consistent = &ConsistentConfig{ - Level: cloned.Consistent.Level, - MaxLogSize: cloned.Consistent.MaxLogSize, - FlushIntervalInMs: cloned.Consistent.FlushIntervalInMs, - Storage: cloned.Consistent.Storage, - UseFileBackend: cloned.Consistent.UseFileBackend, + Level: cloned.Consistent.Level, + MaxLogSize: cloned.Consistent.MaxLogSize, + FlushIntervalInMs: cloned.Consistent.FlushIntervalInMs, + MetaFlushIntervalInMs: cloned.Consistent.MetaFlushIntervalInMs, + Storage: cloned.Consistent.Storage, + UseFileBackend: cloned.Consistent.UseFileBackend, } } if cloned.Mounter != nil { @@ -627,11 +629,12 @@ type ColumnSelector struct { // ConsistentConfig represents replication consistency config for a changefeed // This is a duplicate of config.ConsistentConfig type ConsistentConfig struct { - Level string `json:"level"` - MaxLogSize int64 `json:"max_log_size"` - FlushIntervalInMs int64 `json:"flush_interval"` - Storage string `json:"storage"` - UseFileBackend bool `json:"use_file_backend"` + Level string `json:"level,omitempty"` + MaxLogSize int64 `json:"max_log_size"` + FlushIntervalInMs int64 `json:"flush_interval"` + MetaFlushIntervalInMs int64 `json:"meta_flush_interval"` + Storage string `json:"storage,omitempty"` + UseFileBackend bool `json:"use_file_backend"` } // ChangefeedSchedulerConfig is per changefeed scheduler settings. diff --git a/cdc/api/v2/model_test.go b/cdc/api/v2/model_test.go index 9382356a670..f04fa1386fd 100644 --- a/cdc/api/v2/model_test.go +++ b/cdc/api/v2/model_test.go @@ -56,11 +56,12 @@ var defaultAPIConfig = &ReplicaConfig{ AdvanceTimeoutInSec: config.DefaultAdvanceTimeoutInSec, }, Consistent: &ConsistentConfig{ - Level: "none", - MaxLogSize: 64, - FlushIntervalInMs: redo.DefaultFlushIntervalInMs, - Storage: "", - UseFileBackend: false, + Level: "none", + MaxLogSize: 64, + FlushIntervalInMs: redo.DefaultFlushIntervalInMs, + MetaFlushIntervalInMs: redo.DefaultMetaFlushIntervalInMs, + Storage: "", + UseFileBackend: false, }, } diff --git a/cdc/owner/changefeed_test.go b/cdc/owner/changefeed_test.go index 6d6d9536117..b257b1afba7 100644 --- a/cdc/owner/changefeed_test.go +++ b/cdc/owner/changefeed_test.go @@ -504,9 +504,10 @@ func TestRemoveChangefeed(t *testing.T) { info := ctx.ChangefeedVars().Info dir := t.TempDir() info.Config.Consistent = &config.ConsistentConfig{ - Level: "eventual", - Storage: filepath.Join("nfs://", dir), - FlushIntervalInMs: redo.DefaultFlushIntervalInMs, + Level: "eventual", + Storage: filepath.Join("nfs://", dir), + FlushIntervalInMs: redo.DefaultFlushIntervalInMs, + MetaFlushIntervalInMs: redo.DefaultMetaFlushIntervalInMs, } ctx = cdcContext.WithChangefeedVars(ctx, &cdcContext.ChangefeedVars{ ID: ctx.ChangefeedVars().ID, diff --git a/cdc/owner/feed_state_manager.go b/cdc/owner/feed_state_manager.go index 30c62e86aff..69676a10e11 100644 --- a/cdc/owner/feed_state_manager.go +++ b/cdc/owner/feed_state_manager.go @@ -63,11 +63,11 @@ type feedStateManager struct { // resolvedTs and initCheckpointTs is for checking whether resolved timestamp // has been advanced or not. - resolvedTs model.Ts - initCheckpointTs model.Ts - + resolvedTs model.Ts + checkpointTs model.Ts checkpointTsAdvanced time.Time - lastCheckpointTs model.Ts + + changefeedErrorStuckDuration time.Duration } // newFeedStateManager creates feedStateManager and initialize the exponential backoff @@ -83,34 +83,61 @@ func newFeedStateManager(up *upstream.Upstream) *feedStateManager { // backoff will stop once the defaultBackoffMaxElapsedTime has elapsed. f.errBackoff.MaxElapsedTime = defaultBackoffMaxElapsedTime - f.resetErrRetry() + f.changefeedErrorStuckDuration = defaultBackoffMaxElapsedTime + f.resetErrRetry() + f.isRetrying = false return f } +func (m *feedStateManager) shouldRetry() bool { + // changefeed should not retry within [m.lastErrorRetryTime, m.lastErrorRetryTime + m.backoffInterval). + return time.Since(m.lastErrorRetryTime) >= m.backoffInterval +} + +func (m *feedStateManager) shouldFailWhenRetry() bool { + // retry the changefeed + m.backoffInterval = m.errBackoff.NextBackOff() + // NextBackOff() will return -1 once the MaxElapsedTime has elapsed, + // set the changefeed to failed state. + if m.backoffInterval == m.errBackoff.Stop { + return true + } + + m.lastErrorRetryTime = time.Now() + return false +} + +// resetErrRetry reset the error retry related fields +func (m *feedStateManager) resetErrRetry() { + m.errBackoff.Reset() + m.backoffInterval = m.errBackoff.NextBackOff() + m.lastErrorRetryTime = time.Unix(0, 0) +} + func (m *feedStateManager) Tick( state *orchestrator.ChangefeedReactorState, resolvedTs model.Ts, ) (adminJobPending bool) { + m.checkAndInitLastRetryCheckpointTs(state.Status) + if state.Status != nil { - if m.lastCheckpointTs < state.Status.CheckpointTs { - m.lastCheckpointTs = state.Status.CheckpointTs + if m.checkpointTs < state.Status.CheckpointTs { + m.checkpointTs = state.Status.CheckpointTs m.checkpointTsAdvanced = time.Now() } - if m.state == nil || m.state.Status == nil { - // It's the first time `m.state.Status` gets filled. - m.initCheckpointTs = state.Status.CheckpointTs + if m.resolvedTs < resolvedTs { + m.resolvedTs = resolvedTs + } + if m.checkpointTs >= m.resolvedTs { + m.checkpointTsAdvanced = time.Now() } } - - m.checkAndInitLastRetryCheckpointTs(state.Status) - m.state = state - m.resolvedTs = resolvedTs m.shouldBeRunning = true defer func() { if !m.shouldBeRunning { - m.cleanUpTaskPositions() + m.cleanUp() } }() if m.handleAdminJob() { @@ -131,16 +158,12 @@ func (m *feedStateManager) Tick( m.shouldBeRunning = false return case model.StatePending: - if time.Since(m.lastErrorRetryTime) < m.backoffInterval { + if !m.shouldRetry() { m.shouldBeRunning = false return } - // retry the changefeed - oldBackoffInterval := m.backoffInterval - m.backoffInterval = m.errBackoff.NextBackOff() - // NextBackOff() will return -1 once the MaxElapsedTime has elapsed, - // set the changefeed to failed state. - if m.backoffInterval == m.errBackoff.Stop { + + if m.shouldFailWhenRetry() { log.Error("The changefeed won't be restarted as it has been experiencing failures for "+ "an extended duration", zap.Duration("maxElapsedTime", m.errBackoff.MaxElapsedTime), @@ -154,18 +177,17 @@ func (m *feedStateManager) Tick( return } - m.lastErrorRetryTime = time.Now() + // retry the changefeed + m.shouldBeRunning = true if m.state.Status != nil { m.lastErrorRetryCheckpointTs = m.state.Status.CheckpointTs } - m.shouldBeRunning = true m.patchState(model.StateWarning) log.Info("changefeed retry backoff interval is elapsed,"+ "chengefeed will be restarted", zap.String("namespace", m.state.ID.Namespace), zap.String("changefeed", m.state.ID.ID), zap.Time("lastErrorRetryTime", m.lastErrorRetryTime), - zap.Duration("lastRetryInterval", oldBackoffInterval), zap.Duration("nextRetryInterval", m.backoffInterval)) case model.StateNormal, model.StateWarning: m.checkAndChangeState() @@ -274,7 +296,7 @@ func (m *feedStateManager) handleAdminJob() (jobsPending bool) { m.shouldBeRunning = true // when the changefeed is manually resumed, we must reset the backoff m.resetErrRetry() - // The lastErrorTime also needs to be cleared before a fresh run. + m.isRetrying = false jobsPending = true m.patchState(model.StateNormal) @@ -407,12 +429,15 @@ func (m *feedStateManager) patchState(feedState model.FeedState) { }) } -func (m *feedStateManager) cleanUpTaskPositions() { +func (m *feedStateManager) cleanUp() { for captureID := range m.state.TaskPositions { m.state.PatchTaskPosition(captureID, func(position *model.TaskPosition) (*model.TaskPosition, bool, error) { return nil, position != nil, nil }) } + m.checkpointTs = 0 + m.checkpointTsAdvanced = time.Time{} + m.resolvedTs = 0 } func (m *feedStateManager) errorsReportedByProcessors() []*model.RunningError { @@ -535,12 +560,12 @@ func (m *feedStateManager) handleError(errs ...*model.RunningError) { info.Error = lastError return info, true, nil }) - } - // The errBackoff needs to be reset before the first retry. - if !m.isRetrying { - m.resetErrRetry() - m.isRetrying = true + // The errBackoff needs to be reset before the first retry. + if !m.isRetrying { + m.resetErrRetry() + m.isRetrying = true + } } } @@ -554,13 +579,14 @@ func (m *feedStateManager) handleWarning(errs ...*model.RunningError) { currTime := m.upstream.PDClock.CurrentTime() ckptTime := oracle.GetTimeFromTS(m.state.Status.CheckpointTs) m.lastWarningReportCheckpointTs = m.state.Status.CheckpointTs - // Conditions: - // 1. checkpoint lag is large enough; - // 2. checkpoint hasn't been advanced for a long while; - // 3. the changefeed has been initialized. - if currTime.Sub(ckptTime) > defaultBackoffMaxElapsedTime && - time.Since(m.checkpointTsAdvanced) > defaultBackoffMaxElapsedTime && - m.resolvedTs > m.initCheckpointTs { + + checkpointTsStuck := time.Since(m.checkpointTsAdvanced) > m.changefeedErrorStuckDuration + if checkpointTsStuck { + log.Info("changefeed retry on warning for a very long time and does not resume, "+ + "it will be failed", zap.String("changefeed", m.state.ID.ID), + zap.Uint64("checkpointTs", m.state.Status.CheckpointTs), + zap.Duration("checkpointTime", currTime.Sub(ckptTime)), + ) code, _ := cerrors.RFCCode(cerrors.ErrChangefeedUnretryable) m.handleError(&model.RunningError{ Time: lastError.Time, @@ -592,13 +618,6 @@ func GenerateChangefeedEpoch(ctx context.Context, pdClient pd.Client) uint64 { return oracle.ComposeTS(phyTs, logical) } -// resetErrRetry reset the error retry related fields -func (m *feedStateManager) resetErrRetry() { - m.errBackoff.Reset() - m.backoffInterval = m.errBackoff.NextBackOff() - m.lastErrorRetryTime = time.Unix(0, 0) -} - // checkAndChangeState checks the state of the changefeed and change it if needed. // if the state of the changefeed is warning and the changefeed's checkpointTs is // greater than the lastRetryCheckpointTs, it will change the state to normal. diff --git a/cdc/owner/feed_state_manager_test.go b/cdc/owner/feed_state_manager_test.go index ed738e04b55..6710063f9fe 100644 --- a/cdc/owner/feed_state_manager_test.go +++ b/cdc/owner/feed_state_manager_test.go @@ -64,6 +64,8 @@ func newFeedStateManager4Test( f.resetErrRetry() + f.changefeedErrorStuckDuration = time.Second * 3 + return f } @@ -208,6 +210,7 @@ func TestResumeChangefeedWithCheckpointTs(t *testing.T) { require.Equal(t, state.Status.AdminJobType, model.AdminStop) // resume the changefeed in failed state + manager.isRetrying = true manager.PushAdminJob(&model.AdminJob{ CfID: ctx.ChangefeedVars().ID, Type: model.AdminResume, @@ -220,6 +223,7 @@ func TestResumeChangefeedWithCheckpointTs(t *testing.T) { require.Equal(t, state.Info.State, model.StateNormal) require.Equal(t, state.Info.AdminJobType, model.AdminNone) require.Equal(t, state.Status.AdminJobType, model.AdminNone) + require.False(t, manager.isRetrying) } func TestMarkFinished(t *testing.T) { @@ -743,6 +747,7 @@ func TestUpdateChangefeedEpoch(t *testing.T) { func TestHandleWarning(t *testing.T) { ctx := cdcContext.NewBackendContext4Test(true) manager := newFeedStateManager4Test(200, 1600, 0, 2.0) + manager.changefeedErrorStuckDuration = 100 * time.Millisecond state := orchestrator.NewChangefeedReactorState(etcd.DefaultCDCClusterID, ctx.ChangefeedVars().ID) tester := orchestrator.NewReactorStateTester(t, state, nil) @@ -935,3 +940,114 @@ func TestErrorAfterWarning(t *testing.T) { require.Equal(t, model.StateWarning, state.Info.State) require.True(t, manager.ShouldRunning()) } + +func TestHandleWarningWhileAdvanceResolvedTs(t *testing.T) { + t.Parallel() + + maxElapsedTimeInMs := 2000 + ctx := cdcContext.NewBackendContext4Test(true) + manager := newFeedStateManager4Test(200, 1600, maxElapsedTimeInMs, 2.0) + state := orchestrator.NewChangefeedReactorState(etcd.DefaultCDCClusterID, + ctx.ChangefeedVars().ID) + tester := orchestrator.NewReactorStateTester(t, state, nil) + state.PatchInfo(func(info *model.ChangeFeedInfo) (*model.ChangeFeedInfo, bool, error) { + require.Nil(t, info) + return &model.ChangeFeedInfo{SinkURI: "123", Config: &config.ReplicaConfig{}}, true, nil + }) + state.PatchStatus(func(status *model.ChangeFeedStatus) (*model.ChangeFeedStatus, bool, error) { + require.Nil(t, status) + return &model.ChangeFeedStatus{ + CheckpointTs: 200, + }, true, nil + }) + + tester.MustApplyPatches() + manager.Tick(state, 200) + tester.MustApplyPatches() + require.Equal(t, model.StateNormal, state.Info.State) + require.True(t, manager.ShouldRunning()) + + // 1. test when an warning occurs, the changefeed state will be changed to warning + // and it will still keep running + state.PatchTaskPosition(ctx.GlobalVars().CaptureInfo.ID, + func(position *model.TaskPosition) (*model.TaskPosition, bool, error) { + return &model.TaskPosition{Warning: &model.RunningError{ + Addr: ctx.GlobalVars().CaptureInfo.AdvertiseAddr, + Code: "[CDC:ErrSinkManagerRunError]", // it is fake error + Message: "fake error for test", + }}, true, nil + }) + tester.MustApplyPatches() + manager.Tick(state, 200) + // some patches will be generated when the manager.Tick is called + // so we need to apply the patches before we check the state + tester.MustApplyPatches() + require.Equal(t, model.StateWarning, state.Info.State) + require.True(t, manager.ShouldRunning()) + + // 2. test when the changefeed is in warning state, and the resolvedTs and checkpointTs is not progressing, + // the changefeed state will remain warning whena new warning is encountered. + time.Sleep(manager.changefeedErrorStuckDuration + 10) + state.PatchStatus(func(status *model.ChangeFeedStatus) (*model.ChangeFeedStatus, bool, error) { + require.NotNil(t, status) + return &model.ChangeFeedStatus{ + CheckpointTs: 200, + }, true, nil + }) + state.PatchTaskPosition(ctx.GlobalVars().CaptureInfo.ID, + func(position *model.TaskPosition) (*model.TaskPosition, bool, error) { + return &model.TaskPosition{Warning: &model.RunningError{ + Addr: ctx.GlobalVars().CaptureInfo.AdvertiseAddr, + Code: "[CDC:ErrSinkManagerRunError]", // it is fake error + Message: "fake error for test", + }}, true, nil + }) + tester.MustApplyPatches() + manager.Tick(state, 200) + tester.MustApplyPatches() + require.Equal(t, model.StateWarning, state.Info.State) + require.True(t, manager.ShouldRunning()) + + // 3. Test changefeed remain warning when resolvedTs is progressing after stuck beyond the detection time. + state.PatchStatus(func(status *model.ChangeFeedStatus) (*model.ChangeFeedStatus, bool, error) { + require.NotNil(t, status) + return &model.ChangeFeedStatus{ + CheckpointTs: 200, + }, true, nil + }) + state.PatchTaskPosition(ctx.GlobalVars().CaptureInfo.ID, + func(position *model.TaskPosition) (*model.TaskPosition, bool, error) { + return &model.TaskPosition{Warning: &model.RunningError{ + Addr: ctx.GlobalVars().CaptureInfo.AdvertiseAddr, + Code: "[CDC:ErrSinkManagerRunError]", // it is fake error + Message: "fake error for test", + }}, true, nil + }) + tester.MustApplyPatches() + manager.Tick(state, 400) + tester.MustApplyPatches() + require.Equal(t, model.StateWarning, state.Info.State) + require.True(t, manager.ShouldRunning()) + + // 4. Test changefeed failed when checkpointTs is not progressing for changefeedErrorStuckDuration time. + time.Sleep(manager.changefeedErrorStuckDuration + 10) + state.PatchStatus(func(status *model.ChangeFeedStatus) (*model.ChangeFeedStatus, bool, error) { + require.NotNil(t, status) + return &model.ChangeFeedStatus{ + CheckpointTs: 200, + }, true, nil + }) + state.PatchTaskPosition(ctx.GlobalVars().CaptureInfo.ID, + func(position *model.TaskPosition) (*model.TaskPosition, bool, error) { + return &model.TaskPosition{Warning: &model.RunningError{ + Addr: ctx.GlobalVars().CaptureInfo.AdvertiseAddr, + Code: "[CDC:ErrSinkManagerRunError]", // it is fake error + Message: "fake error for test", + }}, true, nil + }) + tester.MustApplyPatches() + manager.Tick(state, 400) + tester.MustApplyPatches() + require.Equal(t, model.StateFailed, state.Info.State) + require.False(t, manager.ShouldRunning()) +} diff --git a/cdc/redo/manager.go b/cdc/redo/manager.go index 7abea626afb..53b3f81c608 100644 --- a/cdc/redo/manager.go +++ b/cdc/redo/manager.go @@ -265,7 +265,26 @@ func newLogManager( func (m *logManager) Run(ctx context.Context) error { defer m.close() - return m.bgUpdateLog(ctx) + start := time.Now() + w, err := factory.NewRedoLogWriter(ctx, m.cfg) + if err != nil { + log.Error("redo: failed to create redo log writer", + zap.String("namespace", m.cfg.ChangeFeedID.Namespace), + zap.String("changefeed", m.cfg.ChangeFeedID.ID), + zap.Duration("duration", time.Since(start)), + zap.Error(err)) + return err + } + m.writer = w + return m.bgUpdateLog(ctx, m.getFlushDuration()) +} + +func (m *logManager) getFlushDuration() time.Duration { + flushIntervalInMs := m.cfg.FlushIntervalInMs + if m.cfg.LogType == redo.RedoDDLLogFileType { + flushIntervalInMs = m.cfg.MetaFlushIntervalInMs + } + return time.Duration(flushIntervalInMs) * time.Millisecond } // Enabled returns whether this log manager is enabled @@ -468,15 +487,14 @@ func (m *logManager) onResolvedTsMsg(tableID model.TableID, resolvedTs model.Ts) } } -func (m *logManager) bgUpdateLog(ctx context.Context) error { +func (m *logManager) bgUpdateLog(ctx context.Context, flushDuration time.Duration) error { m.releaseMemoryCbs = make([]func(), 0, 1024) - flushIntervalInMs := m.cfg.FlushIntervalInMs - ticker := time.NewTicker(time.Duration(flushIntervalInMs) * time.Millisecond) + ticker := time.NewTicker(flushDuration) defer ticker.Stop() log.Info("redo manager bgUpdateLog is running", zap.String("namespace", m.cfg.ChangeFeedID.Namespace), zap.String("changefeed", m.cfg.ChangeFeedID.ID), - zap.Int64("flushIntervalInMs", flushIntervalInMs)) + zap.Duration("flushIntervalInMs", flushDuration)) var err error // logErrCh is used to retrieve errors from log flushing goroutines. diff --git a/cdc/redo/manager_test.go b/cdc/redo/manager_test.go index a84d5a5a4dc..9d29fe3d74d 100644 --- a/cdc/redo/manager_test.go +++ b/cdc/redo/manager_test.go @@ -16,7 +16,6 @@ package redo import ( "context" "fmt" - "math" "math/rand" "sync" "testing" @@ -24,8 +23,6 @@ import ( "github.com/pingcap/log" "github.com/pingcap/tiflow/cdc/model" - "github.com/pingcap/tiflow/cdc/redo/writer" - "github.com/pingcap/tiflow/cdc/redo/writer/blackhole" "github.com/pingcap/tiflow/pkg/config" "github.com/pingcap/tiflow/pkg/redo" "github.com/stretchr/testify/require" @@ -33,279 +30,6 @@ import ( "golang.org/x/sync/errgroup" ) -func checkResolvedTs(t *testing.T, mgr *logManager, expectedRts uint64) { - time.Sleep(time.Duration(redo.MinFlushIntervalInMs+200) * time.Millisecond) - resolvedTs := uint64(math.MaxUint64) - mgr.rtsMap.Range(func(tableID any, value any) bool { - v, ok := value.(*statefulRts) - require.True(t, ok) - ts := v.getFlushed() - if ts < resolvedTs { - resolvedTs = ts - } - return true - }) - require.Equal(t, expectedRts, resolvedTs) -} - -func TestConsistentConfig(t *testing.T) { - t.Parallel() - levelCases := []struct { - level string - valid bool - }{ - {"none", true}, - {"eventual", true}, - {"NONE", false}, - {"", false}, - } - for _, lc := range levelCases { - require.Equal(t, lc.valid, redo.IsValidConsistentLevel(lc.level)) - } - - levelEnableCases := []struct { - level string - consistent bool - }{ - {"invalid-level", false}, - {"none", false}, - {"eventual", true}, - } - for _, lc := range levelEnableCases { - require.Equal(t, lc.consistent, redo.IsConsistentEnabled(lc.level)) - } - - storageCases := []struct { - storage string - valid bool - }{ - {"local", true}, - {"nfs", true}, - {"s3", true}, - {"blackhole", true}, - {"Local", false}, - {"", false}, - } - for _, sc := range storageCases { - require.Equal(t, sc.valid, redo.IsValidConsistentStorage(sc.storage)) - } - - s3StorageCases := []struct { - storage string - s3Enabled bool - }{ - {"local", false}, - {"nfs", false}, - {"s3", true}, - {"blackhole", false}, - } - for _, sc := range s3StorageCases { - require.Equal(t, sc.s3Enabled, redo.IsExternalStorage(sc.storage)) - } -} - -// TestLogManagerInProcessor tests how redo log manager is used in processor. -func TestLogManagerInProcessor(t *testing.T) { - t.Parallel() - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - - testWriteDMLs := func(storage string, useFileBackend bool) { - ctx, cancel := context.WithCancel(ctx) - cfg := &config.ConsistentConfig{ - Level: string(redo.ConsistentLevelEventual), - MaxLogSize: redo.DefaultMaxLogSize, - Storage: storage, - FlushIntervalInMs: redo.MinFlushIntervalInMs, - UseFileBackend: useFileBackend, - } - dmlMgr, err := NewDMLManager(ctx, cfg) - require.NoError(t, err) - wg := sync.WaitGroup{} - wg.Add(1) - go func() { - defer wg.Done() - dmlMgr.Run(ctx) - }() - - // check emit row changed events can move forward resolved ts - spans := []model.TableID{53, 55, 57, 59} - - startTs := uint64(100) - for _, span := range spans { - dmlMgr.AddTable(span, startTs) - } - testCases := []struct { - span model.TableID - rows []*model.RowChangedEvent - }{ - { - span: 53, - rows: []*model.RowChangedEvent{ - {CommitTs: 120, Table: &model.TableName{TableID: 53}}, - {CommitTs: 125, Table: &model.TableName{TableID: 53}}, - {CommitTs: 130, Table: &model.TableName{TableID: 53}}, - }, - }, - { - span: 55, - rows: []*model.RowChangedEvent{ - {CommitTs: 130, Table: &model.TableName{TableID: 55}}, - {CommitTs: 135, Table: &model.TableName{TableID: 55}}, - }, - }, - { - span: 57, - rows: []*model.RowChangedEvent{ - {CommitTs: 130, Table: &model.TableName{TableID: 57}}, - }, - }, - { - span: 59, - rows: []*model.RowChangedEvent{ - {CommitTs: 128, Table: &model.TableName{TableID: 59}}, - {CommitTs: 130, Table: &model.TableName{TableID: 59}}, - {CommitTs: 133, Table: &model.TableName{TableID: 59}}, - }, - }, - } - for _, tc := range testCases { - err := dmlMgr.EmitRowChangedEvents(ctx, tc.span, nil, tc.rows...) - require.NoError(t, err) - } - - // check UpdateResolvedTs can move forward the resolved ts when there is not row event. - flushResolvedTs := uint64(150) - for _, span := range spans { - checkResolvedTs(t, dmlMgr.logManager, startTs) - err := dmlMgr.UpdateResolvedTs(ctx, span, flushResolvedTs) - require.NoError(t, err) - } - checkResolvedTs(t, dmlMgr.logManager, flushResolvedTs) - - // check remove table can work normally - removeTable := spans[len(spans)-1] - spans = spans[:len(spans)-1] - dmlMgr.RemoveTable(removeTable) - flushResolvedTs = uint64(200) - for _, span := range spans { - err := dmlMgr.UpdateResolvedTs(ctx, span, flushResolvedTs) - require.NoError(t, err) - } - checkResolvedTs(t, dmlMgr.logManager, flushResolvedTs) - - cancel() - wg.Wait() - } - - testWriteDMLs("blackhole://", true) - storages := []string{ - fmt.Sprintf("file://%s", t.TempDir()), - fmt.Sprintf("nfs://%s", t.TempDir()), - } - for _, storage := range storages { - testWriteDMLs(storage, true) - testWriteDMLs(storage, false) - } -} - -// TestLogManagerInOwner tests how redo log manager is used in owner, -// where the redo log manager needs to handle DDL event only. -func TestLogManagerInOwner(t *testing.T) { - t.Parallel() - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - - testWriteDDLs := func(storage string, useFileBackend bool) { - ctx, cancel := context.WithCancel(ctx) - cfg := &config.ConsistentConfig{ - Level: string(redo.ConsistentLevelEventual), - MaxLogSize: redo.DefaultMaxLogSize, - Storage: storage, - FlushIntervalInMs: redo.MinFlushIntervalInMs, - UseFileBackend: useFileBackend, - } - startTs := model.Ts(10) - ddlMgr, err := NewDDLManager(ctx, cfg, startTs) - require.NoError(t, err) - wg := sync.WaitGroup{} - wg.Add(1) - go func() { - defer wg.Done() - ddlMgr.Run(ctx) - }() - - require.Equal(t, startTs, ddlMgr.GetResolvedTs()) - ddl := &model.DDLEvent{StartTs: 100, CommitTs: 120, Query: "CREATE TABLE `TEST.T1`"} - err = ddlMgr.EmitDDLEvent(ctx, ddl) - require.NoError(t, err) - require.Equal(t, startTs, ddlMgr.GetResolvedTs()) - - ddlMgr.UpdateResolvedTs(ctx, ddl.CommitTs) - checkResolvedTs(t, ddlMgr.logManager, ddl.CommitTs) - - cancel() - wg.Wait() - } - - testWriteDDLs("blackhole://", true) - storages := []string{ - fmt.Sprintf("file://%s", t.TempDir()), - fmt.Sprintf("nfs://%s", t.TempDir()), - } - for _, storage := range storages { - testWriteDDLs(storage, true) - testWriteDDLs(storage, false) - } -} - -// TestManagerError tests whether internal error in bgUpdateLog could be managed correctly. -func TestLogManagerError(t *testing.T) { - t.Parallel() - ctx, cancel := context.WithTimeout(context.Background(), time.Second*5) - defer cancel() - - cfg := &config.ConsistentConfig{ - Level: string(redo.ConsistentLevelEventual), - MaxLogSize: redo.DefaultMaxLogSize, - Storage: "blackhole://", - FlushIntervalInMs: redo.MinFlushIntervalInMs, - } - logMgr, err := NewDMLManager(ctx, cfg) - require.NoError(t, err) - err = logMgr.writer.Close() - require.NoError(t, err) - logMgr.writer = blackhole.NewInvalidLogWriter(logMgr.writer) - - wg := sync.WaitGroup{} - wg.Add(1) - go func() { - defer wg.Done() - err := logMgr.Run(ctx) - require.Regexp(t, ".*invalid black hole writer.*", err) - require.Regexp(t, ".*WriteLog.*", err) - }() - - testCases := []struct { - span model.TableID - rows []writer.RedoEvent - }{ - { - span: 53, - rows: []writer.RedoEvent{ - &model.RowChangedEvent{CommitTs: 120, Table: &model.TableName{TableID: 53}}, - &model.RowChangedEvent{CommitTs: 125, Table: &model.TableName{TableID: 53}}, - &model.RowChangedEvent{CommitTs: 130, Table: &model.TableName{TableID: 53}}, - }, - }, - } - for _, tc := range testCases { - err := logMgr.emitRedoEvents(ctx, tc.span, nil, tc.rows...) - require.NoError(t, err) - } - wg.Wait() -} - func BenchmarkBlackhole(b *testing.B) { runBenchTest(b, "blackhole://", false) } @@ -323,11 +47,12 @@ func BenchmarkFileWriter(b *testing.B) { func runBenchTest(b *testing.B, storage string, useFileBackend bool) { ctx, cancel := context.WithCancel(context.Background()) cfg := &config.ConsistentConfig{ - Level: string(redo.ConsistentLevelEventual), - MaxLogSize: redo.DefaultMaxLogSize, - Storage: storage, - FlushIntervalInMs: redo.MinFlushIntervalInMs, - UseFileBackend: useFileBackend, + Level: string(redo.ConsistentLevelEventual), + MaxLogSize: redo.DefaultMaxLogSize, + Storage: storage, + FlushIntervalInMs: redo.MinFlushIntervalInMs, + MetaFlushIntervalInMs: redo.MinFlushIntervalInMs, + UseFileBackend: useFileBackend, } dmlMgr, err := NewDMLManager(ctx, cfg) require.Nil(b, err) diff --git a/cdc/redo/meta_manager.go b/cdc/redo/meta_manager.go index 231f6db99e8..86ce58313eb 100644 --- a/cdc/redo/meta_manager.go +++ b/cdc/redo/meta_manager.go @@ -120,7 +120,13 @@ func NewMetaManager(ctx context.Context, cfg *config.ConsistentConfig) (*metaMan changeFeedID: contextutil.ChangefeedIDFromCtx(ctx), uuidGenerator: uuid.NewGenerator(), enabled: true, - flushIntervalInMs: cfg.FlushIntervalInMs, + flushIntervalInMs: cfg.MetaFlushIntervalInMs, + } + + if m.flushIntervalInMs < redo.MinFlushIntervalInMs { + log.Warn("redo flush interval is too small, use default value", + zap.Int64("interval", m.flushIntervalInMs)) + m.flushIntervalInMs = redo.DefaultMetaFlushIntervalInMs } uri, err := storage.ParseRawURL(cfg.Storage) diff --git a/cdc/redo/meta_manager_test.go b/cdc/redo/meta_manager_test.go index 3a13fdccb35..e630d19f249 100644 --- a/cdc/redo/meta_manager_test.go +++ b/cdc/redo/meta_manager_test.go @@ -72,10 +72,11 @@ func TestInitAndWriteMeta(t *testing.T) { startTs := uint64(10) cfg := &config.ConsistentConfig{ - Level: string(redo.ConsistentLevelEventual), - MaxLogSize: redo.DefaultMaxLogSize, - Storage: uri.String(), - FlushIntervalInMs: redo.MinFlushIntervalInMs, + Level: string(redo.ConsistentLevelEventual), + MaxLogSize: redo.DefaultMaxLogSize, + Storage: uri.String(), + FlushIntervalInMs: redo.MinFlushIntervalInMs, + MetaFlushIntervalInMs: redo.MinFlushIntervalInMs, } m, err := NewMetaManagerWithInit(ctx, cfg, startTs) require.NoError(t, err) @@ -136,10 +137,11 @@ func TestPreCleanupAndWriteMeta(t *testing.T) { startTs := uint64(10) cfg := &config.ConsistentConfig{ - Level: string(redo.ConsistentLevelEventual), - MaxLogSize: redo.DefaultMaxLogSize, - Storage: uri.String(), - FlushIntervalInMs: redo.MinFlushIntervalInMs, + Level: string(redo.ConsistentLevelEventual), + MaxLogSize: redo.DefaultMaxLogSize, + Storage: uri.String(), + FlushIntervalInMs: redo.MinFlushIntervalInMs, + MetaFlushIntervalInMs: redo.MinFlushIntervalInMs, } m, err := NewMetaManagerWithInit(ctx, cfg, startTs) require.NoError(t, err) @@ -267,10 +269,11 @@ func TestGCAndCleanup(t *testing.T) { startTs := uint64(3) cfg := &config.ConsistentConfig{ - Level: string(redo.ConsistentLevelEventual), - MaxLogSize: redo.DefaultMaxLogSize, - Storage: uri.String(), - FlushIntervalInMs: redo.MinFlushIntervalInMs, + Level: string(redo.ConsistentLevelEventual), + MaxLogSize: redo.DefaultMaxLogSize, + Storage: uri.String(), + FlushIntervalInMs: redo.MinFlushIntervalInMs, + MetaFlushIntervalInMs: redo.MinFlushIntervalInMs, } m, err := NewMetaManagerWithInit(ctx, cfg, startTs) require.NoError(t, err) diff --git a/docs/swagger/docs.go b/docs/swagger/docs.go index 6482a027f8c..e9def7b0406 100644 --- a/docs/swagger/docs.go +++ b/docs/swagger/docs.go @@ -1764,6 +1764,9 @@ var doc = `{ "max_log_size": { "type": "integer" }, + "meta_flush_interval": { + "type": "integer" + }, "storage": { "type": "string" }, diff --git a/docs/swagger/swagger.json b/docs/swagger/swagger.json index 59d27bf5b9b..8eb9fec990b 100644 --- a/docs/swagger/swagger.json +++ b/docs/swagger/swagger.json @@ -1745,6 +1745,9 @@ "max_log_size": { "type": "integer" }, + "meta_flush_interval": { + "type": "integer" + }, "storage": { "type": "string" }, diff --git a/docs/swagger/swagger.yaml b/docs/swagger/swagger.yaml index 753fcd376bd..3b84a6495f2 100644 --- a/docs/swagger/swagger.yaml +++ b/docs/swagger/swagger.yaml @@ -392,6 +392,8 @@ definitions: type: string max_log_size: type: integer + meta_flush_interval: + type: integer storage: type: string use_file_backend: diff --git a/pkg/config/config_test_data.go b/pkg/config/config_test_data.go index 57a5b6bb01e..2d1be90650d 100644 --- a/pkg/config/config_test_data.go +++ b/pkg/config/config_test_data.go @@ -60,6 +60,7 @@ const ( "level": "none", "max-log-size": 64, "flush-interval": 2000, + "meta-flush-interval": 200, "storage": "", "use-file-backend": false } @@ -211,6 +212,7 @@ const ( "level": "none", "max-log-size": 64, "flush-interval": 2000, + "meta-flush-interval": 200, "storage": "", "use-file-backend": false } @@ -271,6 +273,7 @@ const ( "level": "none", "max-log-size": 64, "flush-interval": 2000, + "meta-flush-interval": 200, "storage": "", "use-file-backend": false } diff --git a/pkg/config/consistent.go b/pkg/config/consistent.go index 636edcf865f..4a521f98da5 100644 --- a/pkg/config/consistent.go +++ b/pkg/config/consistent.go @@ -24,11 +24,12 @@ import ( // ConsistentConfig represents replication consistency config for a changefeed. type ConsistentConfig struct { - Level string `toml:"level" json:"level"` - MaxLogSize int64 `toml:"max-log-size" json:"max-log-size"` - FlushIntervalInMs int64 `toml:"flush-interval" json:"flush-interval"` - Storage string `toml:"storage" json:"storage"` - UseFileBackend bool `toml:"use-file-backend" json:"use-file-backend"` + Level string `toml:"level" json:"level"` + MaxLogSize int64 `toml:"max-log-size" json:"max-log-size"` + FlushIntervalInMs int64 `toml:"flush-interval" json:"flush-interval"` + MetaFlushIntervalInMs int64 `toml:"meta-flush-interval" json:"meta-flush-interval"` + Storage string `toml:"storage" json:"storage"` + UseFileBackend bool `toml:"use-file-backend" json:"use-file-backend"` } // ValidateAndAdjust validates the consistency config and adjusts it if necessary. @@ -50,6 +51,15 @@ func (c *ConsistentConfig) ValidateAndAdjust() error { c.FlushIntervalInMs, redo.MinFlushIntervalInMs)) } + if c.MetaFlushIntervalInMs == 0 { + c.MetaFlushIntervalInMs = redo.DefaultMetaFlushIntervalInMs + } + if c.MetaFlushIntervalInMs < redo.MinFlushIntervalInMs { + return cerror.ErrInvalidReplicaConfig.FastGenByArgs( + fmt.Sprintf("The consistent.meta-flush-interval:%d must be equal or greater than %d", + c.MetaFlushIntervalInMs, redo.MinFlushIntervalInMs)) + } + uri, err := storage.ParseRawURL(c.Storage) if err != nil { return cerror.ErrInvalidReplicaConfig.GenWithStackByArgs( diff --git a/pkg/config/replica_config.go b/pkg/config/replica_config.go index a602056300b..6dde6c2df06 100644 --- a/pkg/config/replica_config.go +++ b/pkg/config/replica_config.go @@ -65,11 +65,12 @@ var defaultReplicaConfig = &ReplicaConfig{ AdvanceTimeoutInSec: DefaultAdvanceTimeoutInSec, }, Consistent: &ConsistentConfig{ - Level: "none", - MaxLogSize: redo.DefaultMaxLogSize, - FlushIntervalInMs: redo.DefaultFlushIntervalInMs, - Storage: "", - UseFileBackend: false, + Level: "none", + MaxLogSize: redo.DefaultMaxLogSize, + FlushIntervalInMs: redo.DefaultFlushIntervalInMs, + MetaFlushIntervalInMs: redo.DefaultMetaFlushIntervalInMs, + Storage: "", + UseFileBackend: false, }, } diff --git a/pkg/redo/config.go b/pkg/redo/config.go index fbcdb3f1eb3..d0562b83be5 100644 --- a/pkg/redo/config.go +++ b/pkg/redo/config.go @@ -44,6 +44,8 @@ const ( FlushWarnDuration = time.Second * 20 // DefaultFlushIntervalInMs is the default flush interval for redo log. DefaultFlushIntervalInMs = 2000 + // DefaultMetaFlushIntervalInMs is the default flush interval for redo meta. + DefaultMetaFlushIntervalInMs = 200 // MinFlushIntervalInMs is the minimum flush interval for redo log. MinFlushIntervalInMs = 50 diff --git a/tests/integration_tests/api_v2/cases.go b/tests/integration_tests/api_v2/cases.go index 36f7fbe0df9..e5d90e4e88a 100644 --- a/tests/integration_tests/api_v2/cases.go +++ b/tests/integration_tests/api_v2/cases.go @@ -125,11 +125,12 @@ var defaultReplicaConfig = &ReplicaConfig{ EnablePartitionSeparator: true, }, Consistent: &ConsistentConfig{ - Level: "none", - MaxLogSize: redo.DefaultMaxLogSize, - FlushIntervalInMs: redo.DefaultFlushIntervalInMs, - Storage: "", - UseFileBackend: false, + Level: "none", + MaxLogSize: redo.DefaultMaxLogSize, + FlushIntervalInMs: redo.DefaultFlushIntervalInMs, + MetaFlushIntervalInMs: redo.DefaultMetaFlushIntervalInMs, + Storage: "", + UseFileBackend: false, }, } diff --git a/tests/integration_tests/api_v2/model.go b/tests/integration_tests/api_v2/model.go index b25e8e5355c..e6b669aabbe 100644 --- a/tests/integration_tests/api_v2/model.go +++ b/tests/integration_tests/api_v2/model.go @@ -271,11 +271,12 @@ type ColumnSelector struct { // ConsistentConfig represents replication consistency config for a changefeed // This is a duplicate of config.ConsistentConfig type ConsistentConfig struct { - Level string `json:"level"` - MaxLogSize int64 `json:"max_log_size"` - FlushIntervalInMs int64 `json:"flush_interval"` - Storage string `json:"storage"` - UseFileBackend bool `json:"use_file_backend"` + Level string `json:"level"` + MaxLogSize int64 `json:"max_log_size"` + FlushIntervalInMs int64 `json:"flush_interval"` + MetaFlushIntervalInMs int64 `json:"meta_flush_interval"` + Storage string `json:"storage"` + UseFileBackend bool `json:"use_file_backend"` } // ChangefeedSchedulerConfig is per changefeed scheduler settings.