redo(ticdc): use meta flush interval in redo ddl manager (#9999)

ref #9960, close #9998
pingcap · Nov 6, 2023 · 09b6870 · 09b6870
1 parent 999e90b
commit 09b6870
Show file tree

Hide file tree

Showing 2 changed files with 35 additions and 24 deletions.
diff --git a/cdc/redo/manager.go b/cdc/redo/manager.go
@@ -274,7 +274,15 @@ func (m *logManager) Run(ctx context.Context, _ ...chan<- error) error {
 		return err
 	}
 	m.writer = w
-	return m.bgUpdateLog(ctx)
+	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
 }
 
 // WaitForReady implements pkg/util.Runnable.
@@ -483,15 +491,14 @@ func (m *logManager) onResolvedTsMsg(span tablepb.Span, 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
@@ -114,11 +114,12 @@ func TestLogManagerInProcessor(t *testing.T) {
 	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,
+			Level:                 string(redo.ConsistentLevelEventual),
+			MaxLogSize:            redo.DefaultMaxLogSize,
+			Storage:               storage,
+			FlushIntervalInMs:     redo.MinFlushIntervalInMs,
+			MetaFlushIntervalInMs: redo.MinFlushIntervalInMs,
+			UseFileBackend:        useFileBackend,
 		}
 		dmlMgr := NewDMLManager(model.DefaultChangeFeedID("test"), cfg)
 		var eg errgroup.Group
@@ -221,11 +222,12 @@ func TestLogManagerInOwner(t *testing.T) {
 	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,
+			Level:                 string(redo.ConsistentLevelEventual),
+			MaxLogSize:            redo.DefaultMaxLogSize,
+			Storage:               storage,
+			FlushIntervalInMs:     redo.MinFlushIntervalInMs,
+			MetaFlushIntervalInMs: redo.DefaultMetaFlushIntervalInMs,
+			UseFileBackend:        useFileBackend,
 		}
 		startTs := model.Ts(10)
 		ddlMgr := NewDDLManager(model.DefaultChangeFeedID("test"), cfg, startTs)
@@ -266,10 +268,11 @@ func TestLogManagerError(t *testing.T) {
 	defer cancel()
 
 	cfg := &config.ConsistentConfig{
-		Level:             string(redo.ConsistentLevelEventual),
-		MaxLogSize:        redo.DefaultMaxLogSize,
-		Storage:           "blackhole-invalid://",
-		FlushIntervalInMs: redo.MinFlushIntervalInMs,
+		Level:                 string(redo.ConsistentLevelEventual),
+		MaxLogSize:            redo.DefaultMaxLogSize,
+		Storage:               "blackhole-invalid://",
+		FlushIntervalInMs:     redo.MinFlushIntervalInMs,
+		MetaFlushIntervalInMs: redo.MinFlushIntervalInMs,
 	}
 	logMgr := NewDMLManager(model.DefaultChangeFeedID("test"), cfg)
 	var eg errgroup.Group
@@ -317,11 +320,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 := NewDMLManager(model.DefaultChangeFeedID("test"), cfg)
 	var eg errgroup.Group