fix redo.

cdc/redo/manager.go

@@ -65,9 +65,12 @@ func NewDDLManager(
 	cfg *config.ConsistentConfig, ddlStartTs model.Ts,
 ) *ddlManager {
 	m := newLogManager(changefeedID, cfg, redo.RedoDDLLogFileType)
-	m.AddTable(0, ddlStartTs)
+	tableID := int64(0)
+	m.AddTable(tableID, ddlStartTs)
 	return &ddlManager{
 		logManager: m,
+		// The current fakeTableID is meaningless, find a meaningful id in the future.
+		fakeTableID: tableID,
 	}
 }
 
@@ -105,8 +108,8 @@ type DMLManager interface {
 }
 
 // NewDMLManager creates a new dml Manager.
-func NewDMLManager(changefeedID model.ChangeFeedID,
-	cfg *config.ConsistentConfig,
+func NewDMLManager(
+	changefeedID model.ChangeFeedID, cfg *config.ConsistentConfig,
 ) *dmlManager {
 	return &dmlManager{
 		logManager: newLogManager(changefeedID, cfg, redo.RedoRowLogFileType),
@@ -216,8 +219,7 @@ type logManager struct {
 }
 
 func newLogManager(
-	changefeedID model.ChangeFeedID,
-	cfg *config.ConsistentConfig, logType string,
+	changefeedID model.ChangeFeedID, cfg *config.ConsistentConfig, logType string,
 ) *logManager {
 	// return a disabled Manager if no consistent config or normal consistent level
 	if cfg == nil || !redo.IsConsistentEnabled(cfg.Level) {

cdc/redo/meta_manager.go

@@ -22,7 +22,6 @@ import (
 
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/br/pkg/storage"
-	"github.com/pingcap/tiflow/cdc/contextutil"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/redo/common"
 	"github.com/pingcap/tiflow/pkg/config"
@@ -75,8 +74,6 @@ type metaManager struct {
 	lastFlushTime          time.Time
 	cfg                    *config.ConsistentConfig
 	metricFlushLogDuration prometheus.Observer
-
-	flushIntervalInMs int64
 }
 
 // NewDisabledMetaManager creates a disabled Meta Manager.
@@ -96,13 +93,12 @@ func NewMetaManager(
 	}
 
 	m := &metaManager{
-		captureID:         contextutil.CaptureAddrFromCtx(ctx),
-		changeFeedID:      contextutil.ChangefeedIDFromCtx(ctx),
-		uuidGenerator:     uuid.NewGenerator(),
-		enabled:           true,
-		cfg:               cfg,
-		startTs:           checkpoint,
-		flushIntervalInMs: cfg.FlushIntervalInMs,
+		changeFeedID:  changefeedID,
+		captureID:     config.GetGlobalServerConfig().AdvertiseAddr,
+		uuidGenerator: uuid.NewGenerator(),
+		enabled:       true,
+		cfg:           cfg,
+		startTs:       checkpoint,
 	}
 	return m
 }
@@ -266,7 +262,6 @@ func (m *metaManager) initMeta(ctx context.Context) error {
 		zap.String("changefeed", m.changeFeedID.ID),
 		zap.Uint64("checkpointTs", flushedMeta.CheckpointTs),
 		zap.Uint64("resolvedTs", flushedMeta.ResolvedTs))
-
 	return util.DeleteFilesInExtStorage(ctx, m.extStorage, toRemoveMetaFiles)
 }