This is an automated cherry-pick of pingcap#9887

Signed-off-by: ti-chi-bot <[email protected]>

cdc/owner/changefeed.go

@@ -336,6 +336,12 @@ func (c *changefeed) tick(ctx cdcContext.Context, captures map[model.CaptureID]*
 	default:
 	}
 
+	if c.redoMetaMgr.Enabled() {
+		if !c.redoMetaMgr.Running() {
+			return 0, 0, nil
+		}
+	}
+
 	// TODO: pass table checkpointTs when we support concurrent process ddl
 	allPhysicalTables, barrier, err := c.ddlManager.tick(ctx, preCheckpointTs, nil)
 	if err != nil {
@@ -578,6 +584,7 @@ LOOP2:
 	}
 	c.observerLastTick = atomic.NewTime(time.Time{})
 
+<<<<<<< HEAD
 	stdCtx := contextutil.PutChangefeedIDInCtx(cancelCtx, c.id)
 	c.redoDDLMgr, err = redo.NewDDLManager(stdCtx, c.state.Info.Config.Consistent, ddlStartTs)
 	failpoint.Inject("ChangefeedNewRedoManagerError", func() {
@@ -586,6 +593,9 @@ LOOP2:
 	if err != nil {
 		return err
 	}
+=======
+	c.redoDDLMgr = redo.NewDDLManager(c.id, c.latestInfo.Config.Consistent, ddlStartTs)
+>>>>>>> 684d117c67 (redo(ticdc): fix redo initialization block the owner (#9887))
 	if c.redoDDLMgr.Enabled() {
 		c.wg.Add(1)
 		go func() {
@@ -594,11 +604,15 @@ LOOP2:
 		}()
 	}
 
+<<<<<<< HEAD
 	c.redoMetaMgr, err = redo.NewMetaManagerWithInit(stdCtx,
 		c.state.Info.Config.Consistent, checkpointTs)
 	if err != nil {
 		return err
 	}
+=======
+	c.redoMetaMgr = redo.NewMetaManager(c.id, c.latestInfo.Config.Consistent, checkpointTs)
+>>>>>>> 684d117c67 (redo(ticdc): fix redo initialization block the owner (#9887))
 	if c.redoMetaMgr.Enabled() {
 		c.wg.Add(1)
 		go func() {
@@ -765,6 +779,7 @@ func (c *changefeed) cleanupRedoManager(ctx context.Context) {
 		}
 		// when removing a paused changefeed, the redo manager is nil, create a new one
 		if c.redoMetaMgr == nil {
+<<<<<<< HEAD
 			redoMetaMgr, err := redo.NewMetaManager(ctx, c.state.Info.Config.Consistent)
 			if err != nil {
 				log.Info("owner creates redo manager for clean fail",
@@ -774,6 +789,9 @@ func (c *changefeed) cleanupRedoManager(ctx context.Context) {
 				return
 			}
 			c.redoMetaMgr = redoMetaMgr
+=======
+			c.redoMetaMgr = redo.NewMetaManager(c.id, cfInfo.Config.Consistent, 0)
+>>>>>>> 684d117c67 (redo(ticdc): fix redo initialization block the owner (#9887))
 		}
 		err := c.redoMetaMgr.Cleanup(ctx)
 		if err != nil {

cdc/processor/processor.go

@@ -670,10 +670,14 @@ func (p *processor) lazyInitImpl(etcdCtx cdcContext.Context) (err error) {
 	}
 	p.changefeed.Info.Config.Sink.TiDBSourceID = sourceID
 
+<<<<<<< HEAD
 	p.redo.r, err = redo.NewDMLManager(stdCtx, p.changefeed.Info.Config.Consistent)
 	if err != nil {
 		return err
 	}
+=======
+	p.redo.r = redo.NewDMLManager(p.changefeedID, p.latestInfo.Config.Consistent)
+>>>>>>> 684d117c67 (redo(ticdc): fix redo initialization block the owner (#9887))
 	p.redo.name = "RedoManager"
 	p.redo.spawn(stdCtx)
 

cdc/processor/processor_test.go

@@ -70,14 +70,23 @@ func newProcessor4Test(
 		} else {
 			tmpDir := t.TempDir()
 			redoDir := fmt.Sprintf("%s/%s", tmpDir, changefeedID)
+<<<<<<< HEAD
 			dmlMgr, err := redo.NewDMLManager(ctx, &config.ConsistentConfig{
 				Level:             string(redoPkg.ConsistentLevelEventual),
 				MaxLogSize:        redoPkg.DefaultMaxLogSize,
 				FlushIntervalInMs: redoPkg.DefaultFlushIntervalInMs,
 				Storage:           "file://" + redoDir,
 				UseFileBackend:    false,
+=======
+			dmlMgr := redo.NewDMLManager(changefeedID, &config.ConsistentConfig{
+				Level:                 string(redoPkg.ConsistentLevelEventual),
+				MaxLogSize:            redoPkg.DefaultMaxLogSize,
+				FlushIntervalInMs:     redoPkg.DefaultFlushIntervalInMs,
+				MetaFlushIntervalInMs: redoPkg.DefaultMetaFlushIntervalInMs,
+				Storage:               "file://" + redoDir,
+				UseFileBackend:        false,
+>>>>>>> 684d117c67 (redo(ticdc): fix redo initialization block the owner (#9887))
 			})
-			require.NoError(t, err)
 			p.redo.r = dmlMgr
 		}
 		p.redo.name = "RedoManager"

cdc/redo/manager.go

@@ -19,9 +19,13 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/log"
+<<<<<<< HEAD
 	"github.com/pingcap/tidb/br/pkg/storage"
 	"github.com/pingcap/tiflow/cdc/contextutil"
+=======
+>>>>>>> 684d117c67 (redo(ticdc): fix redo initialization block the owner (#9887))
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/processor/tablepb"
 	"github.com/pingcap/tiflow/cdc/redo/common"
@@ -66,19 +70,26 @@ func NewDisabledDDLManager() *ddlManager {
 
 // NewDDLManager creates a new ddl Manager.
 func NewDDLManager(
+<<<<<<< HEAD
 	ctx context.Context, cfg *config.ConsistentConfig, ddlStartTs model.Ts,
 ) (*ddlManager, error) {
 	logManager, err := newLogManager(ctx, cfg, redo.RedoDDLLogFileType)
 	if err != nil {
 		return nil, err
 	}
+=======
+	changefeedID model.ChangeFeedID,
+	cfg *config.ConsistentConfig, ddlStartTs model.Ts,
+) *ddlManager {
+	m := newLogManager(changefeedID, cfg, redo.RedoDDLLogFileType)
+>>>>>>> 684d117c67 (redo(ticdc): fix redo initialization block the owner (#9887))
 	span := spanz.TableIDToComparableSpan(0)
-	logManager.AddTable(span, ddlStartTs)
+	m.AddTable(span, ddlStartTs)
 	return &ddlManager{
-		logManager: logManager,
-		// The current fakeSpan is meaningless, find a meaningful sapn in the future.
+		logManager: m,
+		// The current fakeSpan is meaningless, find a meaningful span in the future.
 		fakeSpan: span,
-	}, nil
+	}
 }
 
 type ddlManager struct {
@@ -115,12 +126,19 @@ type DMLManager interface {
 }
 
 // NewDMLManager creates a new dml Manager.
+<<<<<<< HEAD
 func NewDMLManager(ctx context.Context, cfg *config.ConsistentConfig) (*dmlManager, error) {
 	logManager, err := newLogManager(ctx, cfg, redo.RedoRowLogFileType)
 	if err != nil {
 		return nil, err
+=======
+func NewDMLManager(changefeedID model.ChangeFeedID,
+	cfg *config.ConsistentConfig,
+) *dmlManager {
+	return &dmlManager{
+		logManager: newLogManager(changefeedID, cfg, redo.RedoRowLogFileType),
+>>>>>>> 684d117c67 (redo(ticdc): fix redo initialization block the owner (#9887))
 	}
-	return &dmlManager{logManager: logManager}, nil
 }
 
 // NewDisabledDMLManager creates a disabled dml Manager.
@@ -226,13 +244,20 @@ type logManager struct {
 }
 
 func newLogManager(
+<<<<<<< HEAD
 	ctx context.Context, cfg *config.ConsistentConfig, logType string,
 ) (*logManager, error) {
+=======
+	changefeedID model.ChangeFeedID,
+	cfg *config.ConsistentConfig, logType string,
+) *logManager {
+>>>>>>> 684d117c67 (redo(ticdc): fix redo initialization block the owner (#9887))
 	// return a disabled Manager if no consistent config or normal consistent level
 	if cfg == nil || !redo.IsConsistentEnabled(cfg.Level) {
-		return &logManager{enabled: false}, nil
+		return &logManager{enabled: false}
 	}
 
+<<<<<<< HEAD
 	uri, err := storage.ParseRawURL(cfg.Storage)
 	if err != nil {
 		return nil, err
@@ -248,6 +273,16 @@ func newLogManager(
 			URI:                *uri,
 			UseExternalStorage: redo.IsExternalStorage(uri.Scheme),
 			MaxLogSizeInBytes:  cfg.MaxLogSize * redo.Megabyte,
+=======
+	return &logManager{
+		enabled: true,
+		cfg: &writer.LogWriterConfig{
+			ConsistentConfig:  *cfg,
+			LogType:           logType,
+			CaptureID:         config.GetGlobalServerConfig().AdvertiseAddr,
+			ChangeFeedID:      changefeedID,
+			MaxLogSizeInBytes: cfg.MaxLogSize * redo.Megabyte,
+>>>>>>> 684d117c67 (redo(ticdc): fix redo initialization block the owner (#9887))
 		},
 		logBuffer: chann.NewAutoDrainChann[cacheEvents](),
 		rtsMap:    spanz.SyncMap{},
@@ -260,21 +295,30 @@ func newLogManager(
 		metricRedoWorkerBusyRatio: common.RedoWorkerBusyRatio.
 			WithLabelValues(changefeedID.Namespace, changefeedID.ID),
 	}
-
-	m.writer, err = factory.NewRedoLogWriter(ctx, m.cfg)
-	if err != nil {
-		return nil, err
-	}
-	return m, nil
 }
 
 // Run implements pkg/util.Runnable.
 func (m *logManager) Run(ctx context.Context, _ ...chan<- error) error {
-	if m.Enabled() {
-		defer m.close()
-		return m.bgUpdateLog(ctx)
+	failpoint.Inject("ChangefeedNewRedoManagerError", func() {
+		failpoint.Return(errors.New("changefeed new redo manager injected error"))
+	})
+	if !m.Enabled() {
+		return nil
 	}
-	return nil
+
+	defer m.close()
+	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)
 }
 
 // WaitForReady implements pkg/util.Runnable.
@@ -546,11 +590,13 @@ func (m *logManager) close() {
 	atomic.StoreInt32(&m.closed, 1)
 
 	m.logBuffer.CloseAndDrain()
-	if err := m.writer.Close(); err != nil {
-		log.Error("redo manager fails to close writer",
-			zap.String("namespace", m.cfg.ChangeFeedID.Namespace),
-			zap.String("changefeed", m.cfg.ChangeFeedID.ID),
-			zap.Error(err))
+	if m.writer != nil {
+		if err := m.writer.Close(); err != nil && errors.Cause(err) != context.Canceled {
+			log.Error("redo manager fails to close writer",
+				zap.String("namespace", m.cfg.ChangeFeedID.Namespace),
+				zap.String("changefeed", m.cfg.ChangeFeedID.ID),
+				zap.Error(err))
+		}
 	}
 	log.Info("redo manager closed",
 		zap.String("namespace", m.cfg.ChangeFeedID.Namespace),