Repository: pingcap/tiflow
Pull request #10019 (2023-11-07 01:00:20)

dm: reduce gSet.String() usage by using zeroGSet for checking empty (#9944)
This is an automated cherry-pick of #9944

<!--
Thank you for contributing to TiFlow! 
Please read MD's [CONTRIBUTING](https://github.com/pingcap/tiflow/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.
-->

### What problem does this PR solve?
<!--
Please create an issue first to describe the problem.

There MUST be one line starting with "Issue Number:  " and 
linking the relevant issues via the "close" or "ref".

For more info, check https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue.
 -->

Issue Number: close https://github.com/pingcap/tiflow/issues/9676

### What is changed and how it works?
reduce gSet.String() usage by using zeroGSet for checking empty

### Check List <!--REMOVE the items that are not applicable-->

#### Tests <!-- At least one of them must be included. -->

 - Unit test
 - Integration test

#### Questions <!-- Authors should answer these questions and reviewers should consider these questions. -->

##### Will it cause performance regression or break compatibility?

##### Do you need to update user documentation, design documentation or monitoring documentation?

### Release note <!-- bugfixes or new features need a release note -->

```release-note
None
```


ti-chi-bot[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
**Once this PR has been reviewed and has the lgtm label**, please ask for approval from [lichunzhu](https://github.com/lichunzhu). For more information see [the Kubernetes Code Review Process](https://book.prow.tidb.net/#/workflows/pr).

The full list of commands accepted by this bot can be found [here](https://prow.tidb.net/command-help?repo=pingcap%2Ftiflow).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/pingcap/tiflow/blob/release-6.5/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["lichunzhu"]} -->
ti-chi-bot[bot]: This cherry pick PR is for a release branch and has not yet been approved by triage owners. 
Adding the `do-not-merge/cherry-pick-not-approved` label.

To merge this cherry pick:
1. It must be approved by the approvers firstly.
2. **AFTER** it has been approved by approvers, please wait for the cherry-pick merging approval from triage owners.


<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>

Patch:
diff --git a/Makefile b/Makefile
index e94c2a140..d44176723 100644
--- a/Makefile
+++ b/Makefile
@@ -322,11 +322,11 @@ check-static: tools/bin/golangci-lint
 
 check: check-copyright fmt check-static tidy terror_check errdoc \
 	check-merge-conflicts check-ticdc-dashboard check-diff-line-width \
-	swagger-spec check-makefiles check_engine_integration_test
+	check-makefiles check_engine_integration_test
 	@git --no-pager diff --exit-code || (echo "Please add changed files!" && false)
 
 fast_check: check-copyright fmt check-static tidy terror_check errdoc \
-	check-merge-conflicts check-ticdc-dashboard check-diff-line-width swagger-spec check-makefiles \
+	check-merge-conflicts check-ticdc-dashboard check-diff-line-width check-makefiles \
 	check_cdc_integration_test check_dm_integration_test check_engine_integration_test 
 	@git --no-pager diff --exit-code || (echo "Please add changed files!" && false)
 
@@ -346,10 +346,7 @@ unit_test_coverage:
 
 data-flow-diagram: docs/data-flow.dot
 	dot -Tsvg docs/data-flow.dot > docs/data-flow.svg
-
-swagger-spec: tools/bin/swag
-	tools/bin/swag init --exclude dm,engine --parseVendor -generalInfo cdc/api/v1/api.go --output docs/swagger
-
+	
 generate_mock: ## Generate mock code.
 generate_mock: tools/bin/mockgen
 	scripts/generate-mock.sh
diff --git a/cdc/api/v2/changefeed.go b/cdc/api/v2/changefeed.go
index b163b0ac0..be4fa86b8 100644
--- a/cdc/api/v2/changefeed.go
+++ b/cdc/api/v2/changefeed.go
@@ -15,6 +15,7 @@ package v2
 
 import (
 	"context"
+	"fmt"
 	"net/http"
 	"sort"
 	"strings"
@@ -23,16 +24,17 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
+	tidbkv "github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tiflow/cdc/api"
 	"github.com/pingcap/tiflow/cdc/capture"
 	"github.com/pingcap/tiflow/cdc/model"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/retry"
-	"github.com/pingcap/tiflow/pkg/security"
 	"github.com/pingcap/tiflow/pkg/txnutil/gc"
 	"github.com/pingcap/tiflow/pkg/upstream"
 	"github.com/pingcap/tiflow/pkg/util"
 	"github.com/tikv/client-go/v2/oracle"
+	pd "github.com/tikv/pd/client"
 	"go.uber.org/zap"
 )
 
@@ -346,26 +348,31 @@ func (h *OpenAPIV2) updateChangefeed(c *gin.Context) {
 		zap.String("changefeedInfo", oldCfInfo.String()),
 		zap.Any("upstreamInfo", OldUpInfo))
 
-	var pdAddrs []string
-	var credentials *security.Credential
-	if OldUpInfo != nil {
-		pdAddrs = strings.Split(OldUpInfo.PDEndpoints, ",")
-		credentials = &security.Credential{
-			CAPath:        OldUpInfo.CAPath,
-			CertPath:      OldUpInfo.CertPath,
-			KeyPath:       OldUpInfo.KeyPath,
-			CertAllowedCN: OldUpInfo.CertAllowedCN,
-		}
-	}
-	if len(updateCfConfig.PDAddrs) != 0 {
-		pdAddrs = updateCfConfig.PDAddrs
-		credentials = updateCfConfig.PDConfig.toCredential()
+	upManager, err := h.capture.GetUpstreamManager()
+	if err != nil {
+		_ = c.Error(err)
+		return
 	}
 
-	storage, err := h.helpers.createTiStore(pdAddrs, credentials)
-	if err != nil {
-		_ = c.Error(errors.Trace(err))
+	var storage tidbkv.Storage
+	// if PDAddrs is not empty, use it to create a new kvstore
+	// Note: upManager is nil in some unit test cases
+	if len(updateCfConfig.PDAddrs) != 0 || upManager == nil {
+		pdAddrs := updateCfConfig.PDAddrs
+		credentials := updateCfConfig.PDConfig.toCredential()
+		storage, err = h.helpers.createTiStore(pdAddrs, credentials)
+		if err != nil {
+			_ = c.Error(errors.Trace(err))
+		}
+	} else { // get the upstream of the changefeed to get the kvstore
+		up, ok := upManager.Get(oldCfInfo.UpstreamID)
+		if !ok {
+			_ = c.Error(errors.New(fmt.Sprintf("upstream %d not found", oldCfInfo.UpstreamID)))
+			return
+		}
+		storage = up.KVStorage
 	}
+
 	newCfInfo, newUpInfo, err := h.helpers.verifyUpdateChangefeedConfig(ctx,
 		updateCfConfig, oldCfInfo, OldUpInfo, storage, cfStatus.CheckpointTs)
 	if err != nil {
@@ -588,24 +595,26 @@ func (h *OpenAPIV2) resumeChangefeed(c *gin.Context) {
 		return
 	}
 
+	var pdClient pd.Client
+	// if PDAddrs is empty, use the default pdClient
 	if len(cfg.PDAddrs) == 0 {
 		up, err := getCaptureDefaultUpstream(h.capture)
 		if err != nil {
 			_ = c.Error(err)
 			return
 		}
-		cfg.PDConfig = getUpstreamPDConfig(up)
-	}
-	credential := cfg.PDConfig.toCredential()
-
-	timeoutCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
-	defer cancel()
-	pdClient, err := h.helpers.getPDClient(timeoutCtx, cfg.PDAddrs, credential)
-	if err != nil {
-		_ = c.Error(cerror.WrapError(cerror.ErrAPIInvalidParam, err))
-		return
+		pdClient = up.PDClient
+	} else {
+		credential := cfg.PDConfig.toCredential()
+		timeoutCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
+		defer cancel()
+		pdClient, err = h.helpers.getPDClient(timeoutCtx, cfg.PDAddrs, credential)
+		if err != nil {
+			_ = c.Error(cerror.WrapError(cerror.ErrAPIInvalidParam, err))
+			return
+		}
+		defer pdClient.Close()
 	}
-	defer pdClient.Close()
 
 	if err := h.helpers.verifyResumeChangefeedConfig(
 		ctx,
diff --git a/cdc/api/v2/model.go b/cdc/api/v2/model.go
index 188770983..a6dcc7b2f 100644
--- a/cdc/api/v2/model.go
+++ b/cdc/api/v2/model.go
@@ -189,6 +189,8 @@ type ReplicaConfig struct {
 	Mounter    *MounterConfig    `json:"mounter"`
 	Sink       *SinkConfig       `json:"sink"`
 	Consistent *ConsistentConfig `json:"consistent"`
+
+	ChangefeedErrorStuckDuration *JSONDuration `json:"changefeed_error_stuck_duration,omitempty" swaggertype:"string"`
 }
 
 // ToInternalReplicaConfig coverts *v2.ReplicaConfig into *config.ReplicaConfig
@@ -212,6 +214,10 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 	if c.SyncPointRetention != nil {
 		res.SyncPointRetention = c.SyncPointRetention.duration
 	}
+	if c.ChangefeedErrorStuckDuration != nil {
+		res.ChangefeedErrorStuckDuration = c.ChangefeedErrorStuckDuration.duration
+	}
+
 	res.BDRMode = c.BDRMode
 
 	if c.Filter != nil {
@@ -339,16 +345,17 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 	cloned := c.Clone()
 	res := &ReplicaConfig{
-		MemoryQuota:           cloned.MemoryQuota,
-		CaseSensitive:         cloned.CaseSensitive,
-		EnableOldValue:        cloned.EnableOldValue,
-		ForceReplicate:        cloned.ForceReplicate,
-		IgnoreIneligibleTable: false,
-		CheckGCSafePoint:      cloned.CheckGCSafePoint,
-		EnableSyncPoint:       cloned.EnableSyncPoint,
-		SyncPointInterval:     &JSONDuration{cloned.SyncPointInterval},
-		SyncPointRetention:    &JSONDuration{cloned.SyncPointRetention},
-		BDRMode:               cloned.BDRMode,
+		MemoryQuota:                  cloned.MemoryQuota,
+		CaseSensitive:                cloned.CaseSensitive,
+		EnableOldValue:               cloned.EnableOldValue,
+		ForceReplicate:               cloned.ForceReplicate,
+		IgnoreIneligibleTable:        false,
+		CheckGCSafePoint:             cloned.CheckGCSafePoint,
+		EnableSyncPoint:              cloned.EnableSyncPoint,
+		SyncPointInterval:            &JSONDuration{cloned.SyncPointInterval},
+		SyncPointRetention:           &JSONDuration{cloned.SyncPointRetention},
+		BDRMode:                      cloned.BDRMode,
+		ChangefeedErrorStuckDuration: &JSONDuration{cloned.ChangefeedErrorStuckDuration},
 	}
 
 	if cloned.Filter != nil {
diff --git a/cdc/model/changefeed.go b/cdc/model/changefeed.go
index 3493e3c1e..48f104273 100644
--- a/cdc/model/changefeed.go
+++ b/cdc/model/changefeed.go
@@ -323,6 +323,9 @@ func (info *ChangeFeedInfo) VerifyAndComplete() {
 	if info.Config.Consistent == nil {
 		info.Config.Consistent = defaultConfig.Consistent
 	}
+	if info.Config.ChangefeedErrorStuckDuration == 0 {
+		info.Config.ChangefeedErrorStuckDuration = defaultConfig.ChangefeedErrorStuckDuration
+	}
 }
 
 // FixIncompatible fixes incompatible changefeed meta info.
@@ -359,6 +362,11 @@ func (info *ChangeFeedInfo) FixIncompatible() {
 		log.Info("Fix incompatible enable old value completed", zap.String("changefeed", info.String()),
 			zap.Bool("enableOldValue", info.Config.EnableOldValue))
 	}
+	if info.Config.ChangefeedErrorStuckDuration == 0 {
+		log.Info("Start fixing incompatible changefeed error stuck duration", zap.String("changefeed", info.String()))
+		info.Config.ChangefeedErrorStuckDuration = config.GetDefaultReplicaConfig().ChangefeedErrorStuckDuration
+		log.Info("Fix incompatible changefeed error stuck duration completed", zap.String("changefeed", info.String()))
+	}
 }
 
 // fixState attempts to fix state loss from upgrading the old owner to the new owner.
diff --git a/cdc/model/sink.go b/cdc/model/sink.go
index 33449a67d..adf063b1b 100644
--- a/cdc/model/sink.go
+++ b/cdc/model/sink.go
@@ -608,20 +608,26 @@ func BuildTiDBTableInfo(columns []*Column, indexColumns [][]int) *model.TableInf
 			continue
 		}
 		if firstCol.Flag.IsPrimaryKey() {
-			indexInfo.Primary = true
 			indexInfo.Unique = true
 		}
 		if firstCol.Flag.IsUniqueKey() {
 			indexInfo.Unique = true
 		}
 
+		isPrimary := true
 		for _, offset := range colOffsets {
-			col := ret.Columns[offset]
+			col := columns[offset]
+			// When only all columns in the index are primary key, then the index is primary key.
+			if col == nil || !col.Flag.IsPrimaryKey() {
+				isPrimary = false
+			}
 
+			tiCol := ret.Columns[offset]
 			indexCol := &model.IndexColumn{}
-			indexCol.Name = col.Name
+			indexCol.Name = tiCol.Name
 			indexCol.Offset = offset
 			indexInfo.Columns = append(indexInfo.Columns, indexCol)
+			indexInfo.Primary = isPrimary
 		}
 
 		// TODO: revert the "all column set index related flag" to "only the
diff --git a/cdc/owner/changefeed.go b/cdc/owner/changefeed.go
index 80a32391a..48c4d95d1 100644
--- a/cdc/owner/changefeed.go
+++ b/cdc/owner/changefeed.go
@@ -145,7 +145,7 @@ func newChangefeed(
 		// The scheduler will be created lazily.
 		scheduler:        nil,
 		barriers:         newBarriers(),
-		feedStateManager: newFeedStateManager(up),
+		feedStateManager: newFeedStateManager(up, state.Info.Config),
 		upstream:         up,
 
 		errCh:     make(chan error, defaultErrChSize),
@@ -317,6 +317,12 @@ func (c *changefeed) tick(ctx cdcContext.Context, captures map[model.CaptureID]*
 	default:
 	}
 
+	if c.redoMetaMgr.Enabled() {
+		if !c.redoMetaMgr.Running() {
+			return nil
+		}
+	}
+
 	// TODO: pass table checkpointTs when we support concurrent process ddl
 	allPhysicalTables, barrier, err := c.ddlManager.tick(ctx, preCheckpointTs, nil)
 	if err != nil {
@@ -551,32 +557,21 @@ LOOP2:
 		ctx.Throw(c.ddlPuller.Run(cancelCtx))
 	}()
 
-	stdCtx := contextutil.PutChangefeedIDInCtx(cancelCtx, c.id)
-	c.redoDDLMgr, err = redo.NewDDLManager(stdCtx, c.state.Info.Config.Consistent, ddlStartTs)
-	failpoint.Inject("ChangefeedNewRedoManagerError", func() {
-		err = errors.New("changefeed new redo manager injected error")
-	})
-	if err != nil {
-		return err
-	}
+	c.redoDDLMgr = redo.NewDDLManager(c.id, c.state.Info.Config.Consistent, ddlStartTs)
 	if c.redoDDLMgr.Enabled() {
 		c.wg.Add(1)
 		go func() {
 			defer c.wg.Done()
-			ctx.Throw(c.redoDDLMgr.Run(stdCtx))
+			ctx.Throw(c.redoDDLMgr.Run(cancelCtx))
 		}()
 	}
 
-	c.redoMetaMgr, err = redo.NewMetaManagerWithInit(stdCtx,
-		c.state.Info.Config.Consistent, checkpointTs)
-	if err != nil {
-		return err
-	}
+	c.redoMetaMgr = redo.NewMetaManager(c.id, c.state.Info.Config.Consistent, checkpointTs)
 	if c.redoMetaMgr.Enabled() {
 		c.wg.Add(1)
 		go func() {
 			defer c.wg.Done()
-			ctx.Throw(c.redoMetaMgr.Run(stdCtx))
+			ctx.Throw(c.redoMetaMgr.Run(cancelCtx))
 		}()
 	}
 	log.Info("owner creates redo manager",
@@ -733,15 +728,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 {
-			redoMetaMgr, err := redo.NewMetaManager(ctx, c.state.Info.Config.Consistent)
-			if err != nil {
-				log.Info("owner creates redo manager for clean fail",
-					zap.String("namespace", c.id.Namespace),
-					zap.String("changefeed", c.id.ID),
-					zap.Error(err))
-				return
-			}
-			c.redoMetaMgr = redoMetaMgr
+			c.redoMetaMgr = redo.NewMetaManager(c.id, c.state.Info.Config.Consistent, 0)
 		}
 		err := c.redoMetaMgr.Cleanup(ctx)
 		if err != nil {
diff --git a/cdc/owner/ddl_manager.go b/cdc/owner/ddl_manager.go
index f825465cc..629300bcc 100644
--- a/cdc/owner/ddl_manager.go
+++ b/cdc/owner/ddl_manager.go
@@ -27,6 +27,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/puller"
 	"github.com/pingcap/tiflow/cdc/redo"
 	"github.com/pingcap/tiflow/cdc/scheduler/schedulepb"
+	"github.com/pingcap/tiflow/pkg/filter"
 	"go.uber.org/zap"
 )
 
@@ -219,21 +220,22 @@ func (m *ddlManager) tick(
 			}
 
 			for _, event := range events {
-				// If changefeed is in BDRMode, skip ddl.
-				if m.BDRMode {
-					log.Info("changefeed is in BDRMode, skip a ddl event",
-						zap.String("namespace", m.changfeedID.Namespace),
-						zap.String("ID", m.changfeedID.ID),
-						zap.Any("ddlEvent", event))
-					continue
+				// TODO: find a better place to do this check
+				// check if the ddl event is belong to an ineligible table.
+				// If so, we should ignore it.
+				if !filter.IsSchemaDDL(event.Type) {
+					ignore, err := m.schema.
+						IsIneligibleTable(ctx, event.TableInfo.TableName.TableID, event.CommitTs)
+					if err != nil {
+						return nil, nil, errors.Trace(err)
+					}
+					if ignore {
+						log.Warn("ignore the DDL event of ineligible table",
+							zap.String("changefeed", m.changfeedID.ID), zap.Any("ddl", event))
+						continue
+					}
 				}
 
-				if event.TableInfo != nil &&
-					m.schema.IsIneligibleTableID(event.TableInfo.TableName.TableID) {
-					log.Warn("ignore the DDL event of ineligible table",
-						zap.String("changefeed", m.changfeedID.ID), zap.Any("ddl", event))
-					continue
-				}
 				tableName := event.TableInfo.TableName
 				// Add all valid DDL events to the pendingDDLs.
 				m.pendingDDLs[tableName] = append(m.pendingDDLs[tableName], event)
@@ -337,6 +339,24 @@ func (m *ddlManager) executeDDL(ctx context.Context) error {
 	if m.executingDDL == nil {
 		return nil
 	}
+
+	// If changefeed is in BDRMode, skip ddl.
+	if m.BDRMode {
+		log.Info("changefeed is in BDRMode, skip a ddl event",
+			zap.String("namespace", m.changfeedID.Namespace),
+			zap.String("ID", m.changfeedID.ID),
+			zap.Any("ddlEvent", m.executingDDL))
+		tableName := m.executingDDL.TableInfo.TableName
+		// Set it to nil first to accelerate GC.
+		m.pendingDDLs[tableName][0] = nil
+		m.pendingDDLs[tableName] = m.pendingDDLs[tableName][1:]
+		m.schema.DoGC(m.executingDDL.CommitTs - 1)
+		m.justSentDDL = m.executingDDL
+		m.executingDDL = nil
+		m.cleanCache()
+		return nil
+	}
+
 	failpoint.Inject("ExecuteNotDone", func() {
 		// This ddl will never finish executing.
 		// It is used to test the logic that a ddl only block the related table
diff --git a/cdc/owner/feed_state_manager.go b/cdc/owner/feed_state_manager.go
index 69676a10e..91741a05f 100644
--- a/cdc/owner/feed_state_manager.go
+++ b/cdc/owner/feed_state_manager.go
@@ -21,6 +21,7 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/pingcap/tiflow/pkg/config"
 	cerrors "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/orchestrator"
 	"github.com/pingcap/tiflow/pkg/upstream"
@@ -36,7 +37,6 @@ const (
 	// To avoid thunderherd, a random factor is also added.
 	defaultBackoffInitInterval        = 10 * time.Second
 	defaultBackoffMaxInterval         = 10 * time.Minute
-	defaultBackoffMaxElapsedTime      = 30 * time.Minute
 	defaultBackoffRandomizationFactor = 0.1
 	defaultBackoffMultiplier          = 2.0
 )
@@ -71,23 +71,21 @@ type feedStateManager struct {
 }
 
 // newFeedStateManager creates feedStateManager and initialize the exponential backoff
-func newFeedStateManager(up *upstream.Upstream) *feedStateManager {
-	f := new(feedStateManager)
-	f.upstream = up
-
-	f.errBackoff = backoff.NewExponentialBackOff()
-	f.errBackoff.InitialInterval = defaultBackoffInitInterval
-	f.errBackoff.MaxInterval = defaultBackoffMaxInterval
-	f.errBackoff.Multiplier = defaultBackoffMultiplier
-	f.errBackoff.RandomizationFactor = defaultBackoffRandomizationFactor
+func newFeedStateManager(up *upstream.Upstream, cfg *config.ReplicaConfig) *feedStateManager {
+	m := new(feedStateManager)
+	m.upstream = up
+
+	m.errBackoff = backoff.NewExponentialBackOff()
+	m.errBackoff.InitialInterval = defaultBackoffInitInterval
+	m.errBackoff.MaxInterval = defaultBackoffMaxInterval
+	m.errBackoff.Multiplier = defaultBackoffMultiplier
+	m.errBackoff.RandomizationFactor = defaultBackoffRandomizationFactor
 	// backoff will stop once the defaultBackoffMaxElapsedTime has elapsed.
-	f.errBackoff.MaxElapsedTime = defaultBackoffMaxElapsedTime
+	m.errBackoff.MaxElapsedTime = cfg.ChangefeedErrorStuckDuration
+	m.changefeedErrorStuckDuration = cfg.ChangefeedErrorStuckDuration
 
-	f.changefeedErrorStuckDuration = defaultBackoffMaxElapsedTime
-
-	f.resetErrRetry()
-	f.isRetrying = false
-	return f
+	m.resetErrRetry()
+	return m
 }
 
 func (m *feedStateManager) shouldRetry() bool {
diff --git a/cdc/owner/schema.go b/cdc/owner/schema.go
index 38b3d53ef..b2f0ea39a 100644
--- a/cdc/owner/schema.go
+++ b/cdc/owner/schema.go
@@ -126,8 +126,18 @@ func (s *schemaWrap4Owner) AllTables(
 	return tables, nil
 }
 
-func (s *schemaWrap4Owner) IsIneligibleTableID(tableID model.TableID) bool {
-	return s.GetLastSnapshot().IsIneligibleTableID(tableID)
+// IsIneligibleTable returns whether the table is ineligible.
+// It uses the snapshot of the given ts to check the table.
+func (s *schemaWrap4Owner) IsIneligibleTable(
+	ctx context.Context,
+	tableID model.TableID,
+	ts model.Ts,
+) (bool, error) {
+	snap, err := s.GetSnapshot(ctx, ts)
+	if err != nil {
+		return false, err
+	}
+	return snap.IsIneligibleTableID(tableID), nil
 }
 
 // TODO: find a better way to refactor this function.
diff --git a/cdc/processor/processor.go b/cdc/processor/processor.go
index 56f377ad8..acfe4aa51 100644
--- a/cdc/processor/processor.go
+++ b/cdc/processor/processor.go
@@ -805,10 +805,7 @@ func (p *processor) lazyInitImpl(ctx cdcContext.Context) error {
 	conf := config.GetGlobalServerConfig()
 	p.pullBasedSinking = conf.Debug.EnablePullBasedSink
 
-	p.redoDMLMgr, err = redo.NewDMLManager(stdCtx, p.changefeed.Info.Config.Consistent)
-	if err != nil {
-		return err
-	}
+	p.redoDMLMgr = redo.NewDMLManager(p.changefeedID, p.changefeed.Info.Config.Consistent)
 	if p.redoDMLMgr.Enabled() {
 		p.wg.Add(1)
 		go func() {
@@ -1081,7 +1078,7 @@ func (p *processor) getTableName(ctx context.Context, tableID model.TableID) str
 		retry.WithIsRetryableErr(cerror.IsRetryableError))
 
 	if tableName == nil {
-		log.Warn("failed to get table name for metric")
+		log.Warn("failed to get table name for metric", zap.Any("tableID", tableID))
 		return strconv.Itoa(int(tableID))
 	}
 
diff --git a/cdc/processor/sinkmanager/redo_cache.go b/cdc/processor/sinkmanager/redo_cache.go
index a7e54eaa9..147d4ef67 100644
--- a/cdc/processor/sinkmanager/redo_cache.go
+++ b/cdc/processor/sinkmanager/redo_cache.go
@@ -57,11 +57,20 @@ type popResult struct {
 	events      []*model.RowChangedEvent
 	size        uint64 // size of events.
 	releaseSize uint64 // size of all released events.
-	pushCount   int
-	success     bool
-	// If success, boundary is the upperBound of poped events.
-	// Otherwise, boundary is the lowerBound of cached events.
-	boundary engine.Position
+
+	// many RowChangedEvent can come from one same PolymorphicEvent.
+	// pushCount indicates the count of raw PolymorphicEvents.
+	pushCount int
+
+	// success indicates whether there is a gap between cached events and required events.
+	success bool
+
+	// If success, upperBoundIfSuccess is the upperBound of poped events.
+	// The caller should fetch events (upperBoundIfSuccess, upperBound] from engine.
+	upperBoundIfSuccess engine.Position
+	// If fail, lowerBoundIfFail is the lowerBound of cached events.
+	// The caller should fetch events [lowerBound, lowerBoundIfFail) from engine.
+	lowerBoundIfFail engine.Position
 }
 
 // newRedoEventCache creates a redoEventCache instance.
@@ -142,26 +151,28 @@ func (e *eventAppender) pop(lowerBound, upperBound engine.Position) (res popResu
 		// NOTE: the caller will fetch events [lowerBound, res.boundary) from engine.
 		res.success = false
 		if e.lowerBound.Compare(upperBound.Next()) <= 0 {
-			res.boundary = e.lowerBound
+			res.lowerBoundIfFail = e.lowerBound
 		} else {
-			res.boundary = upperBound.Next()
+			res.lowerBoundIfFail = upperBound.Next()
 		}
 		return
 	}
+
 	if !e.upperBound.Valid() {
-		// if e.upperBound is invalid, it means there are no resolved transactions
-		// in the cache.
+		// It means there are no resolved cached transactions in the required range.
 		// NOTE: the caller will fetch events [lowerBound, res.boundary) from engine.
 		res.success = false
-		res.boundary = upperBound.Next()
+		res.lowerBoundIfFail = upperBound.Next()
 		return
 	}
 
 	res.success = true
-	if upperBound.Compare(e.upperBound) > 0 {
-		res.boundary = e.upperBound
+	if lowerBound.Compare(e.upperBound) > 0 {
+		res.upperBoundIfSuccess = lowerBound.Prev()
+	} else if upperBound.Compare(e.upperBound) > 0 {
+		res.upperBoundIfSuccess = e.upperBound
 	} else {
-		res.boundary = upperBound
+		res.upperBoundIfSuccess = upperBound
 	}
 
 	startIdx := sort.Search(e.readyCount, func(i int) bool {
@@ -173,28 +184,23 @@ func (e *eventAppender) pop(lowerBound, upperBound engine.Position) (res popResu
 		res.releaseSize += e.sizes[i]
 	}
 
-	var endIdx int
-	if startIdx == e.readyCount {
-		endIdx = startIdx
-	} else {
-		endIdx = sort.Search(e.readyCount, func(i int) bool {
-			pos := engine.Position{CommitTs: e.events[i].CommitTs, StartTs: e.events[i].StartTs}
-			return pos.Compare(res.boundary) > 0
-		})
-		res.events = e.events[startIdx:endIdx]
-		for i := startIdx; i < endIdx; i++ {
-			res.size += e.sizes[i]
-			res.pushCount += int(e.pushCounts[i])
-		}
-		res.releaseSize += res.size
+	endIdx := sort.Search(e.readyCount, func(i int) bool {
+		pos := engine.Position{CommitTs: e.events[i].CommitTs, StartTs: e.events[i].StartTs}
+		return pos.Compare(res.upperBoundIfSuccess) > 0
+	})
+	res.events = e.events[startIdx:endIdx]
+	for i := startIdx; i < endIdx; i++ {
+		res.size += e.sizes[i]
+		res.pushCount += int(e.pushCounts[i])
 	}
+	res.releaseSize += res.size
 
 	e.events = e.events[endIdx:]
 	e.sizes = e.sizes[endIdx:]
 	e.pushCounts = e.pushCounts[endIdx:]
 	e.readyCount -= endIdx
 	// Update boundaries. Set upperBound to invalid if the range has been drained.
-	e.lowerBound = res.boundary.Next()
+	e.lowerBound = res.upperBoundIfSuccess.Next()
 	if e.lowerBound.Compare(e.upperBound) > 0 {
 		e.upperBound = engine.Position{}
 	}
diff --git a/cdc/processor/sinkmanager/table_sink_worker.go b/cdc/processor/sinkmanager/table_sink_worker.go
index d90a35ce3..0ed6d47c9 100644
--- a/cdc/processor/sinkmanager/table_sink_worker.go
+++ b/cdc/processor/sinkmanager/table_sink_worker.go
@@ -212,8 +212,9 @@ func (w *sinkWorker) handleTask(ctx context.Context, task *sinkTask) (finalErr e
 		if err != nil {
 			return errors.Trace(err)
 		}
+		lastPos = lowerBound.Prev()
 		if drained {
-			performCallback(lowerBound.Prev())
+			performCallback(lastPos)
 			return nil
 		}
 	}
@@ -405,7 +406,7 @@ func (w *sinkWorker) fetchFromCache(
 	}
 	popRes := cache.pop(*lowerBound, *upperBound)
 	if popRes.success {
-		newLowerBound = popRes.boundary.Next()
+		newLowerBound = popRes.upperBoundIfSuccess.Next()
 		if len(popRes.events) > 0 {
 			metrics.OutputEventCount.WithLabelValues(
 				task.tableSink.changefeed.Namespace,
@@ -420,9 +421,9 @@ func (w *sinkWorker) fetchFromCache(
 
 		// Get a resolvedTs so that we can record it into sink memory quota.
 		var resolvedTs model.ResolvedTs
-		isCommitFence := popRes.boundary.IsCommitFence()
+		isCommitFence := popRes.upperBoundIfSuccess.IsCommitFence()
 		if w.splitTxn {
-			resolvedTs = model.NewResolvedTs(popRes.boundary.CommitTs)
+			resolvedTs = model.NewResolvedTs(popRes.upperBoundIfSuccess.CommitTs)
 			if !isCommitFence {
 				resolvedTs.Mode = model.BatchResolvedMode
 				resolvedTs.BatchID = *batchID
@@ -430,9 +431,9 @@ func (w *sinkWorker) fetchFromCache(
 			}
 		} else {
 			if isCommitFence {
-				resolvedTs = model.NewResolvedTs(popRes.boundary.CommitTs)
+				resolvedTs = model.NewResolvedTs(popRes.upperBoundIfSuccess.CommitTs)
 			} else {
-				resolvedTs = model.NewResolvedTs(popRes.boundary.CommitTs - 1)
+				resolvedTs = model.NewResolvedTs(popRes.upperBoundIfSuccess.CommitTs - 1)
 			}
 		}
 		// Transfer the memory usage from redoMemQuota to sinkMemQuota.
@@ -448,7 +449,7 @@ func (w *sinkWorker) fetchFromCache(
 			zap.Any("resolvedTs", resolvedTs),
 			zap.Error(err))
 	} else {
-		newUpperBound = popRes.boundary.Prev()
+		newUpperBound = popRes.lowerBoundIfFail.Prev()
 	}
 	cacheDrained = newLowerBound.Compare(newUpperBound) > 0
 	log.Debug("fetchFromCache is performed",
diff --git a/cdc/redo/manager.go b/cdc/redo/manager.go
index 53b3f81c6..571f27b25 100644
--- a/cdc/redo/manager.go
+++ b/cdc/redo/manager.go
@@ -19,9 +19,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/pingcap/failpoint"
 	"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/cdc/redo/writer"
@@ -62,19 +61,17 @@ func NewDisabledDDLManager() *ddlManager {
 
 // NewDDLManager creates a new ddl Manager.
 func NewDDLManager(
-	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)
 	tableID := int64(0)
-	logManager.AddTable(tableID, ddlStartTs)
+	m.AddTable(tableID, ddlStartTs)
 	return &ddlManager{
-		logManager: logManager,
+		logManager: m,
 		// The current fakeTableID is meaningless, find a meaningful id in the future.
 		fakeTableID: tableID,
-	}, nil
+	}
 }
 
 type ddlManager struct {
@@ -111,12 +108,12 @@ type DMLManager interface {
 }
 
 // NewDMLManager creates a new dml Manager.
-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),
 	}
-	return &dmlManager{logManager: logManager}, nil
 }
 
 // NewDisabledDMLManager creates a disabled dml Manager.
@@ -222,28 +219,21 @@ type logManager struct {
 }
 
 func newLogManager(
-	ctx context.Context, cfg *config.ConsistentConfig, logType string,
-) (*logManager, error) {
+	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) {
-		return &logManager{enabled: false}, nil
+		return &logManager{enabled: false}
 	}
 
-	uri, err := storage.ParseRawURL(cfg.Storage)
-	if err != nil {
-		return nil, err
-	}
-	changefeedID := contextutil.ChangefeedIDFromCtx(ctx)
-	m := &logManager{
+	return &logManager{
 		enabled: true,
 		cfg: &writer.LogWriterConfig{
-			ConsistentConfig:   *cfg,
-			LogType:            logType,
-			CaptureID:          contextutil.CaptureAddrFromCtx(ctx),
-			ChangeFeedID:       changefeedID,
-			URI:                *uri,
-			UseExternalStorage: redo.IsExternalStorage(uri.Scheme),
-			MaxLogSizeInBytes:  cfg.MaxLogSize * redo.Megabyte,
+			ConsistentConfig:  *cfg,
+			LogType:           logType,
+			CaptureID:         config.GetGlobalServerConfig().AdvertiseAddr,
+			ChangeFeedID:      changefeedID,
+			MaxLogSizeInBytes: cfg.MaxLogSize * redo.Megabyte,
 		},
 		logBuffer: chann.NewDrainableChann[cacheEvents](),
 		metricWriteLogDuration: common.RedoWriteLogDurationHistogram.
@@ -255,15 +245,17 @@ 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) error {
+	failpoint.Inject("ChangefeedNewRedoManagerError", func() {
+		failpoint.Return(errors.New("changefeed new redo manager injected error"))
+	})
+	if !m.Enabled() {
+		return nil
+	}
+
 	defer m.close()
 	start := time.Now()
 	w, err := factory.NewRedoLogWriter(ctx, m.cfg)
@@ -281,8 +273,18 @@ func (m *logManager) Run(ctx context.Context) error {
 
 func (m *logManager) getFlushDuration() time.Duration {
 	flushIntervalInMs := m.cfg.FlushIntervalInMs
+	defaultFlushIntervalInMs := redo.DefaultFlushIntervalInMs
 	if m.cfg.LogType == redo.RedoDDLLogFileType {
 		flushIntervalInMs = m.cfg.MetaFlushIntervalInMs
+		defaultFlushIntervalInMs = redo.DefaultMetaFlushIntervalInMs
+	}
+	if flushIntervalInMs < redo.MinFlushIntervalInMs {
+		log.Warn("redo flush interval is too small, use default value",
+			zap.Stringer("namespace", m.cfg.ChangeFeedID),
+			zap.Int("default", defaultFlushIntervalInMs),
+			zap.String("logType", m.cfg.LogType),
+			zap.Int64("interval", flushIntervalInMs))
+		flushIntervalInMs = int64(defaultFlushIntervalInMs)
 	}
 	return time.Duration(flushIntervalInMs) * time.Millisecond
 }
@@ -549,11 +551,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),
diff --git a/cdc/redo/meta_manager.go b/cdc/redo/meta_manager.go
index 86ce58313..1a5f340c4 100644
--- a/cdc/redo/meta_manager.go
+++ b/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"
@@ -31,6 +30,7 @@ import (
 	"github.com/pingcap/tiflow/pkg/util"
 	"github.com/pingcap/tiflow/pkg/uuid"
 	"github.com/prometheus/client_golang/prometheus"
+	"go.uber.org/atomic"
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
 )
@@ -47,6 +47,9 @@ type MetaManager interface {
 	// Cleanup deletes all redo logs, which are only called from the owner
 	// when changefeed is deleted.
 	Cleanup(ctx context.Context) error
+
+	// Running return true if the meta manager is running or not.
+	Running() bool
 }
 
 type metaManager struct {
@@ -54,6 +57,9 @@ type metaManager struct {
 	changeFeedID model.ChangeFeedID
 	enabled      bool
 
+	// running means the meta manager now running normally.
+	running atomic.Bool
+
 	metaCheckpointTs statefulRts
 	metaResolvedTs   statefulRts
 
@@ -63,8 +69,11 @@ type metaManager struct {
 	uuidGenerator uuid.Generator
 	preMetaFile   string
 
-	lastFlushTime          time.Time
+	startTs model.Ts
+
 	flushIntervalInMs      int64
+	lastFlushTime          time.Time
+	cfg                    *config.ConsistentConfig
 	metricFlushLogDuration prometheus.Observer
 }
 
@@ -75,51 +84,22 @@ func NewDisabledMetaManager() *metaManager {
 	}
 }
 
-// NewMetaManagerWithInit creates a new Manager and initializes the meta.
-func NewMetaManagerWithInit(
-	ctx context.Context, cfg *config.ConsistentConfig, startTs model.Ts,
-) (*metaManager, error) {
-	m, err := NewMetaManager(ctx, cfg)
-	if err != nil {
-		return nil, err
-	}
-
-	// There is no need to perform initialize operation if metaMgr is disabled
-	// or the scheme is blackhole.
-	if m.extStorage != nil {
-		m.metricFlushLogDuration = common.RedoFlushLogDurationHistogram.
-			WithLabelValues(m.changeFeedID.Namespace, m.changeFeedID.ID)
-		if err = m.preCleanupExtStorage(ctx); err != nil {
-			log.Warn("pre clean redo logs fail",
-				zap.String("namespace", m.changeFeedID.Namespace),
-				zap.String("changefeed", m.changeFeedID.ID),
-				zap.Error(err))
-			return nil, err
-		}
-		if err = m.initMeta(ctx, startTs); err != nil {
-			log.Warn("init redo meta fail",
-				zap.String("namespace", m.changeFeedID.Namespace),
-				zap.String("changefeed", m.changeFeedID.ID),
-				zap.Error(err))
-			return nil, err
-		}
-	}
-
-	return m, nil
-}
-
 // NewMetaManager creates a new meta Manager.
-func NewMetaManager(ctx context.Context, cfg *config.ConsistentConfig) (*metaManager, error) {
+func NewMetaManager(
+	changefeedID model.ChangeFeedID, cfg *config.ConsistentConfig, checkpoint model.Ts,
+) *metaManager {
 	// return a disabled Manager if no consistent config or normal consistent level
 	if cfg == nil || !redo.IsConsistentEnabled(cfg.Level) {
-		return &metaManager{enabled: false}, nil
+		return &metaManager{enabled: false}
 	}
 
 	m := &metaManager{
-		captureID:         contextutil.CaptureAddrFromCtx(ctx),
-		changeFeedID:      contextutil.ChangefeedIDFromCtx(ctx),
+		changeFeedID:      changefeedID,
+		captureID:         config.GetGlobalServerConfig().AdvertiseAddr,
 		uuidGenerator:     uuid.NewGenerator(),
 		enabled:           true,
+		cfg:               cfg,
+		startTs:           checkpoint,
 		flushIntervalInMs: cfg.MetaFlushIntervalInMs,
 	}
 
@@ -128,44 +108,70 @@ func NewMetaManager(ctx context.Context, cfg *config.ConsistentConfig) (*metaMan
 			zap.Int64("interval", m.flushIntervalInMs))
 		m.flushIntervalInMs = redo.DefaultMetaFlushIntervalInMs
 	}
+	return m
+}
 
-	uri, err := storage.ParseRawURL(cfg.Storage)
+// Enabled returns whether this log manager is enabled
+func (m *metaManager) Enabled() bool {
+	return m.enabled
+}
+
+// Running return whether the meta manager is initialized,
+// which means the external storage is accessible to the meta manager.
+func (m *metaManager) Running() bool {
+	return m.running.Load()
+}
+
+func (m *metaManager) preStart(ctx context.Context) error {
+	uri, err := storage.ParseRawURL(m.cfg.Storage)
 	if err != nil {
-		return nil, err
-	}
-	if redo.IsBlackholeStorage(uri.Scheme) {
-		return m, nil
+		return err
 	}
-
 	// "nfs" and "local" scheme are converted to "file" scheme
 	redo.FixLocalScheme(uri)
+
 	extStorage, err := redo.InitExternalStorage(ctx, *uri)
 	if err != nil {
-		return nil, err
+		return err
 	}
 	m.extStorage = extStorage
-	return m, nil
-}
 
-// Enabled returns whether this log manager is enabled
-func (m *metaManager) Enabled() bool {
-	return m.enabled
+	m.metricFlushLogDuration = common.RedoFlushLogDurationHistogram.
+		WithLabelValues(m.changeFeedID.Namespace, m.changeFeedID.ID)
+
+	err = m.preCleanupExtStorage(ctx)
+	if err != nil {
+		log.Warn("redo: pre clean redo logs fail",
+			zap.String("namespace", m.changeFeedID.Namespace),
+			zap.String("changefeed", m.changeFeedID.ID),
+			zap.Error(err))
+		return err
+	}
+	err = m.initMeta(ctx)
+	if err != nil {
+		log.Warn("redo: init redo meta fail",
+			zap.String("namespace", m.changeFeedID.Namespace),
+			zap.String("changefeed", m.changeFeedID.ID),
+			zap.Error(err))
+		return err
+	}
+	return nil
 }
 
 // Run runs bgFlushMeta and bgGC.
 func (m *metaManager) Run(ctx context.Context) error {
-	if m.extStorage == nil {
-		log.Warn("extStorage of redo meta manager is nil, skip running")
-		return nil
+	if err := m.preStart(ctx); err != nil {
+		return err
 	}
-
 	eg, egCtx := errgroup.WithContext(ctx)
 	eg.Go(func() error {
-		return m.bgFlushMeta(egCtx, m.flushIntervalInMs)
+		return m.bgFlushMeta(egCtx)
 	})
 	eg.Go(func() error {
 		return m.bgGC(egCtx)
 	})
+
+	m.running.Store(true)
 	return eg.Wait()
 }
 
@@ -194,9 +200,9 @@ func (m *metaManager) GetFlushedMeta() common.LogMeta {
 	return common.LogMeta{CheckpointTs: checkpointTs, ResolvedTs: resolvedTs}
 }
 
-// initMeta will read the meta file from external storage and initialize the meta
-// field of metaManager.
-func (m *metaManager) initMeta(ctx context.Context, startTs model.Ts) error {
+// initMeta will read the meta file from external storage and
+// use it to initialize the meta field of the metaManager.
+func (m *metaManager) initMeta(ctx context.Context) error {
 	select {
 	case <-ctx.Done():
 		return errors.Trace(ctx.Err())
@@ -204,10 +210,14 @@ func (m *metaManager) initMeta(ctx context.Context, startTs model.Ts) error {
 	}
 
 	metas := []*common.LogMeta{
-		{CheckpointTs: startTs, ResolvedTs: startTs},
+		{CheckpointTs: m.startTs, ResolvedTs: m.startTs},
 	}
 	var toRemoveMetaFiles []string
 	err := m.extStorage.WalkDir(ctx, nil, func(path string, size int64) error {
+		log.Info("redo: meta manager walk dir",
+			zap.String("namespace", m.changeFeedID.Namespace),
+			zap.String("changefeed", m.changeFeedID.ID),
+			zap.String("path", path), zap.Int64("size", size))
 		// TODO: use prefix to accelerate traverse operation
 		if !strings.HasSuffix(path, redo.MetaEXT) {
 			return nil
@@ -215,22 +225,30 @@ func (m *metaManager) initMeta(ctx context.Context, startTs model.Ts) error {
 		toRemoveMetaFiles = append(toRemoveMetaFiles, path)
 
 		data, err := m.extStorage.ReadFile(ctx, path)
-		if err != nil && !util.IsNotExistInExtStorage(err) {
-			return err
-		}
-		if len(data) != 0 {
-			var meta common.LogMeta
-			_, err = meta.UnmarshalMsg(data)
-			if err != nil {
+		if err != nil {
+			log.Warn("redo: read meta file failed",
+				zap.String("namespace", m.changeFeedID.Namespace),
+				zap.String("changefeed", m.changeFeedID.ID),
+				zap.String("path", path), zap.Error(err))
+			if !util.IsNotExistInExtStorage(err) {
 				return err
 			}
-			metas = append(metas, &meta)
+			return nil
+		}
+		var meta common.LogMeta
+		_, err = meta.UnmarshalMsg(data)
+		if err != nil {
+			log.Error("redo: unmarshal meta data failed",
+				zap.String("namespace", m.changeFeedID.Namespace),
+				zap.String("changefeed", m.changeFeedID.ID),
+				zap.Error(err), zap.ByteString("data", data))
+			return err
 		}
+		metas = append(metas, &meta)
 		return nil
 	})
 	if err != nil {
-		return errors.WrapError(errors.ErrRedoMetaInitialize,
-			errors.Annotate(err, "read meta file fail"))
+		return errors.WrapError(errors.ErrRedoMetaInitialize, err)
 	}
 
 	var checkpointTs, resolvedTs uint64
@@ -243,9 +261,15 @@ func (m *metaManager) initMeta(ctx context.Context, startTs model.Ts) error {
 	m.metaResolvedTs.unflushed = resolvedTs
 	m.metaCheckpointTs.unflushed = checkpointTs
 	if err := m.maybeFlushMeta(ctx); err != nil {
-		return errors.WrapError(errors.ErrRedoMetaInitialize,
-			errors.Annotate(err, "flush meta file fail"))
+		return errors.WrapError(errors.ErrRedoMetaInitialize, err)
 	}
+
+	flushedMeta := m.GetFlushedMeta()
+	log.Info("redo: meta manager flush init meta success",
+		zap.String("namespace", m.changeFeedID.Namespace),
+		zap.String("changefeed", m.changeFeedID.ID),
+		zap.Uint64("checkpointTs", flushedMeta.CheckpointTs),
+		zap.Uint64("resolvedTs", flushedMeta.ResolvedTs))
 	return util.DeleteFilesInExtStorage(ctx, m.extStorage, toRemoveMetaFiles)
 }
 
@@ -305,8 +329,19 @@ func (m *metaManager) shouldRemoved(path string, checkPointTs uint64) bool {
 
 // deleteAllLogs delete all redo logs and leave a deleted mark.
 func (m *metaManager) deleteAllLogs(ctx context.Context) error {
+	// when one changefeed with redo enabled gets deleted, it's extStorage should always be set to not nil
+	// otherwise it should have already meet panic during changefeed running time.
+	// the extStorage may be nil in the unit test, so just set the external storage to make unit test happy.
 	if m.extStorage == nil {
-		return nil
+		uri, err := storage.ParseRawURL(m.cfg.Storage)
+		redo.FixLocalScheme(uri)
+		if err != nil {
+			return err
+		}
+		m.extStorage, err = redo.InitExternalStorage(ctx, *uri)
+		if err != nil {
+			return err
+		}
 	}
 	// Write deleted mark before clean any files.
 	deleteMarker := getDeletedChangefeedMarker(m.changeFeedID)
@@ -382,6 +417,10 @@ func (m *metaManager) flush(ctx context.Context, meta common.LogMeta) error {
 	}
 	metaFile := getMetafileName(m.captureID, m.changeFeedID, m.uuidGenerator)
 	if err := m.extStorage.WriteFile(ctx, metaFile, data); err != nil {
+		log.Error("redo: meta manager flush meta write file failed",
+			zap.String("namespace", m.changeFeedID.Namespace),
+			zap.String("changefeed", m.changeFeedID.ID),
+			zap.Error(err))
 		return errors.WrapError(errors.ErrExternalStorageAPI, err)
 	}
 
@@ -392,6 +431,10 @@ func (m *metaManager) flush(ctx context.Context, meta common.LogMeta) error {
 		}
 		err := m.extStorage.DeleteFile(ctx, m.preMetaFile)
 		if err != nil && !util.IsNotExistInExtStorage(err) {
+			log.Error("redo: meta manager flush meta delete file failed",
+				zap.String("namespace", m.changeFeedID.Namespace),
+				zap.String("changefeed", m.changeFeedID.ID),
+				zap.Error(err))
 			return errors.WrapError(errors.ErrExternalStorageAPI, err)
 		}
 	}
@@ -418,8 +461,8 @@ func (m *metaManager) Cleanup(ctx context.Context) error {
 	return m.deleteAllLogs(ctx)
 }
 
-func (m *metaManager) bgFlushMeta(egCtx context.Context, flushIntervalInMs int64) (err error) {
-	ticker := time.NewTicker(time.Duration(flushIntervalInMs) * time.Millisecond)
+func (m *metaManager) bgFlushMeta(egCtx context.Context) (err error) {
+	ticker := time.NewTicker(time.Duration(m.flushIntervalInMs) * time.Millisecond)
 	defer func() {
 		ticker.Stop()
 		log.Info("redo metaManager bgFlushMeta exits",
diff --git a/cdc/redo/writer/blackhole/blackhole_log_writer.go b/cdc/redo/writer/blackhole/blackhole_log_writer.go
index 24cc03bd3..9a877f262 100644
--- a/cdc/redo/writer/blackhole/blackhole_log_writer.go
+++ b/cdc/redo/writer/blackhole/blackhole_log_writer.go
@@ -26,14 +26,21 @@ var _ writer.RedoLogWriter = (*blackHoleWriter)(nil)
 
 // blackHoleSink defines a blackHole storage, it receives events and persists
 // without any latency
-type blackHoleWriter struct{}
+type blackHoleWriter struct {
+	invalid bool
+}
 
 // NewLogWriter creates a blackHole writer
-func NewLogWriter() *blackHoleWriter {
-	return &blackHoleWriter{}
+func NewLogWriter(invalid bool) *blackHoleWriter {
+	return &blackHoleWriter{
+		invalid: invalid,
+	}
 }
 
 func (bs *blackHoleWriter) WriteEvents(_ context.Context, events ...writer.RedoEvent) (err error) {
+	if bs.invalid {
+		return errors.New("[WriteLog] invalid black hole writer")
+	}
 	if len(events) == 0 {
 		return nil
 	}
@@ -45,30 +52,12 @@ func (bs *blackHoleWriter) WriteEvents(_ context.Context, events ...writer.RedoE
 }
 
 func (bs *blackHoleWriter) FlushLog(_ context.Context) error {
+	if bs.invalid {
+		return errors.New("[FlushLog] invalid black hole writer")
+	}
 	return nil
 }
 
 func (bs *blackHoleWriter) Close() error {
 	return nil
 }
-
-type invalidBlackHoleWriter struct {
-	*blackHoleWriter
-}
-
-// NewInvalidLogWriter creates a invalid blackHole writer
-func NewInvalidLogWriter(rl writer.RedoLogWriter) *invalidBlackHoleWriter {
-	return &invalidBlackHoleWriter{
-		blackHoleWriter: rl.(*blackHoleWriter),
-	}
-}
-
-func (ibs *invalidBlackHoleWriter) WriteEvents(
-	_ context.Context, _ ...writer.RedoEvent,
-) (err error) {
-	return errors.New("[WriteLog] invalid black hole writer")
-}
-
-func (ibs *invalidBlackHoleWriter) FlushLog(_ context.Context) error {
-	return errors.New("[FlushLog] invalid black hole writer")
-}
diff --git a/cdc/redo/writer/factory/factory.go b/cdc/redo/writer/factory/factory.go
index 7684b3f06..796d803ae 100644
--- a/cdc/redo/writer/factory/factory.go
+++ b/cdc/redo/writer/factory/factory.go
@@ -15,7 +15,9 @@ package factory
 
 import (
 	"context"
+	"strings"
 
+	"github.com/pingcap/tidb/br/pkg/storage"
 	"github.com/pingcap/tiflow/cdc/redo/writer"
 	"github.com/pingcap/tiflow/cdc/redo/writer/blackhole"
 	"github.com/pingcap/tiflow/cdc/redo/writer/file"
@@ -28,14 +30,23 @@ import (
 func NewRedoLogWriter(
 	ctx context.Context, lwCfg *writer.LogWriterConfig,
 ) (writer.RedoLogWriter, error) {
-	scheme := lwCfg.URI.Scheme
-	if !redo.IsValidConsistentStorage(scheme) {
-		return nil, errors.ErrConsistentStorage.GenWithStackByArgs(scheme)
+	uri, err := storage.ParseRawURL(lwCfg.Storage)
+	if err != nil {
+		return nil, err
 	}
 
-	if redo.IsBlackholeStorage(scheme) {
-		return blackhole.NewLogWriter(), nil
+	if !redo.IsValidConsistentStorage(uri.Scheme) {
+		return nil, errors.ErrConsistentStorage.GenWithStackByArgs(uri.Scheme)
 	}
+
+	lwCfg.URI = uri
+	lwCfg.UseExternalStorage = redo.IsExternalStorage(uri.Scheme)
+
+	if redo.IsBlackholeStorage(uri.Scheme) {
+		invalid := strings.HasSuffix(uri.Scheme, "invalid")
+		return blackhole.NewLogWriter(invalid), nil
+	}
+
 	if lwCfg.UseFileBackend {
 		return file.NewLogWriter(ctx, lwCfg)
 	}
diff --git a/cdc/redo/writer/file/file.go b/cdc/redo/writer/file/file.go
index e895bffe8..e12a5df59 100644
--- a/cdc/redo/writer/file/file.go
+++ b/cdc/redo/writer/file/file.go
@@ -86,7 +86,7 @@ func NewFileWriter(
 	var extStorage storage.ExternalStorage
 	if cfg.UseExternalStorage {
 		var err error
-		extStorage, err = redo.InitExternalStorage(ctx, cfg.URI)
+		extStorage, err = redo.InitExternalStorage(ctx, *cfg.URI)
 		if err != nil {
 			return nil, err
 		}
@@ -129,7 +129,7 @@ func NewFileWriter(
 	// if we use S3 as the remote storage, a file allocator can be leveraged to
 	// pre-allocate files for us.
 	// TODO: test whether this improvement can also be applied to NFS.
-	if cfg.UseExternalStorage {
+	if w.cfg.UseExternalStorage {
 		w.allocator = fsutil.NewFileAllocator(cfg.Dir, cfg.LogType, cfg.MaxLogSizeInBytes)
 	}
 
diff --git a/cdc/redo/writer/memory/encoding_worker.go b/cdc/redo/writer/memory/encoding_worker.go
index fa59c6dfe..f2dfa1025 100644
--- a/cdc/redo/writer/memory/encoding_worker.go
+++ b/cdc/redo/writer/memory/encoding_worker.go
@@ -123,7 +123,7 @@ func newEncodingWorkerGroup(workerNum int) *encodingWorkerGroup {
 func (e *encodingWorkerGroup) Run(ctx context.Context) (err error) {
 	defer func() {
 		close(e.closed)
-		if err != nil {
+		if err != nil && errors.Cause(err) != context.Canceled {
 			log.Warn("redo fileWorkerGroup closed with error", zap.Error(err))
 		}
 	}()
diff --git a/cdc/redo/writer/memory/file_worker.go b/cdc/redo/writer/memory/file_worker.go
index 771888a41..45b366117 100644
--- a/cdc/redo/writer/memory/file_worker.go
+++ b/cdc/redo/writer/memory/file_worker.go
@@ -138,7 +138,7 @@ func (f *fileWorkerGroup) Run(
 ) (err error) {
 	defer func() {
 		f.close()
-		if err != nil {
+		if err != nil && errors.Cause(err) != context.Canceled {
 			log.Warn("redo file workers closed with error", zap.Error(err))
 		}
 	}()
diff --git a/cdc/redo/writer/memory/mem_log_writer.go b/cdc/redo/writer/memory/mem_log_writer.go
index a47a83e85..0460fdbf0 100644
--- a/cdc/redo/writer/memory/mem_log_writer.go
+++ b/cdc/redo/writer/memory/mem_log_writer.go
@@ -54,11 +54,11 @@ func NewLogWriter(
 	}
 	// "nfs" and "local" scheme are converted to "file" scheme
 	if !cfg.UseExternalStorage {
-		redo.FixLocalScheme(&cfg.URI)
+		redo.FixLocalScheme(cfg.URI)
 		cfg.UseExternalStorage = redo.IsExternalStorage(cfg.URI.Scheme)
 	}
 
-	extStorage, err := redo.InitExternalStorage(ctx, cfg.URI)
+	extStorage, err := redo.InitExternalStorage(ctx, *cfg.URI)
 	if err != nil {
 		return nil, err
 	}
diff --git a/cdc/redo/writer/writer.go b/cdc/redo/writer/writer.go
index 18779997b..42ac64d4f 100644
--- a/cdc/redo/writer/writer.go
+++ b/cdc/redo/writer/writer.go
@@ -52,7 +52,7 @@ type LogWriterConfig struct {
 	CaptureID    model.CaptureID
 	ChangeFeedID model.ChangeFeedID
 
-	URI                url.URL
+	URI                *url.URL
 	UseExternalStorage bool
 	Dir                string
 	MaxLogSizeInBytes  int64
diff --git a/cdc/scheduler/internal/v3/replication/metrics.go b/cdc/scheduler/internal/v3/replication/metrics.go
index ccd546f67..a427035ef 100644
--- a/cdc/scheduler/internal/v3/replication/metrics.go
+++ b/cdc/scheduler/internal/v3/replication/metrics.go
@@ -125,6 +125,21 @@ var (
 			Name:      "slow_table_region_count",
 			Help:      "The number of regions captured by the slowest table",
 		}, []string{"namespace", "changefeed"})
+
+	slowestTablePullerResolvedTs = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: "ticdc",
+			Subsystem: "scheduler",
+			Name:      "slow_table_puller_resolved_ts",
+			Help:      "Puller Slowest ResolvedTs",
+		}, []string{"namespace", "changefeed"})
+	slowestTablePullerResolvedTsLag = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: "ticdc",
+			Subsystem: "scheduler",
+			Name:      "slow_table_puller_resolved_ts_lag",
+			Help:      "Puller Slowest ResolvedTs lag",
+		}, []string{"namespace", "changefeed"})
 )
 
 // InitMetrics registers all metrics used in scheduler
@@ -144,4 +159,7 @@ func InitMetrics(registry *prometheus.Registry) {
 	registry.MustRegister(slowestTableStageCheckpointTsLagHistogramVec)
 	registry.MustRegister(slowestTableStageResolvedTsLagHistogramVec)
 	registry.MustRegister(slowestTableRegionGaugeVec)
+
+	registry.MustRegister(slowestTablePullerResolvedTs)
+	registry.MustRegister(slowestTablePullerResolvedTsLag)
 }
diff --git a/cdc/scheduler/internal/v3/replication/replication_manager.go b/cdc/scheduler/internal/v3/replication/replication_manager.go
index 04cf71aa8..04455d263 100644
--- a/cdc/scheduler/internal/v3/replication/replication_manager.go
+++ b/cdc/scheduler/internal/v3/replication/replication_manager.go
@@ -15,6 +15,7 @@ package replication
 
 import (
 	"container/heap"
+	"fmt"
 	"math"
 	"time"
 
@@ -49,12 +50,30 @@ type BurstBalance struct {
 	MoveTables   []MoveTable
 }
 
+func (b BurstBalance) String() string {
+	if len(b.AddTables) != 0 {
+		return fmt.Sprintf("BurstBalance, add tables: %v", b.AddTables)
+	}
+	if len(b.RemoveTables) != 0 {
+		return fmt.Sprintf("BurstBalance, remove tables: %v", b.RemoveTables)
+	}
+	if len(b.MoveTables) != 0 {
+		return fmt.Sprintf("BurstBalance, move tables: %v", b.MoveTables)
+	}
+	return "BurstBalance, no tables"
+}
+
 // MoveTable is a schedule task for moving a table.
 type MoveTable struct {
 	TableID     model.TableID
 	DestCapture model.CaptureID
 }
 
+func (t MoveTable) String() string {
+	return fmt.Sprintf("MoveTable, tableID: %d, dest: %s",
+		t.TableID, t.DestCapture)
+}
+
 // AddTable is a schedule task for adding a table.
 type AddTable struct {
 	TableID      model.TableID
@@ -62,12 +81,22 @@ type AddTable struct {
 	CheckpointTs model.Ts
 }
 
+func (t AddTable) String() string {
+	return fmt.Sprintf("AddTable, tableID: %d, capture: %s, checkpointTs: %d",
+		t.TableID, t.CaptureID, t.CheckpointTs)
+}
+
 // RemoveTable is a schedule task for removing a table.
 type RemoveTable struct {
 	TableID   model.TableID
 	CaptureID model.CaptureID
 }
 
+func (t RemoveTable) String() string {
+	return fmt.Sprintf("RemoveTable, tableID: %d, capture: %s",
+		t.TableID, t.CaptureID)
+}
+
 // ScheduleTask is a schedule task that wraps add/move/remove table tasks.
 type ScheduleTask struct { //nolint:revive
 	MoveTable    *MoveTable
@@ -92,6 +121,22 @@ func (s *ScheduleTask) Name() string {
 	return "unknown"
 }
 
+func (s *ScheduleTask) String() string {
+	if s.MoveTable != nil {
+		return s.MoveTable.String()
+	}
+	if s.AddTable != nil {
+		return s.AddTable.String()
+	}
+	if s.RemoveTable != nil {
+		return s.RemoveTable.String()
+	}
+	if s.BurstBalance != nil {
+		return s.BurstBalance.String()
+	}
+	return ""
+}
+
 // Manager manages replications and running scheduling tasks.
 type Manager struct { //nolint:revive
 	tables map[model.TableID]*ReplicationSet
@@ -100,7 +145,8 @@ type Manager struct { //nolint:revive
 	maxTaskConcurrency int
 
 	changefeedID           model.ChangeFeedID
-	slowestTableID         model.TableID
+	slowestPuller          model.TableID
+	slowestSink            model.TableID
 	slowTableHeap          SetHeap
 	acceptAddTableTask     int
 	acceptRemoveTableTask  int
@@ -527,8 +573,11 @@ func (r *Manager) AdvanceCheckpoint(
 		}
 	}()
 
+	r.slowestPuller = model.TableID(0)
+	r.slowestSink = model.TableID(0)
+	var slowestPullerResolvedTs uint64 = math.MaxUint64
+
 	newCheckpointTs, newResolvedTs = math.MaxUint64, math.MaxUint64
-	slowestTableID := int64(0)
 	for _, tableID := range currentTables {
 		table, ok := r.tables[tableID]
 		if !ok {
@@ -551,14 +600,18 @@ func (r *Manager) AdvanceCheckpoint(
 		// Find the minimum checkpoint ts and resolved ts.
 		if newCheckpointTs > table.Checkpoint.CheckpointTs {
 			newCheckpointTs = table.Checkpoint.CheckpointTs
-			slowestTableID = tableID
+			r.slowestSink = tableID
 		}
 		if newResolvedTs > table.Checkpoint.ResolvedTs {
 			newResolvedTs = table.Checkpoint.ResolvedTs
 		}
-	}
-	if slowestTableID != 0 {
-		r.slowestTableID = slowestTableID
+		// Find the minimum puller resolved ts.
+		if pullerCkpt, ok := table.Stats.StageCheckpoints["puller-egress"]; ok {
+			if slowestPullerResolvedTs > pullerCkpt.ResolvedTs {
+				slowestPullerResolvedTs = pullerCkpt.ResolvedTs
+				r.slowestPuller = tableID
+			}
+		}
 	}
 
 	// If currentTables is empty, we should advance newResolvedTs to global barrier ts and
@@ -649,9 +702,9 @@ func (r *Manager) CollectMetrics() {
 	cf := r.changefeedID
 	tableGauge.
 		WithLabelValues(cf.Namespace, cf.ID).Set(float64(len(r.tables)))
-	if table, ok := r.tables[r.slowestTableID]; ok {
+	if table, ok := r.tables[r.slowestSink]; ok {
 		slowestTableIDGauge.
-			WithLabelValues(cf.Namespace, cf.ID).Set(float64(r.slowestTableID))
+			WithLabelValues(cf.Namespace, cf.ID).Set(float64(r.slowestSink))
 		slowestTableStateGauge.
 			WithLabelValues(cf.Namespace, cf.ID).Set(float64(table.State))
 		phyCkpTs := oracle.ExtractPhysical(table.Checkpoint.CheckpointTs)
@@ -732,6 +785,17 @@ func (r *Manager) CollectMetrics() {
 			WithLabelValues(cf.Namespace, cf.ID, ReplicationSetState(s).String()).
 			Set(float64(counter))
 	}
+
+	if table, ok := r.tables[r.slowestSink]; ok {
+		if pullerCkpt, ok := table.Stats.StageCheckpoints["puller-egress"]; ok {
+			phyCkptTs := oracle.ExtractPhysical(pullerCkpt.ResolvedTs)
+			slowestTablePullerResolvedTs.WithLabelValues(cf.Namespace, cf.ID).Set(float64(phyCkptTs))
+
+			phyCurrentTs := oracle.ExtractPhysical(table.Stats.CurrentTs)
+			lag := float64(phyCurrentTs-phyCkptTs) / 1e3
+			slowestTablePullerResolvedTsLag.WithLabelValues(cf.Namespace, cf.ID).Set(lag)
+		}
+	}
 }
 
 // CleanMetrics cleans metrics.
diff --git a/cdc/scheduler/internal/v3/replication/replication_set.go b/cdc/scheduler/internal/v3/replication/replication_set.go
index 128e31d1d..2dc92e75d 100644
--- a/cdc/scheduler/internal/v3/replication/replication_set.go
+++ b/cdc/scheduler/internal/v3/replication/replication_set.go
@@ -200,8 +200,10 @@ func NewReplicationSet(
 			// We need to wait its state becomes Stopped or Absent before
 			// proceeding further scheduling.
 			log.Warn("schedulerv3: found a stopping capture during initializing",
-				zap.Any("replicationSet", r),
+				zap.String("namespace", r.Changefeed.Namespace),
+				zap.String("changefeed", r.Changefeed.ID),
 				zap.Int64("tableID", table.TableID),
+				zap.Any("replicationSet", r),
 				zap.Any("status", tableStatus))
 			err := r.setCapture(captureID, RoleUndetermined)
 			if err != nil {
@@ -213,8 +215,10 @@ func NewReplicationSet(
 			// Ignore stop state.
 		default:
 			log.Warn("schedulerv3: unknown table state",
-				zap.Any("replicationSet", r),
+				zap.String("namespace", r.Changefeed.Namespace),
+				zap.String("changefeed", r.Changefeed.ID),
 				zap.Int64("tableID", table.TableID),
+				zap.Any("replicationSet", r),
 				zap.Any("status", tableStatus))
 		}
 	}
@@ -238,6 +242,8 @@ func NewReplicationSet(
 		r.State = ReplicationSetStateRemoving
 	}
 	log.Info("schedulerv3: initialize replication set",
+		zap.String("namespace", r.Changefeed.Namespace),
+		zap.String("changefeed", r.Changefeed.ID),
 		zap.Any("replicationSet", r))
 
 	return r, nil
@@ -289,7 +295,10 @@ func (r *ReplicationSet) clearCapture(captureID model.CaptureID, role Role) erro
 
 func (r *ReplicationSet) promoteSecondary(captureID model.CaptureID) error {
 	if r.Primary == captureID {
-		log.Warn("schedulerv3: capture is already promoted",
+		log.Warn("schedulerv3: capture is already promoted as the primary",
+			zap.String("namespace", r.Changefeed.Namespace),
+			zap.String("changefeed", r.Changefeed.ID),
+			zap.String("captureID", captureID),
 			zap.Any("replicationSet", r))
 		return nil
 	}
@@ -317,6 +326,8 @@ func (r *ReplicationSet) inconsistentError(
 	input *tablepb.TableStatus, captureID model.CaptureID, msg string, fields ...zap.Field,
 ) error {
 	fields = append(fields, []zap.Field{
+		zap.String("namespace", r.Changefeed.Namespace),
+		zap.String("changefeed", r.Changefeed.ID),
 		zap.String("captureID", captureID),
 		zap.Stringer("tableState", input),
 		zap.Any("replicationSet", r),
@@ -330,6 +341,8 @@ func (r *ReplicationSet) multiplePrimaryError(
 	input *tablepb.TableStatus, captureID model.CaptureID, msg string, fields ...zap.Field,
 ) error {
 	fields = append(fields, []zap.Field{
+		zap.String("namespace", r.Changefeed.Namespace),
+		zap.String("changefeed", r.Changefeed.ID),
 		zap.String("captureID", captureID),
 		zap.Stringer("tableState", input),
 		zap.Any("replicationSet", r),
@@ -392,7 +405,7 @@ func (r *ReplicationSet) poll(
 		var msg *schedulepb.Message
 		switch r.State {
 		case ReplicationSetStateAbsent:
-			msg, stateChanged, err = r.pollOnAbsent(input, captureID)
+			stateChanged, err = r.pollOnAbsent(input, captureID)
 		case ReplicationSetStatePrepare:
 			msg, stateChanged, err = r.pollOnPrepare(input, captureID)
 		case ReplicationSetStateCommit:
@@ -413,12 +426,12 @@ func (r *ReplicationSet) poll(
 		}
 		if stateChanged {
 			log.Info("schedulerv3: replication state transition, poll",
+				zap.String("namespace", r.Changefeed.Namespace),
+				zap.String("changefeed", r.Changefeed.ID),
 				zap.Stringer("tableState", input),
 				zap.String("captureID", captureID),
 				zap.Stringer("old", oldState),
-				zap.Stringer("new", r.State),
-				zap.String("namespace", r.Changefeed.Namespace),
-				zap.String("changefeed", r.Changefeed.ID))
+				zap.Stringer("new", r.State))
 		}
 	}
 
@@ -428,26 +441,28 @@ func (r *ReplicationSet) poll(
 //nolint:unparam
 func (r *ReplicationSet) pollOnAbsent(
 	input *tablepb.TableStatus, captureID model.CaptureID,
-) (*schedulepb.Message, bool, error) {
+) (bool, error) {
 	switch input.State {
 	case tablepb.TableStateAbsent:
 		r.State = ReplicationSetStatePrepare
 		err := r.setCapture(captureID, RoleSecondary)
-		return nil, true, errors.Trace(err)
+		return true, errors.Trace(err)
 
 	case tablepb.TableStateStopped:
 		// Ignore stopped table state as a capture may shutdown unexpectedly.
-		return nil, false, nil
+		return false, nil
 	case tablepb.TableStatePreparing,
 		tablepb.TableStatePrepared,
 		tablepb.TableStateReplicating,
 		tablepb.TableStateStopping:
 	}
 	log.Warn("schedulerv3: ignore input, unexpected replication set state",
+		zap.String("namespace", r.Changefeed.Namespace),
+		zap.String("changefeed", r.Changefeed.ID),
 		zap.Stringer("tableState", input),
 		zap.String("captureID", captureID),
 		zap.Any("replicationSet", r))
-	return nil, false, nil
+	return false, nil
 }
 
 func (r *ReplicationSet) pollOnPrepare(
@@ -491,6 +506,8 @@ func (r *ReplicationSet) pollOnPrepare(
 			// Primary is stopped, but we may still has secondary.
 			// Clear primary and promote secondary when it's prepared.
 			log.Info("schedulerv3: primary is stopped during Prepare",
+				zap.String("namespace", r.Changefeed.Namespace),
+				zap.String("changefeed", r.Changefeed.ID),
 				zap.Stringer("tableState", input),
 				zap.String("captureID", captureID),
 				zap.Any("replicationSet", r))
@@ -499,6 +516,8 @@ func (r *ReplicationSet) pollOnPrepare(
 		}
 		if r.isInRole(captureID, RoleSecondary) {
 			log.Info("schedulerv3: capture is stopped during Prepare",
+				zap.String("namespace", r.Changefeed.Namespace),
+				zap.String("changefeed", r.Changefeed.ID),
 				zap.Stringer("tableState", input),
 				zap.String("captureID", captureID),
 				zap.Any("replicationSet", r))
@@ -519,6 +538,8 @@ func (r *ReplicationSet) pollOnPrepare(
 		}
 	}
 	log.Warn("schedulerv3: ignore input, unexpected replication set state",
+		zap.String("namespace", r.Changefeed.Namespace),
+		zap.String("changefeed", r.Changefeed.ID),
 		zap.Stringer("tableState", input),
 		zap.String("captureID", captureID),
 		zap.Any("replicationSet", r))
@@ -550,6 +571,8 @@ func (r *ReplicationSet) pollOnCommit(
 				// before promoting the secondary, otherwise there may be two
 				// primary that write data and lead to data inconsistency.
 				log.Info("schedulerv3: there are unknown captures during commit",
+					zap.String("namespace", r.Changefeed.Namespace),
+					zap.String("changefeed", r.Changefeed.ID),
 					zap.Any("replicationSet", r),
 					zap.Stringer("tableState", input),
 					zap.String("captureID", captureID))
@@ -562,6 +585,8 @@ func (r *ReplicationSet) pollOnCommit(
 			}
 
 			log.Info("schedulerv3: promote secondary, no primary",
+				zap.String("namespace", r.Changefeed.Namespace),
+				zap.String("changefeed", r.Changefeed.ID),
 				zap.Any("replicationSet", r),
 				zap.Stringer("tableState", input),
 				zap.String("captureID", captureID))
@@ -591,6 +616,8 @@ func (r *ReplicationSet) pollOnCommit(
 			if !r.hasRole(RoleSecondary) {
 				// If there is no secondary, transit to Absent.
 				log.Info("schedulerv3: primary is stopped during Commit",
+					zap.String("namespace", r.Changefeed.Namespace),
+					zap.String("changefeed", r.Changefeed.ID),
 					zap.Stringer("tableState", input),
 					zap.String("captureID", captureID),
 					zap.Any("replicationSet", r))
@@ -604,6 +631,8 @@ func (r *ReplicationSet) pollOnCommit(
 				return nil, false, errors.Trace(err)
 			}
 			log.Info("schedulerv3: replication state promote secondary",
+				zap.String("namespace", r.Changefeed.Namespace),
+				zap.String("changefeed", r.Changefeed.ID),
 				zap.Any("replicationSet", r),
 				zap.Stringer("tableState", input),
 				zap.String("original", original),
@@ -626,6 +655,8 @@ func (r *ReplicationSet) pollOnCommit(
 			// upon entering Commit state. Do not change state and wait
 			// the original primary reports its table.
 			log.Info("schedulerv3: secondary is stopped during Commit",
+				zap.String("namespace", r.Changefeed.Namespace),
+				zap.String("changefeed", r.Changefeed.ID),
 				zap.Stringer("tableState", input),
 				zap.String("captureID", captureID),
 				zap.Any("replicationSet", r))
@@ -640,6 +671,8 @@ func (r *ReplicationSet) pollOnCommit(
 			return nil, true, nil
 		} else if r.isInRole(captureID, RoleUndetermined) {
 			log.Info("schedulerv3: capture is stopped during Commit",
+				zap.String("namespace", r.Changefeed.Namespace),
+				zap.String("changefeed", r.Changefeed.ID),
 				zap.Stringer("tableState", input),
 				zap.String("captureID", captureID),
 				zap.Any("replicationSet", r))
@@ -686,6 +719,8 @@ func (r *ReplicationSet) pollOnCommit(
 			return nil, false, nil
 		} else if r.isInRole(captureID, RoleUndetermined) {
 			log.Info("schedulerv3: capture is stopping during Commit",
+				zap.String("namespace", r.Changefeed.Namespace),
+				zap.String("changefeed", r.Changefeed.ID),
 				zap.Stringer("tableState", input),
 				zap.String("captureID", captureID),
 				zap.Any("replicationSet", r))
@@ -695,6 +730,8 @@ func (r *ReplicationSet) pollOnCommit(
 	case tablepb.TableStatePreparing:
 	}
 	log.Warn("schedulerv3: ignore input, unexpected replication set state",
+		zap.String("namespace", r.Changefeed.Namespace),
+		zap.String("changefeed", r.Changefeed.ID),
 		zap.Stringer("tableState", input),
 		zap.String("captureID", captureID),
 		zap.Any("replicationSet", r))
@@ -724,6 +761,8 @@ func (r *ReplicationSet) pollOnReplicating(
 			// Primary is stopped, but we still has secondary.
 			// Clear primary and promote secondary when it's prepared.
 			log.Info("schedulerv3: primary is stopped during Replicating",
+				zap.String("namespace", r.Changefeed.Namespace),
+				zap.String("changefeed", r.Changefeed.ID),
 				zap.Stringer("tableState", input),
 				zap.String("captureID", captureID),
 				zap.Any("replicationSet", r))
@@ -733,6 +772,8 @@ func (r *ReplicationSet) pollOnReplicating(
 		}
 	}
 	log.Warn("schedulerv3: ignore input, unexpected replication set state",
+		zap.String("namespace", r.Changefeed.Namespace),
+		zap.String("changefeed", r.Changefeed.ID),
 		zap.Stringer("tableState", input),
 		zap.String("captureID", captureID),
 		zap.Any("replicationSet", r))
@@ -757,26 +798,30 @@ func (r *ReplicationSet) pollOnRemoving(
 			},
 		}, false, nil
 	case tablepb.TableStateAbsent, tablepb.TableStateStopped:
-		errField := zap.Skip()
+		var err error
 		if r.Primary == captureID {
 			r.clearPrimary()
 		} else if r.isInRole(captureID, RoleSecondary) {
-			err := r.clearCapture(captureID, RoleSecondary)
-			errField = zap.Error(err)
+			err = r.clearCapture(captureID, RoleSecondary)
 		} else {
-			err := r.clearCapture(captureID, RoleUndetermined)
-			errField = zap.Error(err)
+			err = r.clearCapture(captureID, RoleUndetermined)
+		}
+		if err != nil {
+			log.Warn("schedulerv3: replication state remove capture with error",
+				zap.String("namespace", r.Changefeed.Namespace),
+				zap.String("changefeed", r.Changefeed.ID),
+				zap.Any("replicationSet", r),
+				zap.Stringer("tableState", input),
+				zap.String("captureID", captureID),
+				zap.Error(err))
 		}
-		log.Info("schedulerv3: replication state remove capture",
-			zap.Any("replicationSet", r),
-			zap.Stringer("tableState", input),
-			zap.String("captureID", captureID),
-			errField)
 		return nil, false, nil
 	case tablepb.TableStateStopping:
 		return nil, false, nil
 	}
 	log.Warn("schedulerv3: ignore input, unexpected replication set state",
+		zap.String("namespace", r.Changefeed.Namespace),
+		zap.String("changefeed", r.Changefeed.ID),
 		zap.Stringer("tableState", input),
 		zap.String("captureID", captureID),
 		zap.Any("replicationSet", r))
@@ -795,7 +840,10 @@ func (r *ReplicationSet) handleAddTable(
 	// Ignore add table if it's not in Absent state.
 	if r.State != ReplicationSetStateAbsent {
 		log.Warn("schedulerv3: add table is ignored",
-			zap.Any("replicationSet", r), zap.Int64("tableID", r.TableID))
+			zap.String("namespace", r.Changefeed.Namespace),
+			zap.String("changefeed", r.Changefeed.ID),
+			zap.Int64("tableID", r.TableID),
+			zap.Any("replicationSet", r))
 		return nil, nil
 	}
 	err := r.setCapture(captureID, RoleSecondary)
@@ -814,6 +862,8 @@ func (r *ReplicationSet) handleAddTable(
 	}
 
 	log.Info("schedulerv3: replication state transition, add table",
+		zap.String("namespace", r.Changefeed.Namespace),
+		zap.String("changefeed", r.Changefeed.ID),
 		zap.Any("replicationSet", r),
 		zap.Stringer("old", oldState), zap.Stringer("new", r.State))
 	return msgs, nil
@@ -825,7 +875,10 @@ func (r *ReplicationSet) handleMoveTable(
 	// Ignore move table if it has been removed already.
 	if r.hasRemoved() {
 		log.Warn("schedulerv3: move table is ignored",
-			zap.Any("replicationSet", r), zap.Int64("tableID", r.TableID))
+			zap.String("namespace", r.Changefeed.Namespace),
+			zap.String("changefeed", r.Changefeed.ID),
+			zap.Int64("tableID", r.TableID),
+			zap.Any("replicationSet", r))
 		return nil, nil
 	}
 	// Ignore move table if
@@ -833,7 +886,10 @@ func (r *ReplicationSet) handleMoveTable(
 	// 2) the dest capture is the primary.
 	if r.State != ReplicationSetStateReplicating || r.Primary == dest {
 		log.Warn("schedulerv3: move table is ignored",
-			zap.Any("replicationSet", r), zap.Int64("tableID", r.TableID))
+			zap.String("namespace", r.Changefeed.Namespace),
+			zap.String("changefeed", r.Changefeed.ID),
+			zap.Int64("tableID", r.TableID),
+			zap.Any("replicationSet", r))
 		return nil, nil
 	}
 	oldState := r.State
@@ -843,8 +899,11 @@ func (r *ReplicationSet) handleMoveTable(
 		return nil, errors.Trace(err)
 	}
 	log.Info("schedulerv3: replication state transition, move table",
+		zap.String("namespace", r.Changefeed.Namespace),
+		zap.String("changefeed", r.Changefeed.ID),
+		zap.Stringer("new", r.State),
 		zap.Any("replicationSet", r),
-		zap.Stringer("old", oldState), zap.Stringer("new", r.State))
+		zap.Stringer("old", oldState))
 	status := tablepb.TableStatus{
 		TableID:    r.TableID,
 		State:      tablepb.TableStateAbsent,
@@ -857,20 +916,29 @@ func (r *ReplicationSet) handleRemoveTable() ([]*schedulepb.Message, error) {
 	// Ignore remove table if it has been removed already.
 	if r.hasRemoved() {
 		log.Warn("schedulerv3: remove table is ignored",
-			zap.Any("replicationSet", r), zap.Int64("tableID", r.TableID))
+			zap.String("namespace", r.Changefeed.Namespace),
+			zap.String("changefeed", r.Changefeed.ID),
+			zap.Int64("tableID", r.TableID),
+			zap.Any("replicationSet", r))
 		return nil, nil
 	}
 	// Ignore remove table if it's not in Replicating state.
 	if r.State != ReplicationSetStateReplicating {
 		log.Warn("schedulerv3: remove table is ignored",
-			zap.Any("replicationSet", r), zap.Int64("tableID", r.TableID))
+			zap.String("namespace", r.Changefeed.Namespace),
+			zap.String("changefeed", r.Changefeed.ID),
+			zap.Int64("tableID", r.TableID),
+			zap.Any("replicationSet", r))
 		return nil, nil
 	}
 	oldState := r.State
 	r.State = ReplicationSetStateRemoving
 	log.Info("schedulerv3: replication state transition, remove table",
+		zap.String("namespace", r.Changefeed.Namespace),
+		zap.String("changefeed", r.Changefeed.ID),
+		zap.Int64("tableID", r.TableID),
 		zap.Any("replicationSet", r),
-		zap.Stringer("old", oldState), zap.Stringer("new", r.State))
+		zap.Stringer("old", oldState))
 	status := tablepb.TableStatus{
 		TableID: r.TableID,
 		State:   tablepb.TableStateReplicating,
@@ -904,7 +972,14 @@ func (r *ReplicationSet) handleCaptureShutdown(
 		TableID: r.TableID,
 		State:   tablepb.TableStateStopped,
 	}
+	oldState := r.State
 	msgs, err := r.poll(&status, captureID)
+	log.Info("schedulerv3: replication state transition, capture shutdown",
+		zap.String("namespace", r.Changefeed.Namespace),
+		zap.String("changefeed", r.Changefeed.ID),
+		zap.Int64("tableID", r.TableID),
+		zap.Any("replicationSet", r),
+		zap.Stringer("old", oldState), zap.Stringer("new", r.State))
 	return msgs, true, errors.Trace(err)
 }
 
@@ -913,6 +988,9 @@ func (r *ReplicationSet) updateCheckpointAndStats(
 ) {
 	if checkpoint.ResolvedTs < checkpoint.CheckpointTs {
 		log.Warn("schedulerv3: resolved ts should not less than checkpoint ts",
+			zap.String("namespace", r.Changefeed.Namespace),
+			zap.String("changefeed", r.Changefeed.ID),
+			zap.Int64("tableID", r.TableID),
 			zap.Any("replicationSet", r),
 			zap.Any("checkpoint", checkpoint))
 
@@ -929,6 +1007,9 @@ func (r *ReplicationSet) updateCheckpointAndStats(
 	}
 	if r.Checkpoint.ResolvedTs < r.Checkpoint.CheckpointTs {
 		log.Warn("schedulerv3: resolved ts should not less than checkpoint ts",
+			zap.String("namespace", r.Changefeed.Namespace),
+			zap.String("changefeed", r.Changefeed.ID),
+			zap.Int64("tableID", r.TableID),
 			zap.Any("replicationSet", r),
 			zap.Any("checkpointTs", r.Checkpoint.CheckpointTs),
 			zap.Any("resolvedTs", r.Checkpoint.ResolvedTs))
diff --git a/cdc/scheduler/internal/v3/scheduler/scheduler_basic.go b/cdc/scheduler/internal/v3/scheduler/scheduler_basic.go
index 1e1b7299d..24ebd685a 100644
--- a/cdc/scheduler/internal/v3/scheduler/scheduler_basic.go
+++ b/cdc/scheduler/internal/v3/scheduler/scheduler_basic.go
@@ -105,7 +105,7 @@ func (b *basicScheduler) Schedule(
 			zap.Strings("captureIDs", captureIDs),
 			zap.Int64s("tableIDs", newTables))
 		tasks = append(
-			tasks, newBurstAddTables(checkpointTs, newTables, captureIDs))
+			tasks, newBurstAddTables(b.changefeedID, checkpointTs, newTables, captureIDs))
 	}
 
 	// Build remove table tasks.
@@ -114,7 +114,7 @@ func (b *basicScheduler) Schedule(
 	// Fast path for check whether two sets are identical:
 	// If the length of currentTables and replications are equal,
 	// and for all tables in currentTables have a record in replications.
-	if !tablesLenEqual || !tablesAllFind {
+	if !(tablesLenEqual && tablesAllFind) {
 		// The two sets are not identical. We need to find removed tables.
 		intersectionTable := make(map[model.TableID]struct{}, len(currentTables))
 		for _, tableID := range currentTables {
@@ -131,9 +131,10 @@ func (b *basicScheduler) Schedule(
 				rmTables = append(rmTables, tableID)
 			}
 		}
-		if len(rmTables) > 0 {
-			tasks = append(tasks,
-				newBurstRemoveTables(rmTables, replications, b.changefeedID))
+
+		removeTableTasks := newBurstRemoveTables(rmTables, replications, b.changefeedID)
+		if removeTableTasks != nil {
+			tasks = append(tasks, removeTableTasks)
 		}
 	}
 	return tasks
@@ -141,24 +142,34 @@ func (b *basicScheduler) Schedule(
 
 // newBurstAddTables add each new table to captures in a round-robin way.
 func newBurstAddTables(
+	changefeedID model.ChangeFeedID,
 	checkpointTs model.Ts, newTables []model.TableID, captureIDs []model.CaptureID,
 ) *replication.ScheduleTask {
 	idx := 0
 	tables := make([]replication.AddTable, 0, len(newTables))
 	for _, tableID := range newTables {
+		targetCapture := captureIDs[idx]
 		tables = append(tables, replication.AddTable{
 			TableID:      tableID,
-			CaptureID:    captureIDs[idx],
+			CaptureID:    targetCapture,
 			CheckpointTs: checkpointTs,
 		})
+		log.Info("schedulerv3: burst add table",
+			zap.String("namespace", changefeedID.Namespace),
+			zap.String("changefeed", changefeedID.ID),
+			zap.String("captureID", targetCapture),
+			zap.Any("tableID", tableID))
+
 		idx++
 		if idx >= len(captureIDs) {
 			idx = 0
 		}
 	}
-	return &replication.ScheduleTask{BurstBalance: &replication.BurstBalance{
-		AddTables: tables,
-	}}
+	return &replication.ScheduleTask{
+		BurstBalance: &replication.BurstBalance{
+			AddTables: tables,
+		},
+	}
 }
 
 func newBurstRemoveTables(
@@ -174,7 +185,8 @@ func newBurstRemoveTables(
 			break
 		}
 		if captureID == "" {
-			log.Warn("schedulerv3: primary or secondary not found for removed table",
+			log.Warn("schedulerv3: primary or secondary not found for removed table,"+
+				"this may happen if the capture shutdown",
 				zap.String("namespace", changefeedID.Namespace),
 				zap.String("changefeed", changefeedID.ID),
 				zap.Any("table", rep))
@@ -184,8 +196,20 @@ func newBurstRemoveTables(
 			TableID:   tableID,
 			CaptureID: captureID,
 		})
+		log.Info("schedulerv3: burst remove table",
+			zap.String("namespace", changefeedID.Namespace),
+			zap.String("changefeed", changefeedID.ID),
+			zap.String("captureID", captureID),
+			zap.Any("tableID", tableID))
+	}
+
+	if len(tables) == 0 {
+		return nil
+	}
+
+	return &replication.ScheduleTask{
+		BurstBalance: &replication.BurstBalance{
+			RemoveTables: tables,
+		},
 	}
-	return &replication.ScheduleTask{BurstBalance: &replication.BurstBalance{
-		RemoveTables: tables,
-	}}
 }
diff --git a/cdc/scheduler/internal/v3/scheduler/scheduler_manager.go b/cdc/scheduler/internal/v3/scheduler/scheduler_manager.go
index b48f25303..1a4f0a549 100644
--- a/cdc/scheduler/internal/v3/scheduler/scheduler_manager.go
+++ b/cdc/scheduler/internal/v3/scheduler/scheduler_manager.go
@@ -86,12 +86,6 @@ func (sm *Manager) Schedule(
 			sm.tasksCounter[name]++
 		}
 		if len(tasks) != 0 {
-			log.Info("schedulerv3: new schedule task",
-				zap.String("namespace", sm.changefeedID.Namespace),
-				zap.String("changefeed", sm.changefeedID.ID),
-				zap.Int("taskNumber", len(tasks)),
-				zap.Any("task", tasks),
-				zap.String("scheduler", scheduler.Name()))
 			return tasks
 		}
 	}
diff --git a/cdc/server/server.go b/cdc/server/server.go
index 1513c55ff..b7c510c2d 100644
--- a/cdc/server/server.go
+++ b/cdc/server/server.go
@@ -43,7 +43,6 @@ import (
 	"github.com/pingcap/tiflow/pkg/p2p"
 	"github.com/pingcap/tiflow/pkg/pdutil"
 	"github.com/pingcap/tiflow/pkg/tcpserver"
-	"github.com/pingcap/tiflow/pkg/util"
 	p2pProto "github.com/pingcap/tiflow/proto/p2p"
 	pd "github.com/tikv/pd/client"
 	"go.uber.org/zap"
@@ -61,6 +60,8 @@ const (
 	maxHTTPConnection = 1000
 	// httpConnectionTimeout is used to limit a connection max alive time of http server.
 	httpConnectionTimeout = 10 * time.Minute
+	// maxGcTunerMemory is used to limit the max memory usage of cdc server. if the memory is larger than it, gc tuner will be disabled
+	maxGcTunerMemory = 512 * 1024 * 1024 * 1024
 )
 
 // Server is the interface for the TiCDC server
@@ -208,11 +209,7 @@ func (s *server) prepare(ctx context.Context) error {
 	if err := s.startActorSystems(ctx); err != nil {
 		return errors.Trace(err)
 	}
-
-	if err := s.setMemoryLimit(); err != nil {
-		return errors.Trace(err)
-	}
-
+	s.setMemoryLimit()
 	s.capture = capture.NewCapture(
 		s.pdEndpoints, cdcEtcdClient, s.grpcService,
 		s.tableActorSystem, s.sortEngineFactory, s.sorterSystem, s.pdClient)
@@ -220,22 +217,25 @@ func (s *server) prepare(ctx context.Context) error {
 	return nil
 }
 
-func (s *server) setMemoryLimit() error {
+func (s *server) setMemoryLimit() {
 	conf := config.GetGlobalServerConfig()
-	totalMemory, err := util.GetMemoryLimit()
-	if err != nil {
-		return errors.Trace(err)
+	if conf.GcTunerMemoryThreshold > maxGcTunerMemory {
+		// If total memory is larger than 512GB, we will not set memory limit.
+		// Because the memory limit is not accurate, and it is not necessary to set memory limit.
+		log.Info("total memory is larger than 512GB, skip setting memory limit",
+			zap.Uint64("bytes", conf.GcTunerMemoryThreshold),
+			zap.String("memory", humanize.IBytes(conf.GcTunerMemoryThreshold)),
+		)
+		return
 	}
-	if conf.MaxMemoryPercentage > 0 {
-		goMemLimit := totalMemory * uint64(conf.MaxMemoryPercentage) / 100
+	if conf.GcTunerMemoryThreshold > 0 {
 		gctuner.EnableGOGCTuner.Store(true)
-		gctuner.Tuning(goMemLimit)
+		gctuner.Tuning(conf.GcTunerMemoryThreshold)
 		log.Info("enable gctuner, set memory limit",
-			zap.Uint64("bytes", goMemLimit),
-			zap.String("memory", humanize.IBytes(goMemLimit)),
+			zap.Uint64("bytes", conf.GcTunerMemoryThreshold),
+			zap.String("memory", humanize.IBytes(conf.GcTunerMemoryThreshold)),
 		)
 	}
-	return nil
 }
 
 func (s *server) startActorSystems(ctx context.Context) error {
diff --git a/cdc/sink/mq/manager/kafka_manager.go b/cdc/sink/mq/manager/kafka_manager.go
index 99e78858f..3d6ec99b7 100644
--- a/cdc/sink/mq/manager/kafka_manager.go
+++ b/cdc/sink/mq/manager/kafka_manager.go
@@ -44,6 +44,8 @@ const (
 type kafkaTopicManager struct {
 	changefeedID model.ChangeFeedID
 
+	defaultTopic string
+
 	admin kafka.ClusterAdminClient
 
 	cfg *kafkaconfig.AutoCreateTopicConfig
@@ -59,11 +61,13 @@ type kafkaTopicManager struct {
 // NewKafkaTopicManager creates a new topic manager.
 func NewKafkaTopicManager(
 	ctx context.Context,
+	defaultTopic string,
 	admin kafka.ClusterAdminClient,
 	cfg *kafkaconfig.AutoCreateTopicConfig,
 ) *kafkaTopicManager {
 	changefeedID := contextutil.ChangefeedIDFromCtx(ctx)
 	mgr := &kafkaTopicManager{
+		defaultTopic:      defaultTopic,
 		changefeedID:      changefeedID,
 		admin:             admin,
 		cfg:               cfg,
@@ -113,8 +117,15 @@ func (m *kafkaTopicManager) backgroundRefreshMeta(ctx context.Context) {
 					zap.Error(err))
 			}
 
+			// it may happen the following case:
+			// 1. user create the default topic with partition number set as 3 manually
+			// 2. set the partition-number as 2 in the sink-uri.
+			// in the such case, we should use 2 instead of 3 as the partition number.
 			for _, detail := range topicMetaList {
 				partitionNum := int32(len(detail.Partitions))
+				if detail.Name == m.defaultTopic {
+					partitionNum = m.cfg.PartitionNum
+				}
 				m.tryUpdatePartitionsAndLogging(detail.Name, partitionNum)
 			}
 
@@ -296,6 +307,9 @@ func (m *kafkaTopicManager) CreateTopicAndWaitUntilVisible(topicName string) (in
 					zap.String("topic", topicName),
 					zap.Int32("partitionNumber", int32(len(detail.Partitions))))
 				partitionNum := int32(len(detail.Partitions))
+				if topicName == m.defaultTopic {
+					partitionNum = m.cfg.PartitionNum
+				}
 				m.tryUpdatePartitionsAndLogging(topicName, partitionNum)
 				return partitionNum, nil
 			}
diff --git a/cdc/sink/mq/mq.go b/cdc/sink/mq/mq.go
index e7210e437..ef9f98352 100644
--- a/cdc/sink/mq/mq.go
+++ b/cdc/sink/mq/mq.go
@@ -432,7 +432,7 @@ func NewKafkaSaramaSink(ctx context.Context, sinkURI *url.URL,
 		return nil, cerror.WrapError(cerror.ErrKafkaNewSaramaProducer, err)
 	}
 
-	topicManager := manager.NewKafkaTopicManager(ctx, adminClient, baseConfig.DeriveTopicConfig())
+	topicManager := manager.NewKafkaTopicManager(ctx, topic, adminClient, baseConfig.DeriveTopicConfig())
 	if _, err := topicManager.CreateTopicAndWaitUntilVisible(topic); err != nil {
 		return nil, cerror.WrapError(cerror.ErrKafkaCreateTopic, err)
 	}
diff --git a/cdc/sinkv2/eventsink/cloudstorage/cloud_storage_dml_sink.go b/cdc/sinkv2/eventsink/cloudstorage/cloud_storage_dml_sink.go
index 081098b6a..e86ccf2a0 100644
--- a/cdc/sinkv2/eventsink/cloudstorage/cloud_storage_dml_sink.go
+++ b/cdc/sinkv2/eventsink/cloudstorage/cloud_storage_dml_sink.go
@@ -80,7 +80,7 @@ type dmlSink struct {
 	alive struct {
 		sync.RWMutex
 		// msgCh is a channel to hold eventFragment.
-		msgCh  chan eventFragment
+		msgCh  *chann.DrainableChann[eventFragment]
 		isDead bool
 	}
 
@@ -140,7 +140,7 @@ func NewCloudStorageSink(
 		cancel:          wgCancel,
 		dead:            make(chan struct{}),
 	}
-	s.alive.msgCh = make(chan eventFragment, defaultChannelSize)
+	s.alive.msgCh = chann.NewDrainableChann[eventFragment]()
 
 	encodedCh := make(chan eventFragment, defaultChannelSize)
 	workerChannels := make([]*chann.DrainableChann[eventFragment], cfg.WorkerCount)
@@ -148,7 +148,7 @@ func NewCloudStorageSink(
 	// create a group of encoding workers.
 	for i := 0; i < defaultEncodingConcurrency; i++ {
 		encoder := encoderBuilder.Build()
-		s.encodingWorkers[i] = newEncodingWorker(i, s.changefeedID, encoder, s.alive.msgCh, encodedCh)
+		s.encodingWorkers[i] = newEncodingWorker(i, s.changefeedID, encoder, s.alive.msgCh.Out(), encodedCh)
 	}
 	// create defragmenter.
 	s.defragmenter = newDefragmenter(encodedCh, workerChannels)
@@ -168,7 +168,7 @@ func NewCloudStorageSink(
 
 		s.alive.Lock()
 		s.alive.isDead = true
-		close(s.alive.msgCh)
+		s.alive.msgCh.CloseAndDrain()
 		s.alive.Unlock()
 		close(s.dead)
 
@@ -234,7 +234,7 @@ func (s *dmlSink) WriteEvents(txns ...*eventsink.CallbackableEvent[*model.Single
 
 		s.statistics.ObserveRows(txn.Event.Rows...)
 		// emit a TxnCallbackableEvent encoupled with a sequence number starting from one.
-		s.alive.msgCh <- eventFragment{
+		s.alive.msgCh.In() <- eventFragment{
 			seqNumber:      seq,
 			versionedTable: tbl,
 			event:          txn,
diff --git a/cdc/sinkv2/util/helper.go b/cdc/sinkv2/util/helper.go
index aa49156a9..818e4681d 100644
--- a/cdc/sinkv2/util/helper.go
+++ b/cdc/sinkv2/util/helper.go
@@ -94,7 +94,7 @@ func GetTopicManagerAndTryCreateTopic(
 	topicCfg *kafka.AutoCreateTopicConfig,
 	adminClient pkafka.ClusterAdminClient,
 ) (manager.TopicManager, error) {
-	topicManager := manager.NewKafkaTopicManager(ctx, adminClient, topicCfg)
+	topicManager := manager.NewKafkaTopicManager(ctx, topic, adminClient, topicCfg)
 	if _, err := topicManager.CreateTopicAndWaitUntilVisible(topic); err != nil {
 		return nil, cerror.WrapError(cerror.ErrKafkaCreateTopic, err)
 	}
diff --git a/cmd/kafka-consumer/main.go b/cmd/kafka-consumer/main.go
index 485829057..bee9095f5 100644
--- a/cmd/kafka-consumer/main.go
+++ b/cmd/kafka-consumer/main.go
@@ -686,10 +686,11 @@ func (c *Consumer) appendDDL(ddl *model.DDLEvent) {
 	defer c.ddlListMu.Unlock()
 	// DDL CommitTs fallback, just crash it to indicate the bug.
 	if c.ddlWithMaxCommitTs != nil && ddl.CommitTs < c.ddlWithMaxCommitTs.CommitTs {
-		log.Panic("DDL CommitTs < maxCommitTsDDL.CommitTs",
+		log.Warn("DDL CommitTs < maxCommitTsDDL.CommitTs",
 			zap.Uint64("commitTs", ddl.CommitTs),
 			zap.Uint64("maxCommitTs", c.ddlWithMaxCommitTs.CommitTs),
 			zap.Any("DDL", ddl))
+		return
 	}
 
 	// A rename tables DDL job contains multiple DDL events with same CommitTs.
diff --git a/dm/pkg/binlog/event/common.go b/dm/pkg/binlog/event/common.go
index e84dc8400..5437ec4f7 100644
--- a/dm/pkg/binlog/event/common.go
+++ b/dm/pkg/binlog/event/common.go
@@ -19,6 +19,7 @@ import (
 
 	gmysql "github.com/go-mysql-org/go-mysql/mysql"
 	"github.com/go-mysql-org/go-mysql/replication"
+	"github.com/pingcap/tiflow/dm/pkg/gtid"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 )
 
@@ -172,7 +173,7 @@ func GTIDIncrease(flavor string, gSet gmysql.GTIDSet) (gmysql.GTIDSet, error) {
 
 // verifySingleGTID verifies gSet whether only containing a single valid GTID.
 func verifySingleGTID(flavor string, gSet gmysql.GTIDSet) (interface{}, error) {
-	if gSet == nil || len(gSet.String()) == 0 {
+	if gtid.IsZeroGTIDSet(gSet) {
 		return nil, terror.ErrBinlogEmptyGTID.Generate()
 	}
 
diff --git a/dm/pkg/binlog/event/event.go b/dm/pkg/binlog/event/event.go
index daab72ff7..1d8ca835d 100644
--- a/dm/pkg/binlog/event/event.go
+++ b/dm/pkg/binlog/event/event.go
@@ -22,6 +22,7 @@ import (
 
 	gmysql "github.com/go-mysql-org/go-mysql/mysql"
 	"github.com/go-mysql-org/go-mysql/replication"
+	"github.com/pingcap/tiflow/dm/pkg/gtid"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 )
 
@@ -717,7 +718,7 @@ func GenXIDEvent(header *replication.EventHeader, latestPos uint32, xid uint64)
 // GenMariaDBGTIDListEvent generates a MariadbGTIDListEvent.
 // ref: https://mariadb.com/kb/en/library/gtid_list_event/
 func GenMariaDBGTIDListEvent(header *replication.EventHeader, latestPos uint32, gSet gmysql.GTIDSet) (*replication.BinlogEvent, error) {
-	if gSet == nil || len(gSet.String()) == 0 {
+	if gtid.IsZeroGTIDSet(gSet) {
 		return nil, terror.ErrBinlogEmptyGTID.Generate()
 	}
 
diff --git a/dm/pkg/binlog/position.go b/dm/pkg/binlog/position.go
index 21eae14c3..a54fd70bb 100644
--- a/dm/pkg/binlog/position.go
+++ b/dm/pkg/binlog/position.go
@@ -366,8 +366,8 @@ func IsFreshPosition(location Location, flavor string, cmpGTID bool) bool {
 //
 // but if can't compare gSet1 and gSet2, will returns 0, false.
 func CompareGTID(gSet1, gSet2 gmysql.GTIDSet) (int, bool) {
-	gSetIsEmpty1 := gSet1 == nil || len(gSet1.String()) == 0
-	gSetIsEmpty2 := gSet2 == nil || len(gSet2.String()) == 0
+	gSetIsEmpty1 := gtid.IsZeroGTIDSet(gSet1)
+	gSetIsEmpty2 := gtid.IsZeroGTIDSet(gSet2)
 
 	switch {
 	case gSetIsEmpty1 && gSetIsEmpty2:
diff --git a/dm/pkg/gtid/gtid.go b/dm/pkg/gtid/gtid.go
index 98075a52d..d81734519 100644
--- a/dm/pkg/gtid/gtid.go
+++ b/dm/pkg/gtid/gtid.go
@@ -21,6 +21,11 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 )
 
+var (
+	zeroMySQLGTIDSet   = MustZeroGTIDSet(mysql.MySQLFlavor)
+	zeroMariaDBGTIDSet = MustZeroGTIDSet(mysql.MariaDBFlavor)
+)
+
 // ParserGTID parses GTID from string. If the flavor is not specified, it will
 // try mysql GTID first and then MariaDB GTID.
 func ParserGTID(flavor, gtidStr string) (mysql.GTIDSet, error) {
@@ -42,7 +47,11 @@ func ParserGTID(flavor, gtidStr string) (mysql.GTIDSet, error) {
 			gtid, err = mysql.ParseGTIDSet(fla, gtidStr)
 		}
 	case mysql.MariaDBFlavor:
-		gtid, err = mysql.ParseGTIDSet(fla, gtidStr)
+		if IsZeroMariaDBGTIDSet(gtidStr) {
+			gtid, err = mysql.ParseGTIDSet(fla, "")
+		} else {
+			gtid, err = mysql.ParseGTIDSet(fla, gtidStr)
+		}
 	case "":
 		fla = mysql.MySQLFlavor
 		gtid, err = mysql.ParseGTIDSet(fla, gtidStr)
@@ -74,9 +83,6 @@ func MustZeroGTIDSet(flavor string) mysql.GTIDSet {
 // IsZeroMySQLGTIDSet is used to meet this usage: when user wants to start binlog
 // replication from scratch, a "uuid:0" (MySQL flavor) or "0-0-0" (mariaDB) GTID
 // set must be written, in order to distinguish that user forgets to write it.
-//
-// For above two flavor, only "uuid:0" is illegal, so we use IsZeroMySQLGTIDSet
-// to handle it.
 func IsZeroMySQLGTIDSet(gStr string) bool {
 	sp := strings.Split(gStr, ",")
 	if len(sp) != 1 {
@@ -90,3 +96,28 @@ func IsZeroMySQLGTIDSet(gStr string) bool {
 	interval := strings.TrimSpace(sep[1])
 	return interval == "0"
 }
+
+// IsZeroMariaDBGTIDSet is used to meet this usage: when user wants to start binlog
+// replication from scratch, a "uuid:0" (MySQL flavor) or "0-0-0" (mariaDB) GTID
+// set must be written, in order to distinguish that user forgets to write it.
+//
+// For MariaDB, the GTID set like "0-0-0" will confuse IsZeroGTIDSet function,
+// so we also need to check the interval part.
+func IsZeroMariaDBGTIDSet(gStr string) bool {
+	sp := strings.Split(gStr, ",")
+	if len(sp) != 1 {
+		return false
+	}
+
+	sep := strings.Split(sp[0], "-")
+	if len(sep) != 3 {
+		return false
+	}
+	interval := strings.TrimSpace(sep[2])
+	return interval == "0"
+}
+
+// IsZeroGTIDSet is used to check whether a GTID set is zero.
+func IsZeroGTIDSet(gSet mysql.GTIDSet) bool {
+	return gSet == nil || zeroMySQLGTIDSet.Equal(gSet) || zeroMariaDBGTIDSet.Equal(gSet)
+}
diff --git a/dm/pkg/shardddl/optimism/lock.go b/dm/pkg/shardddl/optimism/lock.go
index 2e148083e..06f3df341 100644
--- a/dm/pkg/shardddl/optimism/lock.go
+++ b/dm/pkg/shardddl/optimism/lock.go
@@ -848,7 +848,7 @@ func (l *Lock) trySyncForOneDDL(source, schema, table string, prevTable, postTab
 
 	// Normal DDL
 	if tableErr == nil {
-		log.L().Debug("receive a normal DDL", zap.String("source", source), zap.String("schema", schema), zap.String("table", table), zap.Stringer("prevTable", prevTable), zap.Stringer("postTable", postTable))
+		log.L().Info("receive a normal DDL", zap.String("source", source), zap.String("schema", schema), zap.String("table", table), zap.Stringer("prevTable", prevTable), zap.Stringer("postTable", postTable))
 		oldJoined, oldErr := l.joinNormalTables()
 
 		l.tables[source][schema][table] = postTable
@@ -883,7 +883,8 @@ func (l *Lock) trySyncForOneDDL(source, schema, table string, prevTable, postTab
 			// oldJoined != newJoined
 			// postTable == oldJoined (CREATE TABLE)
 			// prevTable < postTable
-			return (joinedErr != nil || joinedCmp != 0) || (err2 == nil && cmp == 0) || tableCmp < 0, ConflictNone
+			// prevTable == postTable(Partition/Sequence)
+			return (joinedErr != nil || joinedCmp != 0) || (err2 == nil && cmp == 0) || tableCmp <= 0, ConflictNone
 		}
 	}
 
@@ -920,7 +921,7 @@ func (l *Lock) trySyncForOneDDL(source, schema, table string, prevTable, postTab
 
 		return true, ConflictNone
 	}
-	log.L().Debug("conflict hasn't been resolved", zap.String("source", source), zap.String("schema", schema), zap.String("table", table), zap.Stringer("prevTable", prevTable), zap.Stringer("postTable", postTable))
+	log.L().Info("conflict hasn't been resolved", zap.String("source", source), zap.String("schema", schema), zap.String("table", table), zap.Stringer("prevTable", prevTable), zap.Stringer("postTable", postTable))
 	return false, ConflictSkipWaitRedirect
 }
 
diff --git a/dm/syncer/opt_sharding_group.go b/dm/syncer/opt_sharding_group.go
index d87514371..ce4060de6 100644
--- a/dm/syncer/opt_sharding_group.go
+++ b/dm/syncer/opt_sharding_group.go
@@ -233,3 +233,11 @@ func (k *OptShardingGroupKeeper) RemoveSchema(schema string) {
 		}
 	}
 }
+
+// Reset resets the keeper.
+func (k *OptShardingGroupKeeper) Reset() {
+	k.Lock()
+	defer k.Unlock()
+	k.groups = make(map[string]*OptShardingGroup)
+	k.shardingReSyncs = make(map[string]binlog.Location)
+}
diff --git a/dm/syncer/syncer.go b/dm/syncer/syncer.go
index dddbcd323..9fc3ef33b 100644
--- a/dm/syncer/syncer.go
+++ b/dm/syncer/syncer.go
@@ -223,9 +223,23 @@ type Syncer struct {
 	// `lower_case_table_names` setting of upstream db
 	SourceTableNamesFlavor utils.LowerCaseTableNamesFlavor
 
-	tsOffset                  atomic.Int64    // time offset between upstream and syncer, DM's timestamp - MySQL's timestamp
-	secondsBehindMaster       atomic.Int64    // current task delay second behind upstream
-	workerJobTSArray          []*atomic.Int64 // worker's sync job TS array, note that idx=0 is skip idx and idx=1 is ddl idx,sql worker job idx=(queue id + 2)
+	// time difference between upstream and DM nodes: time of DM - time of upstream.
+	// we use this to calculate replication lag more accurately when clock is not synced
+	// on either upstream or DM nodes.
+	tsOffset atomic.Int64
+	// this field measures the time difference in seconds between current time of DM and
+	// the minimal timestamp of currently processing binlog events.
+	// this lag will consider time difference between upstream and DM nodes
+	secondsBehindMaster atomic.Int64
+	// stores the last job TS(binlog event timestamp) of each worker,
+	// if there's no active job, the corresponding worker's TS is reset to 0.
+	// since DML worker runs jobs in batch, the TS is the TS of the first job in the batch.
+	// We account for skipped events too, if the distance between up and downstream is long,
+	// and there's no interested binlog event in between, we can have a decreasing lag.
+	// 	- 0 is for skip jobs
+	// 	- 1 is ddl worker
+	// 	- 2+ is for DML worker job idx=(queue id + 2)
+	workerJobTSArray          []*atomic.Int64
 	lastCheckpointFlushedTime time.Time
 
 	firstMeetBinlogTS *int64
@@ -615,6 +629,10 @@ func (s *Syncer) reset() {
 	if s.streamerController != nil {
 		s.streamerController.Close()
 	}
+	s.secondsBehindMaster.Store(0)
+	for _, jobTS := range s.workerJobTSArray {
+		jobTS.Store(0)
+	}
 	// create new job chans
 	s.newJobChans()
 	s.checkpoint.DiscardPendingSnapshots()
@@ -639,6 +657,7 @@ func (s *Syncer) reset() {
 		s.sgk.ResetGroups()
 		s.pessimist.Reset()
 	case config.ShardOptimistic:
+		s.osgk.Reset()
 		s.optimist.Reset()
 	}
 }
@@ -878,7 +897,7 @@ func (s *Syncer) calcReplicationLag(headerTS int64) int64 {
 
 // updateReplicationJobTS store job TS, it is called after every batch dml job / one skip job / one ddl job is added and committed.
 func (s *Syncer) updateReplicationJobTS(job *job, jobIdx int) {
-	// when job is nil mean no job in this bucket, need do reset this bucket job ts to 0
+	// when job is nil mean no job in this bucket, need to reset this bucket job ts to 0
 	if job == nil {
 		s.workerJobTSArray[jobIdx].Store(0)
 	} else {
@@ -2839,8 +2858,18 @@ func (s *Syncer) trackOriginDDL(ev *replication.QueryEvent, ec eventContext) (ma
 		return nil, err
 	}
 
-	affectedTbls := make(map[string]map[string]struct{})
 	for _, sql := range qec.splitDDLs {
+		sqls, err := s.ddlWorker.processOneDDL(qec, sql)
+		if err != nil {
+			s.tctx.L().Warn("processOneDDL failed", zap.Error(err))
+			qec.appliedDDLs = append(qec.appliedDDLs, sql)
+		} else {
+			qec.appliedDDLs = append(qec.appliedDDLs, sqls...)
+		}
+	}
+
+	affectedTbls := make(map[string]map[string]struct{})
+	for _, sql := range qec.appliedDDLs {
 		ddlInfo, err := s.ddlWorker.genDDLInfo(qec, sql)
 		if err != nil {
 			return nil, err
diff --git a/dm/worker/subtask.go b/dm/worker/subtask.go
index b0a178d6e..84fcc858a 100644
--- a/dm/worker/subtask.go
+++ b/dm/worker/subtask.go
@@ -585,7 +585,7 @@ func (st *SubTask) Kill() {
 	st.validator = nil
 }
 
-// Pause pauses a running sub task or a sub task paused by error.
+// Pause pauses a running subtask or a subtask paused by error.
 func (st *SubTask) Pause() error {
 	if st.markResultCanceled() {
 		return nil
diff --git a/metrics/grafana/ticdc.json b/metrics/grafana/ticdc.json
index b3bd37f35..53d07c6f1 100644
--- a/metrics/grafana/ticdc.json
+++ b/metrics/grafana/ticdc.json
@@ -3426,8 +3426,16 @@
               "hide": false,
               "interval": "",
               "intervalFactor": 1,
-              "legendFormat": "{{changefeed}}",
+              "legendFormat": "{{changefeed}}-barrier",
               "refId": "C"
+            },
+            {
+              "exemplar": true,
+              "expr": "max(ticdc_scheduler_slow_table_puller_resolved_ts{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", changefeed=~\"$changefeed\"}) by (changefeed)",
+              "hide": false,
+              "interval": "",
+              "legendFormat": "{{changefeed}}-puller",
+              "refId": "B"
             }
           ],
           "thresholds": [],
@@ -3632,8 +3640,16 @@
               "format": "time_series",
               "interval": "",
               "intervalFactor": 1,
-              "legendFormat": "{{changefeed}}",
+              "legendFormat": "{{changefeed}}-barrier",
               "refId": "C"
+            },
+            {
+              "exemplar": true,
+              "expr": "max(ticdc_scheduler_slow_table_puller_resolved_ts_lag{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", changefeed=~\"$changefeed\"}) by (changefeed)",
+              "hide": false,
+              "interval": "",
+              "legendFormat": "{{changefeed}}-puller",
+              "refId": "A"
             }
           ],
           "thresholds": [],
diff --git a/pkg/config/replica_config.go b/pkg/config/replica_config.go
index 6dde6c2df..43c5aa9ef 100644
--- a/pkg/config/replica_config.go
+++ b/pkg/config/replica_config.go
@@ -33,12 +33,13 @@ const (
 	// minSyncPointInterval is the minimum of SyncPointInterval can be set.
 	minSyncPointInterval = time.Second * 30
 	// minSyncPointRetention is the minimum of SyncPointRetention can be set.
-	minSyncPointRetention = time.Hour * 1
+	minSyncPointRetention           = time.Hour * 1
+	minChangeFeedErrorStuckDuration = time.Minute * 30
 )
 
 var defaultReplicaConfig = &ReplicaConfig{
 	MemoryQuota:        DefaultChangefeedMemoryQuota,
-	CaseSensitive:      true,
+	CaseSensitive:      false,
 	EnableOldValue:     true,
 	CheckGCSafePoint:   true,
 	EnableSyncPoint:    false,
@@ -72,6 +73,7 @@ var defaultReplicaConfig = &ReplicaConfig{
 		Storage:               "",
 		UseFileBackend:        false,
 	},
+	ChangefeedErrorStuckDuration: time.Minute * 30,
 }
 
 // GetDefaultReplicaConfig returns the default replica config.
@@ -104,13 +106,14 @@ type replicaConfig struct {
 	// BDR(Bidirectional Replication) is a feature that allows users to
 	// replicate data of same tables from TiDB-1 to TiDB-2 and vice versa.
 	// This feature is only available for TiDB.
-	BDRMode            bool              `toml:"bdr-mode" json:"bdr-mode"`
-	SyncPointInterval  time.Duration     `toml:"sync-point-interval" json:"sync-point-interval"`
-	SyncPointRetention time.Duration     `toml:"sync-point-retention" json:"sync-point-retention"`
-	Filter             *FilterConfig     `toml:"filter" json:"filter"`
-	Mounter            *MounterConfig    `toml:"mounter" json:"mounter"`
-	Sink               *SinkConfig       `toml:"sink" json:"sink"`
-	Consistent         *ConsistentConfig `toml:"consistent" json:"consistent"`
+	BDRMode                      bool              `toml:"bdr-mode" json:"bdr-mode"`
+	SyncPointInterval            time.Duration     `toml:"sync-point-interval" json:"sync-point-interval"`
+	SyncPointRetention           time.Duration     `toml:"sync-point-retention" json:"sync-point-retention"`
+	Filter                       *FilterConfig     `toml:"filter" json:"filter"`
+	Mounter                      *MounterConfig    `toml:"mounter" json:"mounter"`
+	Sink                         *SinkConfig       `toml:"sink" json:"sink"`
+	Consistent                   *ConsistentConfig `toml:"consistent" json:"consistent"`
+	ChangefeedErrorStuckDuration time.Duration     `toml:"changefeed-error-stuck-duration" json:"changefeed-error-stuck-duration,omitempty"`
 }
 
 // Marshal returns the json marshal format of a ReplicationConfig
@@ -210,6 +213,14 @@ func (c *ReplicaConfig) ValidateAndAdjust(sinkURI *url.URL) error { // check sin
 		c.FixMemoryQuota()
 	}
 
+	if c.ChangefeedErrorStuckDuration < minChangeFeedErrorStuckDuration {
+		return cerror.ErrInvalidReplicaConfig.
+			FastGenByArgs(
+				fmt.Sprintf("The ChangefeedErrorStuckDuration:%f must be larger than %f Seconds",
+					c.ChangefeedErrorStuckDuration.Seconds(),
+					minChangeFeedErrorStuckDuration.Seconds()))
+	}
+
 	return nil
 }
 
diff --git a/pkg/config/server_config.go b/pkg/config/server_config.go
index fc7ed53a8..7c3c6ff72 100644
--- a/pkg/config/server_config.go
+++ b/pkg/config/server_config.go
@@ -50,9 +50,9 @@ const (
 	// DefaultChangefeedMemoryQuota is the default memory quota for each changefeed.
 	DefaultChangefeedMemoryQuota = 1024 * 1024 * 1024 // 1GB.
 
-	// DefaultMaxMemoryPercentage is the default max memory percentage
-	// cdc server use 70% of total memory limit as soft limit by default.
-	DefaultMaxMemoryPercentage = 70
+	// DisableMemoryLimit is the default max memory percentage for TiCDC server.
+	// 0 means no memory limit.
+	DisableMemoryLimit = 0
 )
 
 var (
@@ -160,8 +160,8 @@ var defaultServerConfig = &ServerConfig{
 		EnablePullBasedSink:    true,
 		EnableKVConnectBackOff: false,
 	},
-	ClusterID:           "default",
-	MaxMemoryPercentage: DefaultMaxMemoryPercentage,
+	ClusterID:              "default",
+	GcTunerMemoryThreshold: DisableMemoryLimit,
 }
 
 // ServerConfig represents a config for server
@@ -189,7 +189,9 @@ type ServerConfig struct {
 	KVClient            *KVClientConfig `toml:"kv-client" json:"kv-client"`
 	Debug               *DebugConfig    `toml:"debug" json:"debug"`
 	ClusterID           string          `toml:"cluster-id" json:"cluster-id"`
-	MaxMemoryPercentage int             `toml:"max-memory-percentage" json:"max-memory-percentage"`
+	// Deprecated: we don't use this field anymore.
+	MaxMemoryPercentage    int    `toml:"max-memory-percentage" json:"max-memory-percentage"`
+	GcTunerMemoryThreshold uint64 `toml:"gc-tuner-memory-threshold" json:"gc-tuner-memory-threshold"`
 }
 
 // Marshal returns the json marshal format of a ServerConfig
@@ -304,11 +306,6 @@ func (c *ServerConfig) ValidateAndAdjust() error {
 	if err = c.Debug.ValidateAndAdjust(); err != nil {
 		return errors.Trace(err)
 	}
-	if c.MaxMemoryPercentage >= 100 {
-		log.Warn("server max-memory-percentage must be less than 100, set to default value")
-		c.MaxMemoryPercentage = DefaultMaxMemoryPercentage
-	}
-
 	return nil
 }
 
diff --git a/pkg/filter/filter.go b/pkg/filter/filter.go
index 3b321f163..f2d3cb6cb 100644
--- a/pkg/filter/filter.go
+++ b/pkg/filter/filter.go
@@ -155,13 +155,12 @@ func (f *filter) ShouldDiscardDDL(startTs uint64, ddlType timodel.ActionType, sc
 		return
 	}
 
-	switch ddlType {
-	case timodel.ActionCreateSchema, timodel.ActionDropSchema,
-		timodel.ActionModifySchemaCharsetAndCollate:
+	if IsSchemaDDL(ddlType) {
 		discard = !f.tableFilter.MatchSchema(schema)
-	default:
+	} else {
 		discard = f.ShouldIgnoreTable(schema, table)
 	}
+
 	if discard {
 		return
 	}
@@ -204,3 +203,14 @@ func isAllowedDDL(actionType timodel.ActionType) bool {
 	}
 	return false
 }
+
+// IsSchemaDDL returns true if the action type is a schema DDL.
+func IsSchemaDDL(actionType timodel.ActionType) bool {
+	switch actionType {
+	case timodel.ActionCreateSchema, timodel.ActionDropSchema,
+		timodel.ActionModifySchemaCharsetAndCollate:
+		return true
+	default:
+		return false
+	}
+}
diff --git a/pkg/redo/config.go b/pkg/redo/config.go
index d0562b83b..d6b66de45 100644
--- a/pkg/redo/config.go
+++ b/pkg/redo/config.go
@@ -174,7 +174,7 @@ func FixLocalScheme(uri *url.URL) {
 
 // IsBlackholeStorage returns whether a blackhole storage is used.
 func IsBlackholeStorage(scheme string) bool {
-	return ConsistentStorage(scheme) == consistentStorageBlackhole
+	return strings.HasPrefix(scheme, string(consistentStorageBlackhole))
 }
 
 // InitExternalStorage init an external storage.
@@ -222,6 +222,18 @@ func ValidateStorage(uri *url.URL) error {
 		return errors.WrapError(errors.ErrStorageInitialize, errors.Annotate(err,
 			fmt.Sprintf("can't make dir for new redo log: %+v", uri)))
 	}
+
+	file := filepath.Join(uri.Path, "file.test")
+	if err := os.WriteFile(file, []byte(""), DefaultFileMode); err != nil {
+		return errors.WrapError(errors.ErrStorageInitialize, errors.Annotate(err,
+			fmt.Sprintf("can't write file for new redo log: %+v", uri)))
+	}
+
+	if _, err := os.ReadFile(file); err != nil {
+		return errors.WrapError(errors.ErrStorageInitialize, errors.Annotate(err,
+			fmt.Sprintf("can't read file for new redo log: %+v", uri)))
+	}
+	_ = os.Remove(file)
 	return nil
 }
 
diff --git a/pkg/sink/cloudstorage/path.go b/pkg/sink/cloudstorage/path.go
index 448a6217d..ffed62d0d 100644
--- a/pkg/sink/cloudstorage/path.go
+++ b/pkg/sink/cloudstorage/path.go
@@ -192,27 +192,28 @@ func (f *FilePathGenerator) CheckOrWriteSchema(
 	_, checksum := mustParseSchemaName(tblSchemaFile)
 	schemaFileCnt := 0
 	lastVersion := uint64(0)
-	prefix := fmt.Sprintf(tableSchemaPrefix+"schema_", def.Schema, def.Table)
+	subDir := fmt.Sprintf(tableSchemaPrefix, def.Schema, def.Table)
 	checksumSuffix := fmt.Sprintf("%010d.json", checksum)
-	err = f.storage.WalkDir(ctx, &storage.WalkOption{ObjPrefix: prefix},
-		func(path string, _ int64) error {
-			schemaFileCnt++
-			if !strings.HasSuffix(path, checksumSuffix) {
-				return nil
-			}
-			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",
-					zap.String("path", path), zap.Any("checksum", checksum))
-			}
-			if version > lastVersion {
-				lastVersion = version
-			}
+	err = f.storage.WalkDir(ctx, &storage.WalkOption{
+		SubDir:    subDir, /* use subDir to prevent walk the whole storage */
+		ObjPrefix: subDir + "schema_",
+	}, func(path string, _ int64) error {
+		schemaFileCnt++
+		if !strings.HasSuffix(path, checksumSuffix) {
 			return nil
-		},
-	)
+		}
+		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",
+				zap.String("path", path), zap.Any("checksum", checksum))
+		}
+		if version > lastVersion {
+			lastVersion = version
+		}
+		return nil
+	})
 	if err != nil {
 		return err
 	}
diff --git a/pkg/upstream/manager.go b/pkg/upstream/manager.go
index d9c69077f..b12f02265 100644
--- a/pkg/upstream/manager.go
+++ b/pkg/upstream/manager.go
@@ -197,7 +197,9 @@ func (m *Manager) Tick(ctx context.Context,
 
 	activeUpstreams := make(map[uint64]struct{})
 	for _, cf := range globalState.Changefeeds {
-		activeUpstreams[cf.Info.UpstreamID] = struct{}{}
+		if cf != nil && cf.Info != nil {
+			activeUpstreams[cf.Info.UpstreamID] = struct{}{}
+		}
 	}
 	m.mu.Lock()
 	defer m.mu.Unlock()


Test patch:
diff --git a/cdc/api/v2/changefeed_test.go b/cdc/api/v2/changefeed_test.go
index 2c7a01179..2f526556f 100644
--- a/cdc/api/v2/changefeed_test.go
+++ b/cdc/api/v2/changefeed_test.go
@@ -311,14 +311,15 @@ func TestUpdateChangefeed(t *testing.T) {
 	t.Parallel()
 	update := testCase{url: "/api/v2/changefeeds/%s", method: "PUT"}
 	helpers := NewMockAPIV2Helpers(gomock.NewController(t))
-	cp := mock_capture.NewMockCapture(gomock.NewController(t))
-	apiV2 := NewOpenAPIV2ForTest(cp, helpers)
+	mockCapture := mock_capture.NewMockCapture(gomock.NewController(t))
+	apiV2 := NewOpenAPIV2ForTest(mockCapture, helpers)
 	router := newRouter(apiV2)
 
 	statusProvider := &mockStatusProvider{}
-	cp.EXPECT().StatusProvider().Return(statusProvider).AnyTimes()
-	cp.EXPECT().IsReady().Return(true).AnyTimes()
-	cp.EXPECT().IsOwner().Return(true).AnyTimes()
+
+	mockCapture.EXPECT().StatusProvider().Return(statusProvider).AnyTimes()
+	mockCapture.EXPECT().IsReady().Return(true).AnyTimes()
+	mockCapture.EXPECT().IsOwner().Return(true).AnyTimes()
 
 	// case 1 invalid id
 	invalidID := "#Invalid_"
@@ -372,7 +373,7 @@ func TestUpdateChangefeed(t *testing.T) {
 	etcdClient.EXPECT().
 		GetUpstreamInfo(gomock.Any(), gomock.Eq(uint64(100)), gomock.Any()).
 		Return(nil, cerrors.ErrUpstreamNotFound).Times(1)
-	cp.EXPECT().GetEtcdClient().Return(etcdClient).AnyTimes()
+	mockCapture.EXPECT().GetEtcdClient().Return(etcdClient).AnyTimes()
 
 	w = httptest.NewRecorder()
 	req, _ = http.NewRequestWithContext(context.Background(), update.method,
@@ -389,7 +390,7 @@ func TestUpdateChangefeed(t *testing.T) {
 	etcdClient.EXPECT().
 		GetUpstreamInfo(gomock.Any(), gomock.Eq(uint64(1)), gomock.Any()).
 		Return(nil, nil).AnyTimes()
-	cp.EXPECT().GetEtcdClient().Return(etcdClient).AnyTimes()
+	mockCapture.EXPECT().GetEtcdClient().Return(etcdClient).AnyTimes()
 
 	w = httptest.NewRecorder()
 	req, _ = http.NewRequestWithContext(context.Background(), update.method,
@@ -426,6 +427,7 @@ func TestUpdateChangefeed(t *testing.T) {
 		createTiStore(gomock.Any(), gomock.Any()).
 		Return(nil, nil).
 		AnyTimes()
+	mockCapture.EXPECT().GetUpstreamManager().Return(nil, nil).AnyTimes()
 	helpers.EXPECT().
 		verifyUpdateChangefeedConfig(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(&model.ChangeFeedInfo{}, &model.UpstreamInfo{}, cerrors.ErrChangefeedUpdateRefused).
@@ -445,6 +447,7 @@ func TestUpdateChangefeed(t *testing.T) {
 	require.Equal(t, http.StatusBadRequest, w.Code)
 
 	// case 7: update transaction failed
+	mockCapture.EXPECT().GetUpstreamManager().Return(upstream.NewManager4Test(&mockPDClient{}), nil).AnyTimes()
 	helpers.EXPECT().
 		verifyUpdateChangefeedConfig(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(&model.ChangeFeedInfo{}, &model.UpstreamInfo{}, nil).
@@ -468,6 +471,7 @@ func TestUpdateChangefeed(t *testing.T) {
 		verifyUpdateChangefeedConfig(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(oldCfInfo, &model.UpstreamInfo{}, nil).
 		Times(1)
+	mockCapture.EXPECT().GetUpstreamManager().Return(upstream.NewManager4Test(&mockPDClient{}), nil).AnyTimes()
 	etcdClient.EXPECT().
 		UpdateChangefeedAndUpstream(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil).Times(1)
@@ -484,6 +488,7 @@ func TestUpdateChangefeed(t *testing.T) {
 		verifyUpdateChangefeedConfig(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(oldCfInfo, &model.UpstreamInfo{}, nil).
 		Times(1)
+	mockCapture.EXPECT().GetUpstreamManager().Return(upstream.NewManager4Test(&mockPDClient{}), nil).AnyTimes()
 	etcdClient.EXPECT().
 		UpdateChangefeedAndUpstream(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil).Times(1)
diff --git a/cdc/api/v2/model_test.go b/cdc/api/v2/model_test.go
index f04fa1386..e562574af 100644
--- a/cdc/api/v2/model_test.go
+++ b/cdc/api/v2/model_test.go
@@ -30,7 +30,7 @@ import (
 // note: this is api published default value, not change it
 var defaultAPIConfig = &ReplicaConfig{
 	MemoryQuota:        config.DefaultChangefeedMemoryQuota,
-	CaseSensitive:      true,
+	CaseSensitive:      false,
 	EnableOldValue:     true,
 	CheckGCSafePoint:   true,
 	EnableSyncPoint:    false,
@@ -63,6 +63,8 @@ var defaultAPIConfig = &ReplicaConfig{
 		Storage:               "",
 		UseFileBackend:        false,
 	},
+	ChangefeedErrorStuckDuration: &JSONDuration{config.
+		GetDefaultReplicaConfig().ChangefeedErrorStuckDuration},
 }
 
 func TestDefaultReplicaConfig(t *testing.T) {
diff --git a/cdc/entry/mounter_test.go b/cdc/entry/mounter_test.go
index 3bf0f9587..d97627520 100644
--- a/cdc/entry/mounter_test.go
+++ b/cdc/entry/mounter_test.go
@@ -1197,9 +1197,41 @@ func TestBuildTableInfo(t *testing.T) {
 				"  PRIMARY KEY (`a`(0)) /*T![clustered_index] CLUSTERED */\n" +
 				") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin",
 		},
+		{ // This case is to check the primary key is correctly identified by BuildTiDBTableInfo
+			"CREATE TABLE your_table (" +
+				" id INT NOT NULL," +
+				" name VARCHAR(50) NOT NULL," +
+				" email VARCHAR(100) NOT NULL," +
+				" age INT NOT NULL ," +
+				" address VARCHAR(200) NOT NULL," +
+				" PRIMARY KEY (id, name)," +
+				" UNIQUE INDEX idx_unique_1 (id, email, age)," +
+				" UNIQUE INDEX idx_unique_2 (name, email, address)" +
+				" );",
+			"CREATE TABLE `BuildTiDBTableInfo` (\n" +
+				"  `id` int(0) NOT NULL,\n" +
+				"  `name` varchar(0) NOT NULL,\n" +
+				"  `email` varchar(0) NOT NULL,\n" +
+				"  `age` int(0) NOT NULL,\n" +
+				"  `address` varchar(0) NOT NULL,\n" +
+				"  PRIMARY KEY (`id`(0),`name`(0)) /*T![clustered_index] CLUSTERED */,\n" +
+				"  UNIQUE KEY `idx_1` (`id`(0),`email`(0),`age`(0)),\n" +
+				"  UNIQUE KEY `idx_2` (`name`(0),`email`(0),`address`(0))\n" +
+				") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin",
+			"CREATE TABLE `BuildTiDBTableInfo` (\n" +
+				"  `id` int(0) NOT NULL,\n" +
+				"  `name` varchar(0) NOT NULL,\n" +
+				"  `omitted` unspecified GENERATED ALWAYS AS (pass_generated_check) VIRTUAL,\n" +
+				"  `omitted` unspecified GENERATED ALWAYS AS (pass_generated_check) VIRTUAL,\n" +
+				"  `omitted` unspecified GENERATED ALWAYS AS (pass_generated_check) VIRTUAL,\n" +
+				"  PRIMARY KEY (`id`(0),`name`(0)) /*T![clustered_index] CLUSTERED */,\n" +
+				"  UNIQUE KEY `idx_1` (`id`(0),`omitted`(0),`omitted`(0)),\n" +
+				"  UNIQUE KEY `idx_2` (`name`(0),`omitted`(0),`omitted`(0))\n" +
+				") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin",
+		},
 	}
 	p := parser.New()
-	for _, c := range cases {
+	for i, c := range cases {
 		stmt, err := p.ParseOneStmt(c.origin, "", "")
 		require.NoError(t, err)
 		originTI, err := ddl.BuildTableInfoFromAST(stmt.(*ast.CreateTableStmt))
@@ -1211,7 +1243,18 @@ func TestBuildTableInfo(t *testing.T) {
 		handle := sqlmodel.GetWhereHandle(recoveredTI, recoveredTI)
 		require.NotNil(t, handle.UniqueNotNullIdx)
 		require.Equal(t, c.recovered, showCreateTable(t, recoveredTI))
-
+		// make sure BuildTiDBTableInfo indentify the correct primary key
+		if i == 5 {
+			inexes := recoveredTI.Indices
+			primaryCount := 0
+			for i := range inexes {
+				if inexes[i].Primary {
+					primaryCount++
+				}
+			}
+			require.Equal(t, 1, primaryCount)
+			require.Equal(t, 2, len(handle.UniqueNotNullIdx.Columns))
+		}
 		// mimic the columns are set to nil when old value feature is disabled
 		for i := range cols {
 			if !cols[i].Flag.IsHandleKey() {
diff --git a/cdc/model/changefeed_test.go b/cdc/model/changefeed_test.go
index c1dd632b3..a13c9890c 100644
--- a/cdc/model/changefeed_test.go
+++ b/cdc/model/changefeed_test.go
@@ -146,7 +146,7 @@ func TestVerifyAndComplete(t *testing.T) {
 		StartTs: 417257993615179777,
 		Config: &config.ReplicaConfig{
 			MemoryQuota:        1073741824,
-			CaseSensitive:      true,
+			CaseSensitive:      false,
 			EnableOldValue:     true,
 			CheckGCSafePoint:   true,
 			SyncPointInterval:  time.Minute * 10,
diff --git a/cdc/owner/changefeed_test.go b/cdc/owner/changefeed_test.go
index b257b1afb..18e8eedde 100644
--- a/cdc/owner/changefeed_test.go
+++ b/cdc/owner/changefeed_test.go
@@ -201,7 +201,7 @@ func createChangefeed4Test(ctx cdcContext.Context, t *testing.T,
 		info = ctx.ChangefeedVars().Info
 		return info, true, nil
 	})
-
+	tester.MustApplyPatches()
 	cf := newChangefeed4Test(ctx.ChangefeedVars().ID, state, up,
 		// new ddl puller
 		func(ctx context.Context,
diff --git a/cdc/owner/feed_state_manager_test.go b/cdc/owner/feed_state_manager_test.go
index 6710063f9..82824f36d 100644
--- a/cdc/owner/feed_state_manager_test.go
+++ b/cdc/owner/feed_state_manager_test.go
@@ -844,7 +844,7 @@ func TestHandleWarning(t *testing.T) {
 	tester.MustApplyPatches()
 	// mock the checkpointTs is not progressing for defaultBackoffMaxElapsedTime time
 	manager.checkpointTsAdvanced = manager.
-		checkpointTsAdvanced.Add(-(defaultBackoffMaxElapsedTime + 1))
+		checkpointTsAdvanced.Add(-(manager.changefeedErrorStuckDuration + 1))
 	// resolveTs = 202 > checkpointTs = 201
 	manager.Tick(state, 202)
 	// some patches will be generated when the manager.Tick is called
diff --git a/cdc/owner/schema_test.go b/cdc/owner/schema_test.go
index 893ae951e..dd94a352e 100644
--- a/cdc/owner/schema_test.go
+++ b/cdc/owner/schema_test.go
@@ -149,8 +149,31 @@ func TestIsIneligibleTableID(t *testing.T) {
 	tableIDT2 := job.BinlogInfo.TableInfo.ID
 
 	require.Nil(t, schema.HandleDDLJob(job))
-	require.False(t, schema.IsIneligibleTableID(tableIDT1))
-	require.True(t, schema.IsIneligibleTableID(tableIDT2))
+	ctx := context.Background()
+	ignore, err := schema.IsIneligibleTable(ctx, tableIDT1, job.BinlogInfo.FinishedTS)
+	require.Nil(t, err)
+	require.False(t, ignore)
+
+	ignore, err = schema.IsIneligibleTable(ctx, tableIDT2, job.BinlogInfo.FinishedTS)
+	require.Nil(t, err)
+	require.True(t, ignore)
+
+	// test we get the right snapshot to check ineligible table
+	job = helper.DDL2Job("create table test.t3(id int)")
+	tableIDT3 := job.BinlogInfo.TableInfo.ID
+	snapshotTsWithoutPK := job.BinlogInfo.FinishedTS
+	require.Nil(t, schema.HandleDDLJob(job))
+	job = helper.DDL2Job("alter table test.t3 add primary key(id)")
+	snapshotTsWithPK := job.BinlogInfo.FinishedTS
+	require.Nil(t, schema.HandleDDLJob(job))
+	// tableIDT3 is ineligible at snapshotTsWithoutPK
+	ignore, err = schema.IsIneligibleTable(ctx, tableIDT3, snapshotTsWithoutPK)
+	require.Nil(t, err)
+	require.True(t, ignore)
+	// tableIDT3 is eligible at snapshotTsWithPK
+	ignore, err = schema.IsIneligibleTable(ctx, tableIDT3, snapshotTsWithPK)
+	require.Nil(t, err)
+	require.False(t, ignore)
 }
 
 func compareEvents(t *testing.T, e1, e2 *model.DDLEvent) {
diff --git a/cdc/processor/pipeline/table_actor_test.go b/cdc/processor/pipeline/table_actor_test.go
index b763c8089..9c0fcf51c 100644
--- a/cdc/processor/pipeline/table_actor_test.go
+++ b/cdc/processor/pipeline/table_actor_test.go
@@ -101,7 +101,8 @@ func TestTableActorInterface(t *testing.T) {
 	require.Equal(t, model.Ts(5), table.ResolvedTs())
 	ctx, cancel := context.WithCancel(context.Background())
 	eg, egCtx := errgroup.WithContext(ctx)
-	table.redoDMLMgr, _ = redo.NewDMLManager(ctx, &config.ConsistentConfig{
+	changefeedID := model.DefaultChangeFeedID("test")
+	table.redoDMLMgr = redo.NewDMLManager(changefeedID, &config.ConsistentConfig{
 		Level:             string(redoCfg.ConsistentLevelEventual),
 		FlushIntervalInMs: redoCfg.MinFlushIntervalInMs,
 		Storage:           fmt.Sprintf("file://tmp/%s", t.TempDir()),
diff --git a/cdc/processor/sinkmanager/redo_cache_test.go b/cdc/processor/sinkmanager/redo_cache_test.go
index 98b60fe9f..ef4c1c255 100644
--- a/cdc/processor/sinkmanager/redo_cache_test.go
+++ b/cdc/processor/sinkmanager/redo_cache_test.go
@@ -46,14 +46,14 @@ func TestRedoEventCache(t *testing.T) {
 	// Try to pop [{0,1}, {0,4}], shoud fail. And the returned boundary should be {1,4}.
 	popRes = appender.pop(engine.Position{StartTs: 0, CommitTs: 1}, engine.Position{StartTs: 0, CommitTs: 4})
 	require.False(t, popRes.success)
-	require.Equal(t, uint64(1), popRes.boundary.StartTs)
-	require.Equal(t, uint64(4), popRes.boundary.CommitTs)
+	require.Equal(t, uint64(1), popRes.lowerBoundIfFail.StartTs)
+	require.Equal(t, uint64(4), popRes.lowerBoundIfFail.CommitTs)
 
 	// Try to pop [{0,2}, {0,4}], shoud fail. And the returned boundary should be {3,4}.
 	popRes = appender.pop(engine.Position{StartTs: 0, CommitTs: 1}, engine.Position{StartTs: 5, CommitTs: 6})
 	require.False(t, popRes.success)
-	require.Equal(t, uint64(3), popRes.boundary.StartTs)
-	require.Equal(t, uint64(4), popRes.boundary.CommitTs)
+	require.Equal(t, uint64(3), popRes.lowerBoundIfFail.StartTs)
+	require.Equal(t, uint64(4), popRes.lowerBoundIfFail.CommitTs)
 
 	// Try to pop [{3,4}, {3,4}], should success.
 	popRes = appender.pop(engine.Position{StartTs: 3, CommitTs: 4}, engine.Position{StartTs: 3, CommitTs: 4})
@@ -61,22 +61,22 @@ func TestRedoEventCache(t *testing.T) {
 	require.Equal(t, 2, len(popRes.events))
 	require.Equal(t, uint64(300), popRes.size)
 	require.Equal(t, 2, popRes.pushCount)
-	require.Equal(t, uint64(3), popRes.boundary.StartTs)
-	require.Equal(t, uint64(4), popRes.boundary.CommitTs)
+	require.Equal(t, uint64(3), popRes.upperBoundIfSuccess.StartTs)
+	require.Equal(t, uint64(4), popRes.upperBoundIfSuccess.CommitTs)
 
 	// Try to pop [{3,4}, {3,4}] again, shoud fail. And the returned boundary should be {4,4}.
 	popRes = appender.pop(engine.Position{StartTs: 3, CommitTs: 4}, engine.Position{StartTs: 3, CommitTs: 4})
 	require.False(t, popRes.success)
-	require.Equal(t, uint64(4), popRes.boundary.StartTs)
-	require.Equal(t, uint64(4), popRes.boundary.CommitTs)
+	require.Equal(t, uint64(4), popRes.lowerBoundIfFail.StartTs)
+	require.Equal(t, uint64(4), popRes.lowerBoundIfFail.CommitTs)
 
 	popRes = appender.pop(engine.Position{StartTs: 4, CommitTs: 4}, engine.Position{StartTs: 9, CommitTs: 10})
 	require.True(t, popRes.success)
 	require.Equal(t, 1, len(popRes.events))
 	require.Equal(t, uint64(300), popRes.size)
 	require.Equal(t, 1, popRes.pushCount)
-	require.Equal(t, uint64(5), popRes.boundary.StartTs)
-	require.Equal(t, uint64(6), popRes.boundary.CommitTs)
+	require.Equal(t, uint64(5), popRes.upperBoundIfSuccess.StartTs)
+	require.Equal(t, uint64(6), popRes.upperBoundIfSuccess.CommitTs)
 	require.Equal(t, 0, len(appender.events))
 	require.True(t, appender.broken)
 
@@ -85,3 +85,64 @@ func TestRedoEventCache(t *testing.T) {
 	require.Equal(t, uint64(0), appender.upperBound.StartTs)
 	require.Equal(t, uint64(0), appender.upperBound.CommitTs)
 }
+
+func TestRedoEventCacheAllPopBranches(t *testing.T) {
+	cache := newRedoEventCache(model.ChangeFeedID{}, 1000)
+	appender := cache.maybeCreateAppender(1, engine.Position{StartTs: 101, CommitTs: 111})
+	var batch []*model.RowChangedEvent
+	var ok bool
+	var popRes popResult
+
+	batch = []*model.RowChangedEvent{{StartTs: 1, CommitTs: 11}, {StartTs: 1, CommitTs: 11}}
+	ok, _ = appender.pushBatch(batch, 0, engine.Position{})
+	require.True(t, ok)
+
+	batch = []*model.RowChangedEvent{{StartTs: 2, CommitTs: 12}}
+	ok, _ = appender.pushBatch(batch, 0, engine.Position{})
+	require.True(t, ok)
+
+	popRes = appender.pop(engine.Position{StartTs: 1, CommitTs: 2}, engine.Position{StartTs: 3, CommitTs: 4})
+	require.False(t, popRes.success)
+	require.Equal(t, engine.Position{StartTs: 4, CommitTs: 4}, popRes.lowerBoundIfFail)
+
+	popRes = appender.pop(engine.Position{StartTs: 1, CommitTs: 2}, engine.Position{StartTs: 300, CommitTs: 400})
+	require.False(t, popRes.success)
+	require.Equal(t, engine.Position{StartTs: 101, CommitTs: 111}, popRes.lowerBoundIfFail)
+
+	popRes = appender.pop(engine.Position{StartTs: 1, CommitTs: 11}, engine.Position{StartTs: 2, CommitTs: 12})
+	require.False(t, popRes.success)
+	require.Equal(t, engine.Position{StartTs: 3, CommitTs: 12}, popRes.lowerBoundIfFail)
+
+	batch = []*model.RowChangedEvent{{StartTs: 101, CommitTs: 111}, {StartTs: 101, CommitTs: 111}}
+	ok, _ = appender.pushBatch(batch, 0, engine.Position{StartTs: 101, CommitTs: 111})
+	require.True(t, ok)
+
+	batch = []*model.RowChangedEvent{{StartTs: 102, CommitTs: 112}}
+	ok, _ = appender.pushBatch(batch, 0, engine.Position{})
+	require.True(t, ok)
+	require.Equal(t, 5, appender.readyCount)
+
+	popRes = appender.pop(engine.Position{StartTs: 101, CommitTs: 111}, engine.Position{StartTs: 102, CommitTs: 112})
+	require.True(t, popRes.success)
+	require.Equal(t, engine.Position{StartTs: 101, CommitTs: 111}, popRes.upperBoundIfSuccess)
+	require.Equal(t, 2, len(popRes.events))
+	require.Equal(t, 1, popRes.pushCount)
+	require.Equal(t, uint64(101), popRes.events[1].StartTs)
+	require.Equal(t, 0, appender.readyCount)
+
+	popRes = appender.pop(engine.Position{StartTs: 102, CommitTs: 111}, engine.Position{StartTs: 102, CommitTs: 112})
+	require.False(t, popRes.success)
+	require.Equal(t, engine.Position{StartTs: 103, CommitTs: 112}, popRes.lowerBoundIfFail)
+
+	batch = []*model.RowChangedEvent{{StartTs: 102, CommitTs: 112}}
+	ok, _ = appender.pushBatch(batch, 0, engine.Position{StartTs: 102, CommitTs: 102})
+	require.True(t, ok)
+	require.Equal(t, 2, appender.readyCount)
+
+	popRes = appender.pop(engine.Position{StartTs: 501, CommitTs: 502}, engine.Position{StartTs: 701, CommitTs: 702})
+	require.True(t, popRes.success)
+	require.Equal(t, 0, len(popRes.events))
+	require.Equal(t, engine.Position{StartTs: 500, CommitTs: 502}, popRes.upperBoundIfSuccess)
+	require.Equal(t, 0, appender.readyCount)
+	require.Equal(t, 0, len(appender.events))
+}
diff --git a/cdc/processor/sinkmanager/table_sink_worker_test.go b/cdc/processor/sinkmanager/table_sink_worker_test.go
index a2376b8ae..4f9891deb 100644
--- a/cdc/processor/sinkmanager/table_sink_worker_test.go
+++ b/cdc/processor/sinkmanager/table_sink_worker_test.go
@@ -1231,3 +1231,51 @@ func genUpperBoundGetter(commitTs model.Ts) func(_ model.Ts) engine.Position {
 		}
 	}
 }
+
+func (suite *workerSuite) TestHandleTaskWithCache() {
+	ctx, cancel := context.WithCancel(context.Background())
+	events := []*model.PolymorphicEvent{
+		genPolymorphicEvent(2, 4, 1),
+		genPolymorphicEvent(2, 4, 1),
+		genPolymorphicResolvedEvent(4),
+	}
+	w, e := createWorker(model.ChangeFeedID{}, 0, true, 1)
+	w.eventCache = newRedoEventCache(model.ChangeFeedID{}, 1024*1024)
+	appender := w.eventCache.maybeCreateAppender(1, engine.Position{StartTs: 1, CommitTs: 3})
+	appender.pushBatch(
+		[]*model.RowChangedEvent{events[0].Row, events[1].Row},
+		uint64(0), engine.Position{StartTs: 2, CommitTs: 4},
+	)
+	defer w.sinkMemQuota.Close()
+	addEventsToSortEngine(suite.T(), events, e, 1)
+
+	taskChan := make(chan *sinkTask)
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		err := w.handleTasks(ctx, taskChan)
+		require.Equal(suite.T(), context.Canceled, err)
+	}()
+
+	wrapper, sink := createTableSinkWrapper(model.ChangeFeedID{}, 1)
+	defer sink.Close()
+
+	chShouldBeClosed := make(chan struct{}, 1)
+	callback := func(lastWrittenPos engine.Position) {
+		require.Equal(suite.T(), engine.Position{StartTs: 2, CommitTs: 4}, lastWrittenPos)
+		close(chShouldBeClosed)
+	}
+	taskChan <- &sinkTask{
+		tableID:       1,
+		lowerBound:    engine.Position{StartTs: 1, CommitTs: 3},
+		getUpperBound: genUpperBoundGetter(4),
+		tableSink:     wrapper,
+		callback:      callback,
+		isCanceled:    func() bool { return true },
+	}
+
+	<-chShouldBeClosed
+	cancel()
+	wg.Wait()
+}
diff --git a/cdc/redo/manager_test.go b/cdc/redo/manager_test.go
index 9d29fe3d7..e30eae138 100644
--- a/cdc/redo/manager_test.go
+++ b/cdc/redo/manager_test.go
@@ -54,9 +54,8 @@ func runBenchTest(b *testing.B, storage string, useFileBackend bool) {
 		MetaFlushIntervalInMs: redo.MinFlushIntervalInMs,
 		UseFileBackend:        useFileBackend,
 	}
-	dmlMgr, err := NewDMLManager(ctx, cfg)
-	require.Nil(b, err)
-	eg := errgroup.Group{}
+	dmlMgr := NewDMLManager(model.DefaultChangeFeedID("test"), cfg)
+	var eg errgroup.Group
 	eg.Go(func() error {
 		return dmlMgr.Run(ctx)
 	})
@@ -124,6 +123,6 @@ func runBenchTest(b *testing.B, storage string, useFileBackend bool) {
 		time.Sleep(time.Millisecond * 500)
 	}
 	cancel()
-	err = eg.Wait()
-	require.ErrorIs(b, err, context.Canceled)
+
+	require.ErrorIs(b, eg.Wait(), context.Canceled)
 }
diff --git a/cdc/redo/meta_manager_test.go b/cdc/redo/meta_manager_test.go
index e630d19f2..51ddb37e9 100644
--- a/cdc/redo/meta_manager_test.go
+++ b/cdc/redo/meta_manager_test.go
@@ -51,17 +51,18 @@ func TestInitAndWriteMeta(t *testing.T) {
 		{CheckpointTs: 8, ResolvedTs: 9},
 		{CheckpointTs: 9, ResolvedTs: 11},
 	}
-	toReomoveFiles := []string{}
+
+	var toRemoveFiles []string
 	for _, meta := range metas {
 		data, err := meta.MarshalMsg(nil)
 		require.NoError(t, err)
 		metaName := getMetafileName(captureID, changefeedID, uuid.NewGenerator())
 		err = extStorage.WriteFile(ctx, metaName, data)
 		require.NoError(t, err)
-		toReomoveFiles = append(toReomoveFiles, metaName)
+		toRemoveFiles = append(toRemoveFiles, metaName)
 	}
-	// err = extStorage.WriteFile(ctx, getDeletedChangefeedMarker(changefeedID), []byte{})
-	notRemoveFiles := []string{}
+
+	var notRemoveFiles []string
 	require.NoError(t, err)
 	for i := 0; i < 10; i++ {
 		fileName := "dummy" + getChangefeedMatcher(changefeedID) + strconv.Itoa(i)
@@ -78,11 +79,22 @@ func TestInitAndWriteMeta(t *testing.T) {
 		FlushIntervalInMs:     redo.MinFlushIntervalInMs,
 		MetaFlushIntervalInMs: redo.MinFlushIntervalInMs,
 	}
-	m, err := NewMetaManagerWithInit(ctx, cfg, startTs)
-	require.NoError(t, err)
-	require.Equal(t, startTs, m.metaCheckpointTs.getFlushed())
-	require.Equal(t, uint64(11), m.metaResolvedTs.getFlushed())
-	for _, fileName := range toReomoveFiles {
+	m := NewMetaManager(changefeedID, cfg, startTs)
+
+	var eg errgroup.Group
+	eg.Go(func() error {
+		return m.Run(ctx)
+	})
+
+	require.Eventually(t, func() bool {
+		return startTs == m.metaCheckpointTs.getFlushed()
+	}, time.Second, 50*time.Millisecond)
+
+	require.Eventually(t, func() bool {
+		return uint64(11) == m.metaResolvedTs.getFlushed()
+	}, time.Second, 50*time.Millisecond)
+
+	for _, fileName := range toRemoveFiles {
 		ret, err := extStorage.FileExists(ctx, fileName)
 		require.NoError(t, err)
 		require.False(t, ret, "file %s should be removed", fileName)
@@ -93,7 +105,10 @@ func TestInitAndWriteMeta(t *testing.T) {
 		require.True(t, ret, "file %s should not be removed", fileName)
 	}
 
-	testWriteMeta(t, m)
+	testWriteMeta(ctx, t, m)
+
+	cancel()
+	require.ErrorIs(t, eg.Wait(), context.Canceled)
 }
 
 func TestPreCleanupAndWriteMeta(t *testing.T) {
@@ -116,7 +131,7 @@ func TestPreCleanupAndWriteMeta(t *testing.T) {
 		{CheckpointTs: 9, ResolvedTs: 11},
 		{CheckpointTs: 11, ResolvedTs: 12},
 	}
-	toRemoveFiles := []string{}
+	var toRemoveFiles []string
 	for _, meta := range metas {
 		data, err := meta.MarshalMsg(nil)
 		require.NoError(t, err)
@@ -143,24 +158,36 @@ func TestPreCleanupAndWriteMeta(t *testing.T) {
 		FlushIntervalInMs:     redo.MinFlushIntervalInMs,
 		MetaFlushIntervalInMs: redo.MinFlushIntervalInMs,
 	}
-	m, err := NewMetaManagerWithInit(ctx, cfg, startTs)
-	require.NoError(t, err)
-	require.Equal(t, startTs, m.metaCheckpointTs.getFlushed())
-	require.Equal(t, startTs, m.metaResolvedTs.getFlushed())
+	m := NewMetaManager(changefeedID, cfg, startTs)
+
+	var eg errgroup.Group
+	eg.Go(func() error {
+		return m.Run(ctx)
+	})
+
+	require.Eventually(t, func() bool {
+		return startTs == m.metaCheckpointTs.getFlushed()
+	}, time.Second, 50*time.Millisecond)
+
+	require.Eventually(t, func() bool {
+		return startTs == m.metaResolvedTs.getFlushed()
+	}, time.Second, 50*time.Millisecond)
+
 	for _, fileName := range toRemoveFiles {
 		ret, err := extStorage.FileExists(ctx, fileName)
 		require.NoError(t, err)
 		require.False(t, ret, "file %s should be removed", fileName)
 	}
-	testWriteMeta(t, m)
-}
+	testWriteMeta(ctx, t, m)
 
-func testWriteMeta(t *testing.T, m *metaManager) {
-	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	require.ErrorIs(t, eg.Wait(), context.Canceled)
+}
 
-	checkMeta := func(targetCkpts, targetRts uint64) {
+func testWriteMeta(ctx context.Context, t *testing.T, m *metaManager) {
+	checkMeta := func(targetCheckpointTs, targetResolvedTs uint64) {
 		var checkpointTs, resolvedTs uint64
-		metas := []*common.LogMeta{}
+		var metas []*common.LogMeta
 		cnt := 0
 		m.extStorage.WalkDir(ctx, nil, func(path string, size int64) error {
 			if !strings.HasSuffix(path, redo.MetaEXT) {
@@ -177,14 +204,10 @@ func testWriteMeta(t *testing.T, m *metaManager) {
 		})
 		require.Equal(t, 1, cnt)
 		common.ParseMeta(metas, &checkpointTs, &resolvedTs)
-		require.Equal(t, targetCkpts, checkpointTs)
-		require.Equal(t, targetRts, resolvedTs)
+		require.Equal(t, targetCheckpointTs, checkpointTs)
+		require.Equal(t, targetResolvedTs, resolvedTs)
 	}
 
-	eg := errgroup.Group{}
-	eg.Go(func() error {
-		return m.Run(ctx)
-	})
 	// test both regressed
 	meta := m.GetFlushedMeta()
 	m.UpdateMeta(1, 2)
@@ -210,9 +233,6 @@ func testWriteMeta(t *testing.T, m *metaManager) {
 		return m.metaCheckpointTs.getFlushed() == 16
 	}, time.Second, 50*time.Millisecond)
 	checkMeta(16, 21)
-
-	cancel()
-	require.ErrorIs(t, eg.Wait(), context.Canceled)
 }
 
 func TestGCAndCleanup(t *testing.T) {
@@ -275,15 +295,21 @@ func TestGCAndCleanup(t *testing.T) {
 		FlushIntervalInMs:     redo.MinFlushIntervalInMs,
 		MetaFlushIntervalInMs: redo.MinFlushIntervalInMs,
 	}
-	m, err := NewMetaManagerWithInit(ctx, cfg, startTs)
-	require.NoError(t, err)
-	require.Equal(t, startTs, m.metaCheckpointTs.getFlushed())
-	require.Equal(t, startTs, m.metaResolvedTs.getFlushed())
+	m := NewMetaManager(changefeedID, cfg, startTs)
 
-	eg := errgroup.Group{}
+	var eg errgroup.Group
 	eg.Go(func() error {
 		return m.Run(ctx)
 	})
+
+	require.Eventually(t, func() bool {
+		return startTs == m.metaCheckpointTs.getFlushed()
+	}, time.Second, 50*time.Millisecond)
+
+	require.Eventually(t, func() bool {
+		return startTs == m.metaResolvedTs.getFlushed()
+	}, time.Second, 50*time.Millisecond)
+
 	checkGC(startTs)
 
 	for i := startTs; i <= uint64(maxCommitTs); i++ {
diff --git a/cdc/redo/writer/memory/mem_log_writer_test.go b/cdc/redo/writer/memory/mem_log_writer_test.go
index 8d1a03a71..53943e52a 100644
--- a/cdc/redo/writer/memory/mem_log_writer_test.go
+++ b/cdc/redo/writer/memory/mem_log_writer_test.go
@@ -68,7 +68,7 @@ func testWriteEvents(t *testing.T, events []writer.RedoEvent) {
 		LogType:            redo.RedoDDLLogFileType,
 		CaptureID:          "test-capture",
 		ChangeFeedID:       model.DefaultChangeFeedID("test-changefeed"),
-		URI:                *uri,
+		URI:                uri,
 		UseExternalStorage: true,
 		MaxLogSizeInBytes:  10 * redo.Megabyte,
 	}
diff --git a/cdc/scheduler/internal/v3/replication/replication_manager_test.go b/cdc/scheduler/internal/v3/replication/replication_manager_test.go
index c5e7aad2c..cc52f042b 100644
--- a/cdc/scheduler/internal/v3/replication/replication_manager_test.go
+++ b/cdc/scheduler/internal/v3/replication/replication_manager_test.go
@@ -589,6 +589,10 @@ func (m *mockRedoMetaManager) Run(ctx context.Context) error {
 	return nil
 }
 
+func (m *mockRedoMetaManager) Running() bool {
+	return true
+}
+
 func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 	t.Parallel()
 
diff --git a/cdc/sink/mq/manager/kafka_manager_test.go b/cdc/sink/mq/manager/kafka_manager_test.go
index a1d619637..6f817bc76 100644
--- a/cdc/sink/mq/manager/kafka_manager_test.go
+++ b/cdc/sink/mq/manager/kafka_manager_test.go
@@ -35,11 +35,12 @@ func TestPartitions(t *testing.T) {
 		ReplicationFactor: 1,
 	}
 
-	manager := NewKafkaTopicManager(context.TODO(), adminClient, cfg)
+	ctx := context.Background()
+	manager := NewKafkaTopicManager(ctx, kafka.DefaultMockTopicName, adminClient, cfg)
 	defer manager.Close()
 	partitionsNum, err := manager.GetPartitionNum(kafka.DefaultMockTopicName)
-	require.Nil(t, err)
-	require.Equal(t, int32(3), partitionsNum)
+	require.NoError(t, err)
+	require.Equal(t, int32(2), partitionsNum)
 }
 
 func TestCreateTopic(t *testing.T) {
@@ -57,22 +58,22 @@ func TestCreateTopic(t *testing.T) {
 		ReplicationFactor: 1,
 	}
 
-	manager := NewKafkaTopicManager(ctx, adminClient, cfg)
+	manager := NewKafkaTopicManager(ctx, kafka.DefaultMockTopicName, adminClient, cfg)
 	defer manager.Close()
 	partitionNum, err := manager.CreateTopicAndWaitUntilVisible(kafka.DefaultMockTopicName)
-	require.Nil(t, err)
-	require.Equal(t, int32(3), partitionNum)
+	require.NoError(t, err)
+	require.Equal(t, int32(2), partitionNum)
 
 	partitionNum, err = manager.CreateTopicAndWaitUntilVisible("new-topic")
-	require.Nil(t, err)
+	require.NoError(t, err)
 	require.Equal(t, int32(2), partitionNum)
 	partitionsNum, err := manager.GetPartitionNum("new-topic")
-	require.Nil(t, err)
+	require.NoError(t, err)
 	require.Equal(t, int32(2), partitionsNum)
 
 	// Try to create a topic without auto create.
 	cfg.AutoCreate = false
-	manager = NewKafkaTopicManager(ctx, adminClient, cfg)
+	manager = NewKafkaTopicManager(ctx, "new-topic2", adminClient, cfg)
 	defer manager.Close()
 	_, err = manager.CreateTopicAndWaitUntilVisible("new-topic2")
 	require.Regexp(
@@ -81,6 +82,7 @@ func TestCreateTopic(t *testing.T) {
 		err,
 	)
 
+	topic := "new-topic-failed"
 	// Invalid replication factor.
 	// It happens when replication-factor is greater than the number of brokers.
 	cfg = &kafkaconfig.AutoCreateTopicConfig{
@@ -88,9 +90,9 @@ func TestCreateTopic(t *testing.T) {
 		PartitionNum:      2,
 		ReplicationFactor: 4,
 	}
-	manager = NewKafkaTopicManager(ctx, adminClient, cfg)
+	manager = NewKafkaTopicManager(ctx, topic, adminClient, cfg)
 	defer manager.Close()
-	_, err = manager.CreateTopicAndWaitUntilVisible("new-topic-failed")
+	_, err = manager.CreateTopicAndWaitUntilVisible(topic)
 	require.Regexp(
 		t,
 		"new sarama producer: kafka server: Replication-factor is invalid",
@@ -111,13 +113,15 @@ func TestCreateTopicWithDelay(t *testing.T) {
 		ReplicationFactor: 1,
 	}
 
-	manager := NewKafkaTopicManager(context.TODO(), adminClient, cfg)
+	topic := "new_topic"
+	ctx := context.Background()
+	manager := NewKafkaTopicManager(ctx, topic, adminClient, cfg)
 	defer manager.Close()
-	partitionNum, err := manager.createTopic("new_topic")
-	require.Nil(t, err)
-	err = adminClient.SetRemainingFetchesUntilTopicVisible("new_topic", 3)
-	require.Nil(t, err)
-	err = manager.waitUntilTopicVisible("new_topic")
-	require.Nil(t, err)
+	partitionNum, err := manager.createTopic(topic)
+	require.NoError(t, err)
+	err = adminClient.SetRemainingFetchesUntilTopicVisible(topic, 3)
+	require.NoError(t, err)
+	err = manager.waitUntilTopicVisible(topic)
+	require.NoError(t, err)
 	require.Equal(t, int32(2), partitionNum)
 }
diff --git a/cdc/sinkv2/ddlsink/mq/mq_ddl_sink_test.go b/cdc/sinkv2/ddlsink/mq/mq_ddl_sink_test.go
index d386fe39d..d6d4008f8 100644
--- a/cdc/sinkv2/ddlsink/mq/mq_ddl_sink_test.go
+++ b/cdc/sinkv2/ddlsink/mq/mq_ddl_sink_test.go
@@ -82,8 +82,9 @@ func TestWriteDDLEventToAllPartitions(t *testing.T) {
 
 	leader, topic := initBroker(t, kafka.DefaultMockPartitionNum)
 	defer leader.Close()
+	// partition-number is 2, so only send DDL events to 2 partitions.
 	uriTemplate := "kafka://%s/%s?kafka-version=0.9.0.0&max-batch-size=1" +
-		"&max-message-bytes=1048576&partition-num=1" +
+		"&max-message-bytes=1048576&partition-num=2" +
 		"&kafka-client-id=unit-test&auto-create-topic=false&compression=gzip&protocol=open-protocol"
 	uri := fmt.Sprintf(uriTemplate, leader.Addr(), topic)
 
@@ -110,7 +111,7 @@ func TestWriteDDLEventToAllPartitions(t *testing.T) {
 	err = s.WriteDDLEvent(ctx, ddl)
 	require.Nil(t, err)
 	require.Len(t, s.producer.(*ddlproducer.MockDDLProducer).GetAllEvents(),
-		3, "All partitions should be broadcast")
+		2, "All partitions should be broadcast")
 	require.Len(t, s.producer.(*ddlproducer.MockDDLProducer).GetEvents(mqv1.TopicPartitionKey{
 		Topic:     "mock_topic",
 		Partition: 0,
@@ -119,10 +120,6 @@ func TestWriteDDLEventToAllPartitions(t *testing.T) {
 		Topic:     "mock_topic",
 		Partition: 1,
 	}), 1)
-	require.Len(t, s.producer.(*ddlproducer.MockDDLProducer).GetEvents(mqv1.TopicPartitionKey{
-		Topic:     "mock_topic",
-		Partition: 2,
-	}), 1)
 }
 
 func TestWriteDDLEventToZeroPartition(t *testing.T) {
@@ -184,8 +181,9 @@ func TestWriteCheckpointTsToDefaultTopic(t *testing.T) {
 
 	leader, topic := initBroker(t, kafka.DefaultMockPartitionNum)
 	defer leader.Close()
+	// partition-num is set to 2, so send checkpoint to 2 partitions.
 	uriTemplate := "kafka://%s/%s?kafka-version=0.9.0.0&max-batch-size=1" +
-		"&max-message-bytes=1048576&partition-num=1" +
+		"&max-message-bytes=1048576&partition-num=2" +
 		"&kafka-client-id=unit-test&auto-create-topic=false&compression=gzip" +
 		"&protocol=canal-json&enable-tidb-extension=true"
 	uri := fmt.Sprintf(uriTemplate, leader.Addr(), topic)
@@ -206,7 +204,7 @@ func TestWriteCheckpointTsToDefaultTopic(t *testing.T) {
 	require.Nil(t, err)
 
 	require.Len(t, s.producer.(*ddlproducer.MockDDLProducer).GetAllEvents(),
-		3, "All partitions should be broadcast")
+		2, "All partitions should be broadcast")
 	require.Len(t, s.producer.(*ddlproducer.MockDDLProducer).GetEvents(mqv1.TopicPartitionKey{
 		Topic:     "mock_topic",
 		Partition: 0,
@@ -215,10 +213,6 @@ func TestWriteCheckpointTsToDefaultTopic(t *testing.T) {
 		Topic:     "mock_topic",
 		Partition: 1,
 	}), 1)
-	require.Len(t, s.producer.(*ddlproducer.MockDDLProducer).GetEvents(mqv1.TopicPartitionKey{
-		Topic:     "mock_topic",
-		Partition: 2,
-	}), 1)
 }
 
 func TestWriteCheckpointTsToTableTopics(t *testing.T) {
@@ -278,19 +272,11 @@ func TestWriteCheckpointTsToTableTopics(t *testing.T) {
 	require.Nil(t, err)
 
 	require.Len(t, s.producer.(*ddlproducer.MockDDLProducer).GetAllEvents(),
-		6, "All topics and partitions should be broadcast")
+		4, "All topics and partitions should be broadcast")
 	require.Len(t, s.producer.(*ddlproducer.MockDDLProducer).GetEvents(mqv1.TopicPartitionKey{
 		Topic:     "mock_topic",
 		Partition: 0,
 	}), 1)
-	require.Len(t, s.producer.(*ddlproducer.MockDDLProducer).GetEvents(mqv1.TopicPartitionKey{
-		Topic:     "mock_topic",
-		Partition: 1,
-	}), 1)
-	require.Len(t, s.producer.(*ddlproducer.MockDDLProducer).GetEvents(mqv1.TopicPartitionKey{
-		Topic:     "mock_topic",
-		Partition: 2,
-	}), 1)
 	require.Len(t, s.producer.(*ddlproducer.MockDDLProducer).GetEvents(mqv1.TopicPartitionKey{
 		Topic:     "cdc_person",
 		Partition: 0,
diff --git a/cdc/sinkv2/util/helper_test.go b/cdc/sinkv2/util/helper_test.go
index df7911a85..635737de2 100644
--- a/cdc/sinkv2/util/helper_test.go
+++ b/cdc/sinkv2/util/helper_test.go
@@ -14,12 +14,54 @@
 package util
 
 import (
+	"context"
 	"net/url"
 	"testing"
 
+	"github.com/Shopify/sarama"
+	"github.com/pingcap/tiflow/cdc/sink/mq/producer/kafka"
+	pkafka "github.com/pingcap/tiflow/pkg/sink/kafka"
 	"github.com/stretchr/testify/require"
 )
 
+func TestPartition(t *testing.T) {
+	t.Parallel()
+
+	adminClient := pkafka.NewClusterAdminClientMockImpl()
+	defer adminClient.Close()
+
+	cfg := &kafka.AutoCreateTopicConfig{
+		AutoCreate:        true,
+		PartitionNum:      2,
+		ReplicationFactor: 1,
+	}
+
+	ctx := context.Background()
+	manager, err := GetTopicManagerAndTryCreateTopic(ctx, pkafka.DefaultMockTopicName, cfg, adminClient)
+	require.NoError(t, err)
+	defer manager.Close()
+
+	// default topic, real partition is 3, but 2 is set in the sink-uri, so return 2.
+	partitionsNum, err := manager.GetPartitionNum(pkafka.DefaultMockTopicName)
+	require.NoError(t, err)
+	require.Equal(t, int32(2), partitionsNum)
+
+	// new topic, create it with partition number as 2.
+	partitionsNum, err = manager.GetPartitionNum("new-topic")
+	require.NoError(t, err)
+	require.Equal(t, int32(2), partitionsNum)
+
+	// assume a topic already exist, the not default topic won't be affected by the default topic's partition number.
+	err = adminClient.CreateTopic("new-topic-2", &sarama.TopicDetail{
+		NumPartitions: 3,
+	}, false)
+	require.NoError(t, err)
+
+	partitionsNum, err = manager.GetPartitionNum("new-topic-2")
+	require.NoError(t, err)
+	require.Equal(t, int32(3), partitionsNum)
+}
+
 func TestGetTopic(t *testing.T) {
 	t.Parallel()
 
diff --git a/dm/master/shardddl/optimist_test.go b/dm/master/shardddl/optimist_test.go
index f996b3914..469f413e7 100644
--- a/dm/master/shardddl/optimist_test.go
+++ b/dm/master/shardddl/optimist_test.go
@@ -759,7 +759,7 @@ func (t *testOptimistSuite) TestOptimistLockConflict() {
 	case <-time.After(watchTimeout):
 		t.T().Fatal("timeout")
 	case op3 := <-opCh:
-		require.Equal(t.T(), []string{}, op3.DDLs)
+		require.Equal(t.T(), DDLs1, op3.DDLs)
 		require.Equal(t.T(), optimism.ConflictNone, op3.ConflictStage)
 	}
 	cancel2()
diff --git a/dm/pkg/gtid/gtid_test.go b/dm/pkg/gtid/gtid_test.go
index 275b82527..7698128c5 100644
--- a/dm/pkg/gtid/gtid_test.go
+++ b/dm/pkg/gtid/gtid_test.go
@@ -67,7 +67,7 @@ func TestParseGTIDNoFlavor(t *testing.T) {
 	require.Error(t, err)
 }
 
-func TestIsNilGTIDSet(t *testing.T) {
+func TestIsNilMySQLGTIDSet(t *testing.T) {
 	t.Parallel()
 
 	require.False(t, IsZeroMySQLGTIDSet(""))
@@ -79,6 +79,17 @@ func TestIsNilGTIDSet(t *testing.T) {
 	require.True(t, IsZeroMySQLGTIDSet(" xxxxx:0 "))
 }
 
+func TestIsNilMariaDBGTIDSet(t *testing.T) {
+	t.Parallel()
+
+	require.False(t, IsZeroMariaDBGTIDSet(""))
+	require.False(t, IsZeroMariaDBGTIDSet("xxxxx"))
+	require.False(t, IsZeroMariaDBGTIDSet("a-b-0,c-d:0"))
+	require.False(t, IsZeroMariaDBGTIDSet("xxxxx:1"))
+	require.True(t, IsZeroMariaDBGTIDSet("x-y-0"))
+	require.True(t, IsZeroMariaDBGTIDSet(" x-y-0 "))
+}
+
 func TestParseZeroAsEmptyGTIDSet(t *testing.T) {
 	t.Parallel()
 
@@ -94,3 +105,50 @@ func TestParseZeroAsEmptyGTIDSet(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, "", gset.String())
 }
+
+func TestIsZeroGTIDSet(t *testing.T) {
+	t.Parallel()
+
+	testCases := []struct {
+		gsetStr string
+		isZero  bool
+		flavor  string
+	}{
+		{
+			"",
+			true,
+			mysql.MySQLFlavor,
+		},
+		{
+			"",
+			true,
+			mysql.MariaDBFlavor,
+		},
+		{
+			"3ccc475b-2343-11e7-be21-6c0b84d59f30:0",
+			true,
+			mysql.MySQLFlavor,
+		},
+		{
+			"0-0-0",
+			true,
+			mysql.MariaDBFlavor,
+		},
+		{
+			"3ccc475b-2343-11e7-be21-6c0b84d59f30:1-14",
+			false,
+			mysql.MySQLFlavor,
+		},
+		{
+			"1-2-3",
+			false,
+			mysql.MariaDBFlavor,
+		},
+	}
+	for i, testCase := range testCases {
+		t.Logf("test case %d", i)
+		gset, err := ParserGTID(testCase.flavor, testCase.gsetStr)
+		require.NoError(t, err)
+		require.Equal(t, testCase.isZero, IsZeroGTIDSet(gset))
+	}
+}
diff --git a/dm/pkg/shardddl/optimism/lock_test.go b/dm/pkg/shardddl/optimism/lock_test.go
index e4a50a696..063b53b6e 100644
--- a/dm/pkg/shardddl/optimism/lock_test.go
+++ b/dm/pkg/shardddl/optimism/lock_test.go
@@ -2738,6 +2738,11 @@ func (t *testLock) TestTrySyncForOneDDL(c *C) {
 	c.Assert(schemaChanged, IsTrue)
 	c.Assert(conflictStage, Equals, ConflictNone)
 
+	// check create partition, no changed since https://github.com/pingcap/tidb-tools/blob/d671b0840063bc2532941f02e02e12627402844c/pkg/schemacmp/table.go#L251
+	schemaChanged, conflictStage = l.trySyncForOneDDL(source, schema, table1, t0, t1)
+	c.Assert(schemaChanged, IsTrue)
+	c.Assert(conflictStage, Equals, ConflictNone)
+
 	// check alter table drop column
 	schemaChanged, conflictStage = l.trySyncForOneDDL(source, schema, table2, t0, t2)
 	c.Assert(schemaChanged, IsFalse)
diff --git a/dm/syncer/opt_sharding_group_test.go b/dm/syncer/opt_sharding_group_test.go
index 88ad02cb8..9cd4eec7e 100644
--- a/dm/syncer/opt_sharding_group_test.go
+++ b/dm/syncer/opt_sharding_group_test.go
@@ -73,6 +73,11 @@ func (s *optShardingGroupSuite) TestLowestFirstPosInOptGroups() {
 	k.removeShardingReSync(&ShardingReSync{targetTable: utils.UnpackTableID(db2tbl)})
 	// should be pos11 now, pos21 is totally resolved
 	require.Equal(s.T(), pos11.Position, k.lowestFirstLocationInGroups().Position)
+
+	// reset
+	k.Reset()
+	require.Len(s.T(), k.groups, 0)
+	require.Len(s.T(), k.shardingReSyncs, 0)
 }
 
 func (s *optShardingGroupSuite) TestSync() {
diff --git a/dm/syncer/syncer_test.go b/dm/syncer/syncer_test.go
index 20e87d271..f308cfb20 100644
--- a/dm/syncer/syncer_test.go
+++ b/dm/syncer/syncer_test.go
@@ -978,7 +978,11 @@ func (s *testSyncerSuite) TestRun(c *C) {
 	ctx, cancel = context.WithCancel(context.Background())
 	resultCh = make(chan pb.ProcessResult)
 	// simulate `syncer.Resume` here, but doesn't reset database conns
+	syncer.secondsBehindMaster.Store(100)
+	syncer.workerJobTSArray[ddlJobIdx].Store(100)
 	syncer.reset()
+	c.Assert(syncer.secondsBehindMaster.Load(), Equals, int64(0))
+	c.Assert(syncer.workerJobTSArray[ddlJobIdx].Load(), Equals, int64(0))
 	mockStreamerProducer = &MockStreamProducer{s.generateEvents(events2, c)}
 	mockStreamer, err = mockStreamerProducer.GenerateStreamFrom(binlog.MustZeroLocation(mysql.MySQLFlavor))
 	c.Assert(err, IsNil)
diff --git a/dm/tests/shardddl2/run.sh b/dm/tests/shardddl2/run.sh
index e744700e5..1be8dde17 100644
--- a/dm/tests/shardddl2/run.sh
+++ b/dm/tests/shardddl2/run.sh
@@ -507,6 +507,27 @@ function DM_DropAddColumn() {
 	done
 }
 
+function DM_ADD_DROP_PARTITIONS_CASE() {
+	run_sql_source1 "insert into ${shardddl1}.${tb1} values(1);"
+	run_sql_source2 "insert into ${shardddl1}.${tb1} values(2);"
+	run_sql_source1 "alter table ${shardddl1}.${tb1} add column new_col1 int;"
+	run_sql_source1 "insert into ${shardddl1}.${tb1} values(3,3);"
+	run_sql_source2 "insert into ${shardddl1}.${tb1} values(4);"
+
+	run_sql_source1 "ALTER TABLE ${shardddl1}.${tb1} ADD PARTITION (partition p1 VALUES LESS THAN (10000))"
+	run_sql_tidb_with_retry "SELECT count(1) FROM information_schema.partitions WHERE TABLE_SCHEMA='${shardddl}' AND TABLE_NAME = '${tb}' AND PARTITION_NAME IS NOT NULL;" "count(1): 2"
+
+	run_sql_source1 "ALTER TABLE ${shardddl1}.${tb1} DROP PARTITION p1;"
+	run_sql_tidb_with_retry "SELECT count(1) FROM information_schema.partitions WHERE TABLE_SCHEMA='${shardddl}' AND TABLE_NAME = '${tb}' AND PARTITION_NAME IS NOT NULL;" "count(1): 1"
+}
+
+function DM_ADD_DROP_PARTITIONS() {
+	run_case ADD_DROP_PARTITIONS "double-source-optimistic" \
+		"run_sql_source1 \"create table ${shardddl1}.${tb1} (a int) PARTITION BY RANGE (a) (PARTITION p0 VALUES LESS THAN (100));\"; \
+     run_sql_source2 \"create table ${shardddl1}.${tb1} (a int) PARTITION BY RANGE (a) (PARTITION p0 VALUES LESS THAN (100));\";" \
+		"clean_table" "optimistic"
+}
+
 function run() {
 	init_cluster
 	init_database
@@ -515,6 +536,7 @@ function run() {
 	DM_DROP_COLUMN_ALL_DONE
 	DM_RECOVER_LOCK
 	DM_DropAddColumn
+	DM_ADD_DROP_PARTITIONS
 	start=36
 	end=45
 	except=(042 044 045)
diff --git a/pkg/cmd/server/server_test.go b/pkg/cmd/server/server_test.go
index a3fec8717..3eaa97aaf 100644
--- a/pkg/cmd/server/server_test.go
+++ b/pkg/cmd/server/server_test.go
@@ -216,7 +216,7 @@ func TestParseCfg(t *testing.T) {
 			EnableNewSink: true,
 		},
 		ClusterID:           "default",
-		MaxMemoryPercentage: config.DefaultMaxMemoryPercentage,
+		MaxMemoryPercentage: config.DisableMemoryLimit,
 	}, o.serverConfig)
 }
 
@@ -376,7 +376,7 @@ check-balance-interval = "10s"
 			EnableNewSink: true,
 		},
 		ClusterID:           "default",
-		MaxMemoryPercentage: config.DefaultMaxMemoryPercentage,
+		MaxMemoryPercentage: config.DisableMemoryLimit,
 	}, o.serverConfig)
 }
 
@@ -522,7 +522,7 @@ cert-allowed-cn = ["dd","ee"]
 			EnableNewSink: true,
 		},
 		ClusterID:           "default",
-		MaxMemoryPercentage: config.DefaultMaxMemoryPercentage,
+		MaxMemoryPercentage: config.DisableMemoryLimit,
 	}, o.serverConfig)
 }
 
diff --git a/pkg/config/config_test_data.go b/pkg/config/config_test_data.go
index 2d1be9065..e77580a05 100644
--- a/pkg/config/config_test_data.go
+++ b/pkg/config/config_test_data.go
@@ -63,7 +63,8 @@ const (
     "meta-flush-interval": 200,
     "storage": "",
     "use-file-backend": false
-  }
+  },
+  "changefeed-error-stuck-duration": 1800000000000
 }`
 
 	testCfgTestServerConfigMarshal = `{
@@ -151,7 +152,8 @@ const (
     "enable-kv-connect-backoff": false
   },
   "cluster-id": "default",
-  "max-memory-percentage": 70
+  "max-memory-percentage": 0,
+  "gc-tuner-memory-threshold": 0
 }`
 
 	testCfgTestReplicaConfigMarshal1 = `{
@@ -215,7 +217,8 @@ const (
     "meta-flush-interval": 200,
     "storage": "",
     "use-file-backend": false
-  }
+  },
+  "changefeed-error-stuck-duration": 1800000000000
 }`
 
 	testCfgTestReplicaConfigMarshal2 = `{
@@ -276,6 +279,7 @@ const (
     "meta-flush-interval": 200,
     "storage": "",
     "use-file-backend": false
-  }
+  },
+  "changefeed-error-stuck-duration": 1800000000000
 }`
 )
diff --git a/pkg/config/replica_config_test.go b/pkg/config/replica_config_test.go
index 90319372f..63d5c7ca2 100644
--- a/pkg/config/replica_config_test.go
+++ b/pkg/config/replica_config_test.go
@@ -181,6 +181,17 @@ func TestValidateAndAdjust(t *testing.T) {
 
 	cfg.Sink.EncoderConcurrency = -1
 	require.Error(t, cfg.ValidateAndAdjust(sinkURL))
+
+	// changefeed error stuck duration is less than 30 minutes
+	cfg = GetDefaultReplicaConfig()
+	duration := minChangeFeedErrorStuckDuration - time.Second*1
+	cfg.ChangefeedErrorStuckDuration = duration
+	err = cfg.ValidateAndAdjust(sinkURL)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "The ChangefeedErrorStuckDuration")
+	duration = minChangeFeedErrorStuckDuration
+	cfg.ChangefeedErrorStuckDuration = duration
+	require.NoError(t, cfg.ValidateAndAdjust(sinkURL))
 }
 
 func TestAdjustEnableOldValueAndVerifyForceReplicate(t *testing.T) {
diff --git a/pkg/filter/filter_test.go b/pkg/filter/filter_test.go
index 14af7130f..608aadb8b 100644
--- a/pkg/filter/filter_test.go
+++ b/pkg/filter/filter_test.go
@@ -306,3 +306,21 @@ func TestIsAllowedDDL(t *testing.T) {
 		require.Equal(t, tc.allowed, isAllowedDDL(tc.ActionType), "%#v", tc)
 	}
 }
+
+func TestIsSchemaDDL(t *testing.T) {
+	cases := []struct {
+		actionType timodel.ActionType
+		isSchema   bool
+	}{
+		{timodel.ActionCreateSchema, true},
+		{timodel.ActionDropSchema, true},
+		{timodel.ActionModifySchemaCharsetAndCollate, true},
+		{timodel.ActionCreateTable, false},
+		{timodel.ActionDropTable, false},
+		{timodel.ActionTruncateTable, false},
+		{timodel.ActionAddColumn, false},
+	}
+	for _, tc := range cases {
+		require.Equal(t, tc.isSchema, IsSchemaDDL(tc.actionType), "%#v", tc)
+	}
+}
diff --git a/pkg/orchestrator/reactor_state_test.go b/pkg/orchestrator/reactor_state_test.go
index af79b9994..6a33970dd 100644
--- a/pkg/orchestrator/reactor_state_test.go
+++ b/pkg/orchestrator/reactor_state_test.go
@@ -129,6 +129,8 @@ func TestChangefeedStateUpdate(t *testing.T) {
 							AdvanceTimeoutInSec: config.DefaultAdvanceTimeoutInSec,
 						},
 						Consistent: &config.ConsistentConfig{Level: "normal", Storage: "local"},
+						ChangefeedErrorStuckDuration: config.
+							GetDefaultReplicaConfig().ChangefeedErrorStuckDuration,
 					},
 				},
 				Status: &model.ChangeFeedStatus{CheckpointTs: 421980719742451713},
@@ -180,6 +182,8 @@ func TestChangefeedStateUpdate(t *testing.T) {
 							AdvanceTimeoutInSec: config.DefaultAdvanceTimeoutInSec,
 						},
 						Consistent: &config.ConsistentConfig{Level: "normal", Storage: "local"},
+						ChangefeedErrorStuckDuration: config.
+							GetDefaultReplicaConfig().ChangefeedErrorStuckDuration,
 					},
 				},
 				Status: &model.ChangeFeedStatus{CheckpointTs: 421980719742451713},
@@ -236,6 +240,8 @@ func TestChangefeedStateUpdate(t *testing.T) {
 							AdvanceTimeoutInSec: config.DefaultAdvanceTimeoutInSec,
 						},
 						Consistent: &config.ConsistentConfig{Level: "normal", Storage: "local"},
+						ChangefeedErrorStuckDuration: config.
+							GetDefaultReplicaConfig().ChangefeedErrorStuckDuration,
 					},
 				},
 				Status: &model.ChangeFeedStatus{CheckpointTs: 421980719742451713},
@@ -324,6 +330,8 @@ func TestPatchInfo(t *testing.T) {
 			Mounter:    defaultConfig.Mounter,
 			Sink:       defaultConfig.Sink,
 			Consistent: defaultConfig.Consistent,
+			ChangefeedErrorStuckDuration: config.
+				GetDefaultReplicaConfig().ChangefeedErrorStuckDuration,
 		},
 	})
 	state.PatchInfo(func(info *model.ChangeFeedInfo) (*model.ChangeFeedInfo, bool, error) {
@@ -340,6 +348,8 @@ func TestPatchInfo(t *testing.T) {
 			Mounter:    defaultConfig.Mounter,
 			Sink:       defaultConfig.Sink,
 			Consistent: defaultConfig.Consistent,
+			ChangefeedErrorStuckDuration: config.
+				GetDefaultReplicaConfig().ChangefeedErrorStuckDuration,
 		},
 	})
 	state.PatchInfo(func(info *model.ChangeFeedInfo) (*model.ChangeFeedInfo, bool, error) {
diff --git a/pkg/sink/cloudstorage/path_test.go b/pkg/sink/cloudstorage/path_test.go
index 6a0dbe56c..ead5254ff 100644
--- a/pkg/sink/cloudstorage/path_test.go
+++ b/pkg/sink/cloudstorage/path_test.go
@@ -17,9 +17,14 @@ import (
 	"context"
 	"fmt"
 	"net/url"
+	"os"
+	"path/filepath"
 	"testing"
 	"time"
 
+	timodel "github.com/pingcap/tidb/parser/model"
+	"github.com/pingcap/tidb/parser/mysql"
+	"github.com/pingcap/tidb/parser/types"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/engine/pkg/clock"
 	"github.com/pingcap/tiflow/pkg/config"
@@ -275,3 +280,51 @@ func TestIsSchemaFile(t *testing.T) {
 			"testCase: %s, path: %v", tt.name, tt.path)
 	}
 }
+
+func TestCheckOrWriteSchema(t *testing.T) {
+	t.Parallel()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	dir := t.TempDir()
+	f := testFilePathGenerator(ctx, t, dir)
+
+	var columns []*timodel.ColumnInfo
+	ft := types.NewFieldType(mysql.TypeLong)
+	ft.SetFlag(mysql.PriKeyFlag | mysql.NotNullFlag)
+	col := &timodel.ColumnInfo{
+		Name:         timodel.NewCIStr("Id"),
+		FieldType:    *ft,
+		DefaultValue: 10,
+	}
+	columns = append(columns, col)
+	tableInfo := &model.TableInfo{
+		TableInfo: &timodel.TableInfo{Columns: columns},
+		Version:   100,
+		TableName: model.TableName{
+			Schema:  "test",
+			Table:   "table1",
+			TableID: 20,
+		},
+	}
+
+	table := VersionedTableName{
+		TableNameWithPhysicTableID: tableInfo.TableName,
+		TableInfoVersion:           tableInfo.Version,
+	}
+
+	err := f.CheckOrWriteSchema(ctx, table, tableInfo)
+	require.NoError(t, err)
+	require.Equal(t, tableInfo.Version, f.versionMap[table])
+
+	// test only table version changed, schema file should be reused
+	table.TableInfoVersion = 101
+	err = f.CheckOrWriteSchema(ctx, table, tableInfo)
+	require.NoError(t, err)
+	require.Equal(t, tableInfo.Version, f.versionMap[table])
+
+	dir = filepath.Join(dir, "test/table1/meta")
+	cnt, err := os.ReadDir(dir)
+	require.NoError(t, err)
+	require.Equal(t, 1, len(cnt))
+}
diff --git a/tests/integration_tests/api_v2/cases.go b/tests/integration_tests/api_v2/cases.go
index e5d90e4e8..926f5a771 100644
--- a/tests/integration_tests/api_v2/cases.go
+++ b/tests/integration_tests/api_v2/cases.go
@@ -101,7 +101,7 @@ var customReplicaConfig = &ReplicaConfig{
 // defaultReplicaConfig check if the default values is changed
 var defaultReplicaConfig = &ReplicaConfig{
 	MemoryQuota:        1024 * 1024 * 1024,
-	CaseSensitive:      true,
+	CaseSensitive:      false,
 	EnableOldValue:     true,
 	CheckGCSafePoint:   true,
 	EnableSyncPoint:    false,
diff --git a/tests/integration_tests/bdr_mode/data/start.sql b/tests/integration_tests/bdr_mode/data/start.sql
index 2d5007efb..33e6cc75c 100644
--- a/tests/integration_tests/bdr_mode/data/start.sql
+++ b/tests/integration_tests/bdr_mode/data/start.sql
@@ -3,3 +3,4 @@ create database `bdr_mode`;
 use `bdr_mode`;
 
 create table `t1` (id int primary key, name varchar(20));
+create table `t2` (id int primary key, name varchar(20));
\ No newline at end of file
diff --git a/tests/integration_tests/bdr_mode/data/up.sql b/tests/integration_tests/bdr_mode/data/up.sql
index dd926b155..4e36df615 100644
--- a/tests/integration_tests/bdr_mode/data/up.sql
+++ b/tests/integration_tests/bdr_mode/data/up.sql
@@ -16,3 +16,8 @@ insert into `t1` values (22, '22'), (44, '44'), (66, '66'), (88, '88'), (108, '1
 rollback;
 
 insert into `t1` values (100, '100'), (300, '300'), (500, '500'), (700, '700'), (900, '900');
+
+drop table `t2`;
+create table `t2` (id int primary key, name varchar(20));
+insert into `t2` values (1, '1'), (3, '3'), (5, '5'), (7, '7'), (9, '9');
+insert into `t2` values (2, '2'), (4, '4'), (6, '6'), (8, '8'), (10, '10');
\ No newline at end of file
diff --git a/tests/integration_tests/changefeed_error/conf/enable_redo.toml b/tests/integration_tests/changefeed_error/conf/enable_redo.toml
new file mode 100644
index 000000000..4f1ab09b1
--- /dev/null
+++ b/tests/integration_tests/changefeed_error/conf/enable_redo.toml
@@ -0,0 +1,3 @@
+[consistent]
+level = "eventual"
+storage = "file:///tmp/tidb_cdc_test/changefeed_error/redo"
diff --git a/tests/integration_tests/changefeed_error/run.sh b/tests/integration_tests/changefeed_error/run.sh
index abaf3d6b5..67ef5e549 100755
--- a/tests/integration_tests/changefeed_error/run.sh
+++ b/tests/integration_tests/changefeed_error/run.sh
@@ -104,11 +104,11 @@ function run() {
 	cleanup_process $CDC_BINARY
 
 	# make sure initialize changefeed error will not stuck the owner
-	export GO_FAILPOINTS='github.com/pingcap/tiflow/cdc/owner/ChangefeedNewRedoManagerError=2*return(true)'
+	export GO_FAILPOINTS='github.com/pingcap/tiflow/cdc/redo/ChangefeedNewRedoManagerError=2*return(true)'
 	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
 
 	changefeedid_3="changefeed-initialize-error"
-	run_cdc_cli changefeed create --start-ts=0 --sink-uri="$SINK_URI" -c $changefeedid_3
+	run_cdc_cli changefeed create --start-ts=0 --sink-uri="$SINK_URI" -c $changefeedid_3 --config=$CUR/conf/enable_redo.toml
 	ensure $MAX_RETRIES check_changefeed_state http://${UP_PD_HOST_1}:${UP_PD_PORT_1} ${changefeedid_3} "normal" "null" ""
 	run_cdc_cli changefeed pause -c $changefeedid_3
 	ensure $MAX_RETRIES check_changefeed_state http://${UP_PD_HOST_1}:${UP_PD_PORT_1} ${changefeedid_3} "stopped" "changefeed new redo manager injected error" ""
diff --git a/tests/integration_tests/cli/run.sh b/tests/integration_tests/cli/run.sh
index a412bb6b4..28c12c81f 100644
--- a/tests/integration_tests/cli/run.sh
+++ b/tests/integration_tests/cli/run.sh
@@ -69,7 +69,7 @@ function run() {
 
 	# Update changefeed failed because changefeed is running
 	cat - >"$WORK_DIR/changefeed.toml" <<EOF
-case-sensitive = false
+case-sensitive = true
 EOF
 	set +e
 	update_result=$(cdc cli changefeed update --pd=$pd_addr --config="$WORK_DIR/changefeed.toml" --no-confirm --changefeed-id $uuid)
@@ -85,7 +85,7 @@ EOF
 	# Update changefeed
 	run_cdc_cli changefeed update --pd=$pd_addr --config="$WORK_DIR/changefeed.toml" --no-confirm --changefeed-id $uuid
 	changefeed_info=$(curl -X GET "http://127.0.0.1:8300/api/v2/changefeeds/$uuid/meta_info" 2>&1)
-	if [[ ! $changefeed_info == *"\"case_sensitive\":false"* ]]; then
+	if [[ ! $changefeed_info == *"\"case_sensitive\":true"* ]]; then
 		echo "[$(date)] <<<<< changefeed info is not updated as expected ${changefeed_info} >>>>>"
 		exit 1
 	fi