Repository: pingcap/tiflow
Pull request #9944 (2023-10-23 12:02:25)

dm: reduce gSet.String() usage by using zeroGSet for checking empty
<!--
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/master/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]: @lichunzhu: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-dm-integration-test | ce16d1be52740095b0e2ef593365ff7f23d9be5f | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_integration_test/2211/display/redirect) | true | `/test dm-integration-test`

[Full PR test history](https://prow.tidb.net/pr-history?org=pingcap&repo=tiflow&pr=9944). [Your PR dashboard](https://prow.tidb.net/pr?query=is%3Apr%20state%3Aopen%20author%3Alichunzhu).

<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. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->

Patch:
diff --git a/.golangci.yml b/.golangci.yml
index 36205d71f..a60f59e4f 100644
--- a/.golangci.yml
+++ b/.golangci.yml
@@ -103,6 +103,8 @@ linters-settings:
     excludes:
       - G404
       - G601
+      - G108
+      - G114
 
 issues:
   exclude-rules:
diff --git a/Makefile b/Makefile
index 4d1c39caa..bf86de985 100644
--- a/Makefile
+++ b/Makefile
@@ -94,7 +94,7 @@ MAKE_FILES = $(shell find . \( -name 'Makefile' -o -name '*.mk' \) -print)
 
 RELEASE_VERSION =
 ifeq ($(RELEASE_VERSION),)
-	RELEASE_VERSION := v7.5.0-master
+	RELEASE_VERSION := v7.6.0-master
 	release_version_regex := ^v[0-9]\..*$$
 	release_branch_regex := "^release-[0-9]\.[0-9].*$$|^HEAD$$|^.*/*tags/v[0-9]\.[0-9]\..*$$"
 	ifneq ($(shell git rev-parse --abbrev-ref HEAD | grep -E $(release_branch_regex)),)
diff --git a/cdc/api/v2/model.go b/cdc/api/v2/model.go
index af9887b02..01c20515b 100644
--- a/cdc/api/v2/model.go
+++ b/cdc/api/v2/model.go
@@ -193,7 +193,8 @@ type ReplicaConfig struct {
 	Consistent                   *ConsistentConfig          `json:"consistent,omitempty"`
 	Scheduler                    *ChangefeedSchedulerConfig `json:"scheduler"`
 	Integrity                    *IntegrityConfig           `json:"integrity"`
-	ChangefeedErrorStuckDuration *JSONDuration              `json:"changefeed_error_stuck_duration,omitempty" swaggertype:"string"`
+	ChangefeedErrorStuckDuration *JSONDuration              `json:"changefeed_error_stuck_duration,omitempty"`
+	SQLMode                      string                     `json:"sql_mode,omitempty"`
 }
 
 // ToInternalReplicaConfig coverts *v2.ReplicaConfig into *config.ReplicaConfig
@@ -211,6 +212,7 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 	res.CheckGCSafePoint = c.CheckGCSafePoint
 	res.EnableSyncPoint = c.EnableSyncPoint
 	res.IgnoreIneligibleTable = c.IgnoreIneligibleTable
+	res.SQLMode = c.SQLMode
 	if c.SyncPointInterval != nil {
 		res.SyncPointInterval = &c.SyncPointInterval.duration
 	}
@@ -260,11 +262,12 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 	}
 	if c.Consistent != nil {
 		res.Consistent = &config.ConsistentConfig{
-			Level:             c.Consistent.Level,
-			MaxLogSize:        c.Consistent.MaxLogSize,
-			FlushIntervalInMs: c.Consistent.FlushIntervalInMs,
-			Storage:           c.Consistent.Storage,
-			UseFileBackend:    c.Consistent.UseFileBackend,
+			Level:                 c.Consistent.Level,
+			MaxLogSize:            c.Consistent.MaxLogSize,
+			FlushIntervalInMs:     c.Consistent.FlushIntervalInMs,
+			MetaFlushIntervalInMs: c.Consistent.MetaFlushIntervalInMs,
+			Storage:               c.Consistent.Storage,
+			UseFileBackend:        c.Consistent.UseFileBackend,
 		}
 	}
 	if c.Sink != nil {
@@ -497,6 +500,7 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 		CheckGCSafePoint:      cloned.CheckGCSafePoint,
 		EnableSyncPoint:       cloned.EnableSyncPoint,
 		BDRMode:               cloned.BDRMode,
+		SQLMode:               cloned.SQLMode,
 	}
 
 	if cloned.SyncPointInterval != nil {
@@ -741,11 +745,12 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 	}
 	if cloned.Consistent != nil {
 		res.Consistent = &ConsistentConfig{
-			Level:             cloned.Consistent.Level,
-			MaxLogSize:        cloned.Consistent.MaxLogSize,
-			FlushIntervalInMs: cloned.Consistent.FlushIntervalInMs,
-			Storage:           cloned.Consistent.Storage,
-			UseFileBackend:    cloned.Consistent.UseFileBackend,
+			Level:                 cloned.Consistent.Level,
+			MaxLogSize:            cloned.Consistent.MaxLogSize,
+			FlushIntervalInMs:     cloned.Consistent.FlushIntervalInMs,
+			MetaFlushIntervalInMs: cloned.Consistent.MetaFlushIntervalInMs,
+			Storage:               cloned.Consistent.Storage,
+			UseFileBackend:        cloned.Consistent.UseFileBackend,
 		}
 	}
 	if cloned.Mounter != nil {
@@ -933,11 +938,12 @@ type ColumnSelector struct {
 // ConsistentConfig represents replication consistency config for a changefeed
 // This is a duplicate of config.ConsistentConfig
 type ConsistentConfig struct {
-	Level             string `json:"level,omitempty"`
-	MaxLogSize        int64  `json:"max_log_size"`
-	FlushIntervalInMs int64  `json:"flush_interval"`
-	Storage           string `json:"storage,omitempty"`
-	UseFileBackend    bool   `json:"use_file_backend"`
+	Level                 string `json:"level,omitempty"`
+	MaxLogSize            int64  `json:"max_log_size"`
+	FlushIntervalInMs     int64  `json:"flush_interval"`
+	MetaFlushIntervalInMs int64  `json:"meta_flush_interval"`
+	Storage               string `json:"storage,omitempty"`
+	UseFileBackend        bool   `json:"use_file_backend"`
 }
 
 // ChangefeedSchedulerConfig is per changefeed scheduler settings.
diff --git a/cdc/kv/client.go b/cdc/kv/client.go
index f500e0c32..a71b6910a 100644
--- a/cdc/kv/client.go
+++ b/cdc/kv/client.go
@@ -19,6 +19,7 @@ import (
 	"io"
 	"math/rand"
 	"strconv"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -42,6 +43,7 @@ import (
 	"github.com/pingcap/tiflow/pkg/version"
 	"github.com/prometheus/client_golang/prometheus"
 	tidbkv "github.com/tikv/client-go/v2/kv"
+	"github.com/tikv/client-go/v2/oracle"
 	"github.com/tikv/client-go/v2/tikv"
 	pd "github.com/tikv/pd/client"
 	"go.uber.org/zap"
@@ -423,13 +425,11 @@ func (s *eventFeedSession) eventFeed(ctx context.Context) error {
 
 	g, ctx := errgroup.WithContext(ctx)
 
-	g.Go(func() error {
-		return s.dispatchRequest(ctx)
-	})
+	g.Go(func() error { return s.dispatchRequest(ctx) })
 
-	g.Go(func() error {
-		return s.requestRegionToStore(ctx, g)
-	})
+	g.Go(func() error { return s.requestRegionToStore(ctx, g) })
+
+	g.Go(func() error { return s.logSlowRegions(ctx) })
 
 	g.Go(func() error {
 		for {
@@ -689,13 +689,13 @@ func (s *eventFeedSession) requestRegionToStore(
 		state := newRegionFeedState(sri, requestID)
 		pendingRegions.setByRequestID(requestID, state)
 
-		log.Debug("start new request",
+		log.Info("start new request",
 			zap.String("namespace", s.changefeed.Namespace),
 			zap.String("changefeed", s.changefeed.ID),
 			zap.Int64("tableID", s.tableID),
 			zap.String("tableName", s.tableName),
-			zap.String("addr", storeAddr),
-			zap.Any("request", req))
+			zap.Uint64("regionID", sri.verID.GetID()),
+			zap.String("addr", storeAddr))
 
 		err = stream.client.Send(req)
 
@@ -889,6 +889,13 @@ func (s *eventFeedSession) handleError(ctx context.Context, errInfo regionErrorI
 	switch eerr := errors.Cause(err).(type) {
 	case *eventError:
 		innerErr := eerr.err
+		log.Info("cdc region error",
+			zap.String("namespace", s.changefeed.Namespace),
+			zap.String("changefeed", s.changefeed.ID),
+			zap.Int64("tableID", s.tableID),
+			zap.String("tableName", s.tableName),
+			zap.Stringer("error", innerErr))
+
 		if notLeader := innerErr.GetNotLeader(); notLeader != nil {
 			metricFeedNotLeaderCounter.Inc()
 			s.client.regionCache.UpdateLeader(errInfo.verID, notLeader.GetLeader(), errInfo.rpcCtx.AccessIdx)
@@ -1216,6 +1223,17 @@ func (s *eventFeedSession) sendRegionChangeEvents(
 			continue
 		}
 
+		switch x := event.Event.(type) {
+		case *cdcpb.Event_Error:
+			log.Info("event feed receives a region error",
+				zap.String("namespace", s.changefeed.Namespace),
+				zap.String("changefeed", s.changefeed.ID),
+				zap.Int64("tableID", s.tableID),
+				zap.String("tableName", s.tableName),
+				zap.Uint64("regionID", event.RegionId),
+				zap.Any("error", x.Error))
+		}
+
 		slot := worker.inputCalcSlot(event.RegionId)
 		statefulEvents[slot] = append(statefulEvents[slot], &regionStatefulEvent{
 			changeEvent: event,
@@ -1308,6 +1326,51 @@ func (s *eventFeedSession) getStreamCancel(storeAddr string) (cancel context.Can
 	return
 }
 
+func (s *eventFeedSession) logSlowRegions(ctx context.Context) error {
+	ticker := time.NewTicker(10 * time.Second)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+		}
+
+		currTime := s.client.pdClock.CurrentTime()
+		attr := s.rangeLock.CollectLockedRangeAttrs(nil)
+		if attr.SlowestRegion.Initialized {
+			ckptTime := oracle.GetTimeFromTS(attr.SlowestRegion.CheckpointTs)
+			if currTime.Sub(ckptTime) > 2*resolveLockMinInterval {
+				log.Info("event feed finds a slow region",
+					zap.String("namespace", s.changefeed.Namespace),
+					zap.String("changefeed", s.changefeed.ID),
+					zap.Int64("tableID", s.tableID),
+					zap.String("tableName", s.tableName),
+					zap.Any("slowRegion", attr.SlowestRegion))
+			}
+		} else if currTime.Sub(attr.SlowestRegion.Created) > 10*time.Minute {
+			log.Info("event feed initializes a region too slow",
+				zap.String("namespace", s.changefeed.Namespace),
+				zap.String("changefeed", s.changefeed.ID),
+				zap.Int64("tableID", s.tableID),
+				zap.String("tableName", s.tableName),
+				zap.Any("slowRegion", attr.SlowestRegion))
+		}
+		if len(attr.Holes) > 0 {
+			holes := make([]string, 0, len(attr.Holes))
+			for _, hole := range attr.Holes {
+				holes = append(holes, fmt.Sprintf("[%s,%s)", hole.StartKey, hole.EndKey))
+			}
+			log.Info("event feed holes exist",
+				zap.String("namespace", s.changefeed.Namespace),
+				zap.String("changefeed", s.changefeed.ID),
+				zap.Int64("tableID", s.tableID),
+				zap.String("tableName", s.tableName),
+				zap.String("holes", strings.Join(holes, ", ")))
+		}
+	}
+}
+
 func assembleRowEvent(regionID uint64, entry *cdcpb.Event_Row) (model.RegionFeedEvent, error) {
 	var opType model.OpType
 	switch entry.GetOpType() {
diff --git a/cdc/kv/region_state.go b/cdc/kv/region_state.go
index 0019f3bff..00cee2b85 100644
--- a/cdc/kv/region_state.go
+++ b/cdc/kv/region_state.go
@@ -16,7 +16,6 @@ package kv
 import (
 	"runtime"
 	"sync"
-	"sync/atomic"
 
 	"github.com/pingcap/tiflow/cdc/kv/regionlock"
 	"github.com/pingcap/tiflow/cdc/processor/tablepb"
@@ -67,13 +66,12 @@ type regionFeedState struct {
 	// stopped: some error happens.
 	// removed: the region is returned into the pending list,
 	//   will be re-resolved and re-scheduled later.
-	state atomic.Uint32
-
-	// All region errors should be handled in region workers.
-	// `err` is used to retrieve errors generated outside.
-	err struct {
-		sync.Mutex
-		e error
+	state struct {
+		sync.RWMutex
+		v uint32
+		// All region errors should be handled in region workers.
+		// `err` is used to retrieve errors generated outside.
+		err error
 	}
 }
 
@@ -90,30 +88,36 @@ func (s *regionFeedState) start() {
 
 // mark regionFeedState as stopped with the given error if possible.
 func (s *regionFeedState) markStopped(err error) {
-	if s.state.CompareAndSwap(stateNormal, stateStopped) {
-		if err != nil {
-			s.err.Lock()
-			defer s.err.Unlock()
-			s.err.e = err
-		}
+	s.state.Lock()
+	defer s.state.Unlock()
+	if s.state.v == stateNormal {
+		s.state.v = stateStopped
+		s.state.err = err
 	}
 }
 
 // mark regionFeedState as removed if possible.
 func (s *regionFeedState) markRemoved() (changed bool) {
-	return s.state.CompareAndSwap(stateStopped, stateRemoved)
+	s.state.Lock()
+	defer s.state.Unlock()
+	if s.state.v == stateStopped {
+		s.state.v = stateRemoved
+		changed = true
+	}
+	return
 }
 
 func (s *regionFeedState) isStale() bool {
-	state := s.state.Load()
-	return state == stateStopped || state == stateRemoved
+	s.state.RLock()
+	defer s.state.RUnlock()
+	return s.state.v == stateStopped || s.state.v == stateRemoved
 }
 
 func (s *regionFeedState) takeError() (err error) {
-	s.err.Lock()
-	defer s.err.Unlock()
-	err = s.err.e
-	s.err.e = nil
+	s.state.Lock()
+	defer s.state.Unlock()
+	err = s.state.err
+	s.state.err = nil
 	return
 }
 
diff --git a/cdc/kv/region_worker.go b/cdc/kv/region_worker.go
index 7a8d55227..b38639221 100644
--- a/cdc/kv/region_worker.go
+++ b/cdc/kv/region_worker.go
@@ -204,6 +204,7 @@ func (w *regionWorker) checkShouldExit() error {
 
 func (w *regionWorker) handleSingleRegionError(err error, state *regionFeedState) error {
 	regionID := state.getRegionID()
+	isStale := state.isStale()
 	log.Info("single region event feed disconnected",
 		zap.String("namespace", w.session.client.changefeed.Namespace),
 		zap.String("changefeed", w.session.client.changefeed.ID),
@@ -211,9 +212,10 @@ func (w *regionWorker) handleSingleRegionError(err error, state *regionFeedState
 		zap.Uint64("requestID", state.requestID),
 		zap.Stringer("span", &state.sri.span),
 		zap.Uint64("resolvedTs", state.sri.resolvedTs()),
+		zap.Bool("isStale", isStale),
 		zap.Error(err))
 	// if state is already marked stopped, it must have been or would be processed by `onRegionFail`
-	if state.isStale() {
+	if isStale {
 		return w.checkShouldExit()
 	}
 	// We need to ensure when the error is handled, `isStale` must be set. So set it before sending the error.
diff --git a/cdc/kv/shared_region_worker.go b/cdc/kv/shared_region_worker.go
index 1ec521b7e..83b780571 100644
--- a/cdc/kv/shared_region_worker.go
+++ b/cdc/kv/shared_region_worker.go
@@ -102,13 +102,20 @@ func (w *sharedRegionWorker) run(ctx context.Context) error {
 }
 
 func (w *sharedRegionWorker) handleSingleRegionError(state *regionFeedState, stream *requestedStream) {
-	if stream != nil {
-		// stream can be nil if it's obviously unnecessary to re-schedule the region.
-		stream.takeState(SubscriptionID(state.requestID), state.getRegionID())
+	stepsToRemoved := state.markRemoved()
+	err := state.takeError()
+	if err != nil {
+		log.Info("region worker get a region error",
+			zap.String("namespace", w.changefeed.Namespace),
+			zap.String("changefeed", w.changefeed.ID),
+			zap.Uint64("streamID", stream.streamID),
+			zap.Any("subscriptionID", state.getRegionID()),
+			zap.Uint64("regionID", state.sri.verID.GetID()),
+			zap.Bool("reschedule", stepsToRemoved),
+			zap.Error(err))
 	}
-	if state.markRemoved() {
-		// For SharedClient and SharedWorker, err will never be nil.
-		err := state.takeError()
+	if stepsToRemoved {
+		stream.takeState(SubscriptionID(state.requestID), state.getRegionID())
 		w.client.onRegionFail(newRegionErrorInfo(state.getRegionInfo(), err))
 	}
 }
@@ -128,16 +135,16 @@ func (w *sharedRegionWorker) processEvent(ctx context.Context, event statefulEve
 				w.handleSingleRegionError(state, event.stream)
 				return
 			}
-		case *cdcpb.Event_Admin_:
-		case *cdcpb.Event_Error:
-			state.markStopped(&eventError{err: x.Error})
-			w.handleSingleRegionError(state, event.stream)
-			return
 		case *cdcpb.Event_ResolvedTs:
 			w.handleResolvedTs(ctx, resolvedTsBatch{
 				ts:      x.ResolvedTs,
 				regions: []*regionFeedState{state},
 			})
+		case *cdcpb.Event_Error:
+			state.markStopped(&eventError{err: x.Error})
+			w.handleSingleRegionError(state, event.stream)
+			return
+		case *cdcpb.Event_Admin_:
 		}
 	} else if len(event.resolvedTsBatch.regions) > 0 {
 		w.handleResolvedTs(ctx, event.resolvedTsBatch)
diff --git a/cdc/kv/shared_stream.go b/cdc/kv/shared_stream.go
index ad3cc8402..5e13007f0 100644
--- a/cdc/kv/shared_stream.go
+++ b/cdc/kv/shared_stream.go
@@ -56,11 +56,8 @@ type tableExclusive struct {
 }
 
 func newStream(ctx context.Context, c *SharedClient, g *errgroup.Group, r *requestedStore) *requestedStream {
-	stream := &requestedStream{
-		streamID: streamIDGen.Add(1),
-		requests: chann.NewAutoDrainChann[singleRegionInfo](),
-	}
-	stream.requestedRegions.m = make(map[SubscriptionID]map[uint64]*regionFeedState)
+	stream := newRequestedStream(streamIDGen.Add(1))
+	stream.requests = chann.NewAutoDrainChann[singleRegionInfo]()
 
 	waitForPreFetching := func() error {
 		if stream.preFetchForConnecting != nil {
@@ -96,8 +93,8 @@ func newStream(ctx context.Context, c *SharedClient, g *errgroup.Group, r *reque
 			for _, m := range stream.clearStates() {
 				for _, state := range m {
 					state.markStopped(&sendRequestToStoreErr{})
+					sfEvent := newEventItem(nil, state, stream)
 					slot := hashRegionID(state.sri.verID.GetID(), len(c.workers))
-					sfEvent := statefulEvent{eventItem: eventItem{state: state}}
 					_ = c.workers[slot].sendEvent(ctx, sfEvent)
 				}
 			}
@@ -119,6 +116,12 @@ func newStream(ctx context.Context, c *SharedClient, g *errgroup.Group, r *reque
 	return stream
 }
 
+func newRequestedStream(streamID uint64) *requestedStream {
+	stream := &requestedStream{streamID: streamID}
+	stream.requestedRegions.m = make(map[SubscriptionID]map[uint64]*regionFeedState)
+	return stream
+}
+
 func (s *requestedStream) run(ctx context.Context, c *SharedClient, rs *requestedStore) (canceled bool) {
 	isCanceled := func() bool {
 		select {
@@ -352,8 +355,8 @@ func (s *requestedStream) send(ctx context.Context, c *SharedClient, rs *request
 			//    and then those regions from the bad requestID will be unsubscribed finally.
 			for _, state := range s.takeStates(subscriptionID) {
 				state.markStopped(&sendRequestToStoreErr{})
+				sfEvent := newEventItem(nil, state, s)
 				slot := hashRegionID(state.sri.verID.GetID(), len(c.workers))
-				sfEvent := statefulEvent{eventItem: eventItem{state: state}}
 				if err = c.workers[slot].sendEvent(ctx, sfEvent); err != nil {
 					return errors.Trace(err)
 				}
@@ -423,6 +426,9 @@ func (s *requestedStream) takeState(subscriptionID SubscriptionID, regionID uint
 	if m, ok := s.requestedRegions.m[subscriptionID]; ok {
 		state = m[regionID]
 		delete(m, regionID)
+		if len(m) == 0 {
+			delete(s.requestedRegions.m, subscriptionID)
+		}
 	}
 	return
 }
@@ -468,7 +474,21 @@ func (s *requestedStream) sendRegionChangeEvents(
 		} else {
 			subscriptionID = tableSubID
 		}
-		if state := s.getState(subscriptionID, regionID); state != nil {
+
+		state := s.getState(subscriptionID, regionID)
+		switch x := event.Event.(type) {
+		case *cdcpb.Event_Error:
+			log.Info("event feed receives a region error",
+				zap.String("namespace", c.changefeed.Namespace),
+				zap.String("changefeed", c.changefeed.ID),
+				zap.Uint64("streamID", s.streamID),
+				zap.Any("subscriptionID", subscriptionID),
+				zap.Uint64("regionID", event.RegionId),
+				zap.Bool("stateIsNil", state == nil),
+				zap.Any("error", x.Error))
+		}
+
+		if state != nil {
 			sfEvent := newEventItem(event, state, s)
 			slot := hashRegionID(regionID, len(c.workers))
 			if err := c.workers[slot].sendEvent(ctx, sfEvent); err != nil {
diff --git a/cdc/model/changefeed.go b/cdc/model/changefeed.go
index fec7bbed7..fbf942595 100644
--- a/cdc/model/changefeed.go
+++ b/cdc/model/changefeed.go
@@ -334,6 +334,9 @@ func (info *ChangeFeedInfo) VerifyAndComplete() {
 	if info.Config.ChangefeedErrorStuckDuration == nil {
 		info.Config.ChangefeedErrorStuckDuration = defaultConfig.ChangefeedErrorStuckDuration
 	}
+	if info.Config.SQLMode == "" {
+		info.Config.SQLMode = defaultConfig.SQLMode
+	}
 	info.RmUnusedFields()
 }
 
diff --git a/cdc/model/sink.go b/cdc/model/sink.go
index decb3d1e8..6f6a8afa3 100644
--- a/cdc/model/sink.go
+++ b/cdc/model/sink.go
@@ -604,20 +604,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 2370380eb..1dd073911 100644
--- a/cdc/owner/changefeed.go
+++ b/cdc/owner/changefeed.go
@@ -369,6 +369,12 @@ func (c *changefeed) tick(ctx cdcContext.Context,
 	default:
 	}
 
+	if c.redoMetaMgr.Enabled() {
+		if !c.redoMetaMgr.Running() {
+			return 0, 0, nil
+		}
+	}
+
 	// TODO: pass table checkpointTs when we support concurrent process ddl
 	allPhysicalTables, barrier, err := c.ddlManager.tick(ctx, preCheckpointTs, nil)
 	if err != nil {
@@ -481,7 +487,9 @@ LOOP2:
 	}
 
 	checkpointTs := c.latestStatus.CheckpointTs
-	c.resolvedTs = checkpointTs
+	if c.resolvedTs == 0 {
+		c.resolvedTs = checkpointTs
+	}
 	minTableBarrierTs := c.latestStatus.MinTableBarrierTs
 
 	failpoint.Inject("NewChangefeedNoRetryError", func() {
@@ -608,13 +616,7 @@ LOOP2:
 	}
 	c.observerLastTick = atomic.NewTime(time.Time{})
 
-	c.redoDDLMgr, err = redo.NewDDLManager(cancelCtx, c.id, c.latestInfo.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.latestInfo.Config.Consistent, ddlStartTs)
 	if c.redoDDLMgr.Enabled() {
 		c.wg.Add(1)
 		go func() {
@@ -623,14 +625,8 @@ LOOP2:
 		}()
 	}
 
-	c.redoMetaMgr, err = redo.NewMetaManagerWithInit(cancelCtx,
-		c.id,
-		c.latestInfo.Config.Consistent, checkpointTs)
-	if err != nil {
-		return err
-	}
+	c.redoMetaMgr = redo.NewMetaManager(c.id, c.latestInfo.Config.Consistent, checkpointTs)
 	if c.redoMetaMgr.Enabled() {
-		c.resolvedTs = c.redoMetaMgr.GetFlushedMeta().ResolvedTs
 		c.wg.Add(1)
 		go func() {
 			defer c.wg.Done()
@@ -748,7 +744,6 @@ func (c *changefeed) releaseResources(ctx cdcContext.Context) {
 	c.barriers = nil
 	c.initialized = false
 	c.isReleased = true
-	c.resolvedTs = 0
 
 	log.Info("changefeed closed",
 		zap.String("namespace", c.id.Namespace),
@@ -798,15 +793,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.id, cfInfo.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, cfInfo.Config.Consistent, 0)
 		}
 		err := c.redoMetaMgr.Cleanup(ctx)
 		if err != nil {
diff --git a/cdc/owner/ddl_sink.go b/cdc/owner/ddl_sink.go
index 5bdb8a8ca..f06d01e6f 100644
--- a/cdc/owner/ddl_sink.go
+++ b/cdc/owner/ddl_sink.go
@@ -25,6 +25,7 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/parser"
 	"github.com/pingcap/tidb/parser/format"
+	"github.com/pingcap/tidb/parser/mysql"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/sink/ddlsink"
 	"github.com/pingcap/tiflow/cdc/sink/ddlsink/factory"
@@ -426,7 +427,18 @@ func (s *ddlSinkImpl) close(ctx context.Context) (err error) {
 
 // addSpecialComment translate tidb feature to comment
 func (s *ddlSinkImpl) addSpecialComment(ddl *model.DDLEvent) (string, error) {
-	stms, _, err := parser.New().Parse(ddl.Query, ddl.Charset, ddl.Collate)
+	p := parser.New()
+	// We need to use the correct SQL mode to parse the DDL query.
+	// Otherwise, the parser may fail to parse the DDL query.
+	// For example, it is needed to parse the following DDL query:
+	//  `alter table "t" add column "c" int default 1;`
+	// by adding `ANSI_QUOTES` to the SQL mode.
+	mode, err := mysql.GetSQLMode(s.info.Config.SQLMode)
+	if err != nil {
+		return "", errors.Trace(err)
+	}
+	p.SetSQLMode(mode)
+	stms, _, err := p.Parse(ddl.Query, ddl.Charset, ddl.Collate)
 	if err != nil {
 		return "", errors.Trace(err)
 	}
diff --git a/cdc/processor/processor.go b/cdc/processor/processor.go
index 5a148240f..58205b110 100644
--- a/cdc/processor/processor.go
+++ b/cdc/processor/processor.go
@@ -166,13 +166,12 @@ func (p *processor) AddTableSpan(
 			// table is `prepared`, and a `isPrepare = false` request indicate that old table should
 			// be stopped on original capture already, it's safe to start replicating data now.
 			if !isPrepare {
-				redoStartTs := checkpoint.ResolvedTs
 				if p.redo.r.Enabled() {
 					// ResolvedTs is store in external storage when redo log is enabled, so we need to
 					// start table with ResolvedTs in redoDMLManager.
-					p.redo.r.StartTable(span, redoStartTs)
+					p.redo.r.StartTable(span, checkpoint.ResolvedTs)
 				}
-				if err := p.sinkManager.r.StartTable(span, startTs, redoStartTs); err != nil {
+				if err := p.sinkManager.r.StartTable(span, startTs); err != nil {
 					return false, errors.Trace(err)
 				}
 			}
@@ -606,10 +605,7 @@ func (p *processor) lazyInitImpl(etcdCtx cdcContext.Context) (err error) {
 	}
 	p.latestInfo.Config.Sink.TiDBSourceID = sourceID
 
-	p.redo.r, err = redo.NewDMLManager(prcCtx, p.changefeedID, p.latestInfo.Config.Consistent)
-	if err != nil {
-		return err
-	}
+	p.redo.r = redo.NewDMLManager(p.changefeedID, p.latestInfo.Config.Consistent)
 	p.redo.name = "RedoManager"
 	p.redo.changefeedID = p.changefeedID
 	p.redo.spawn(prcCtx)
diff --git a/cdc/processor/sinkmanager/manager.go b/cdc/processor/sinkmanager/manager.go
index 76cf16854..74c551388 100644
--- a/cdc/processor/sinkmanager/manager.go
+++ b/cdc/processor/sinkmanager/manager.go
@@ -821,11 +821,7 @@ func (m *SinkManager) AddTable(span tablepb.Span, startTs model.Ts, targetTs mod
 }
 
 // StartTable sets the table(TableSink) state to replicating.
-func (m *SinkManager) StartTable(
-	span tablepb.Span,
-	startTs model.Ts,
-	redoStartTs model.Ts,
-) error {
+func (m *SinkManager) StartTable(span tablepb.Span, startTs model.Ts) error {
 	log.Info("Start table sink",
 		zap.String("namespace", m.changefeedID.Namespace),
 		zap.String("changefeed", m.changefeedID.ID),
@@ -852,7 +848,7 @@ func (m *SinkManager) StartTable(
 	if m.redoDMLMgr != nil {
 		m.redoProgressHeap.push(&progress{
 			span:              span,
-			nextLowerBoundPos: engine.Position{StartTs: 0, CommitTs: redoStartTs + 1},
+			nextLowerBoundPos: engine.Position{StartTs: 0, CommitTs: startTs + 1},
 			version:           tableSink.(*tableSinkWrapper).version,
 		})
 	}
diff --git a/cdc/redo/manager.go b/cdc/redo/manager.go
index 0fc954489..21c2e2457 100644
--- a/cdc/redo/manager.go
+++ b/cdc/redo/manager.go
@@ -19,8 +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/model"
 	"github.com/pingcap/tiflow/cdc/processor/tablepb"
 	"github.com/pingcap/tiflow/cdc/redo/common"
@@ -65,20 +65,17 @@ func NewDisabledDDLManager() *ddlManager {
 
 // NewDDLManager creates a new ddl Manager.
 func NewDDLManager(
-	ctx context.Context, changefeedID model.ChangeFeedID,
+	changefeedID model.ChangeFeedID,
 	cfg *config.ConsistentConfig, ddlStartTs model.Ts,
-) (*ddlManager, error) {
-	logManager, err := newLogManager(ctx, changefeedID, cfg, redo.RedoDDLLogFileType)
-	if err != nil {
-		return nil, err
-	}
+) *ddlManager {
+	m := newLogManager(changefeedID, cfg, redo.RedoDDLLogFileType)
 	span := spanz.TableIDToComparableSpan(0)
-	logManager.AddTable(span, ddlStartTs)
+	m.AddTable(span, ddlStartTs)
 	return &ddlManager{
-		logManager: logManager,
-		// The current fakeSpan is meaningless, find a meaningful sapn in the future.
+		logManager: m,
+		// The current fakeSpan is meaningless, find a meaningful span in the future.
 		fakeSpan: span,
-	}, nil
+	}
 }
 
 type ddlManager struct {
@@ -115,14 +112,12 @@ type DMLManager interface {
 }
 
 // NewDMLManager creates a new dml Manager.
-func NewDMLManager(ctx context.Context, changefeedID model.ChangeFeedID,
+func NewDMLManager(changefeedID model.ChangeFeedID,
 	cfg *config.ConsistentConfig,
-) (*dmlManager, error) {
-	logManager, err := newLogManager(ctx, changefeedID, cfg, redo.RedoRowLogFileType)
-	if err != nil {
-		return nil, err
+) *dmlManager {
+	return &dmlManager{
+		logManager: newLogManager(changefeedID, cfg, redo.RedoRowLogFileType),
 	}
-	return &dmlManager{logManager: logManager}, nil
 }
 
 // NewDisabledDMLManager creates a disabled dml Manager.
@@ -228,29 +223,22 @@ type logManager struct {
 }
 
 func newLogManager(
-	ctx context.Context,
 	changefeedID model.ChangeFeedID,
 	cfg *config.ConsistentConfig, logType string,
-) (*logManager, error) {
+) *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
-	}
-	m := &logManager{
+	return &logManager{
 		enabled: true,
 		cfg: &writer.LogWriterConfig{
-			ConsistentConfig:   *cfg,
-			LogType:            logType,
-			CaptureID:          config.GetGlobalServerConfig().AdvertiseAddr,
-			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.NewAutoDrainChann[cacheEvents](),
 		rtsMap:    spanz.SyncMap{},
@@ -263,21 +251,38 @@ func newLogManager(
 		metricRedoWorkerBusyRatio: common.RedoWorkerBusyRatio.
 			WithLabelValues(changefeedID.Namespace, changefeedID.ID),
 	}
+}
+
+// Run implements pkg/util.Runnable.
+func (m *logManager) Run(ctx context.Context, _ ...chan<- error) error {
+	failpoint.Inject("ChangefeedNewRedoManagerError", func() {
+		failpoint.Return(errors.New("changefeed new redo manager injected error"))
+	})
+	if !m.Enabled() {
+		return nil
+	}
 
-	m.writer, err = factory.NewRedoLogWriter(ctx, m.cfg)
+	defer m.close()
+	start := time.Now()
+	w, err := factory.NewRedoLogWriter(ctx, m.cfg)
 	if err != nil {
-		return nil, err
+		log.Error("redo: failed to create redo log writer",
+			zap.String("namespace", m.cfg.ChangeFeedID.Namespace),
+			zap.String("changefeed", m.cfg.ChangeFeedID.ID),
+			zap.Duration("duration", time.Since(start)),
+			zap.Error(err))
+		return err
 	}
-	return m, nil
+	m.writer = w
+	return m.bgUpdateLog(ctx, m.getFlushDuration())
 }
 
-// Run implements pkg/util.Runnable.
-func (m *logManager) Run(ctx context.Context, _ ...chan<- error) error {
-	if m.Enabled() {
-		defer m.close()
-		return m.bgUpdateLog(ctx)
+func (m *logManager) getFlushDuration() time.Duration {
+	flushIntervalInMs := m.cfg.FlushIntervalInMs
+	if m.cfg.LogType == redo.RedoDDLLogFileType {
+		flushIntervalInMs = m.cfg.MetaFlushIntervalInMs
 	}
-	return nil
+	return time.Duration(flushIntervalInMs) * time.Millisecond
 }
 
 // WaitForReady implements pkg/util.Runnable.
@@ -486,15 +491,14 @@ func (m *logManager) onResolvedTsMsg(span tablepb.Span, resolvedTs model.Ts) {
 	}
 }
 
-func (m *logManager) bgUpdateLog(ctx context.Context) error {
+func (m *logManager) bgUpdateLog(ctx context.Context, flushDuration time.Duration) error {
 	m.releaseMemoryCbs = make([]func(), 0, 1024)
-	flushIntervalInMs := m.cfg.FlushIntervalInMs
-	ticker := time.NewTicker(time.Duration(flushIntervalInMs) * time.Millisecond)
+	ticker := time.NewTicker(flushDuration)
 	defer ticker.Stop()
 	log.Info("redo manager bgUpdateLog is running",
 		zap.String("namespace", m.cfg.ChangeFeedID.Namespace),
 		zap.String("changefeed", m.cfg.ChangeFeedID.ID),
-		zap.Int64("flushIntervalInMs", flushIntervalInMs))
+		zap.Duration("flushIntervalInMs", flushDuration))
 
 	var err error
 	// logErrCh is used to retrieve errors from log flushing goroutines.
@@ -549,11 +553,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 e0aefe30d..eb969aaa0 100644
--- a/cdc/redo/meta_manager.go
+++ b/cdc/redo/meta_manager.go
@@ -30,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"
 )
@@ -46,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 {
@@ -53,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
 
@@ -62,9 +69,13 @@ type metaManager struct {
 	uuidGenerator uuid.Generator
 	preMetaFile   string
 
+	startTs model.Ts
+
 	lastFlushTime          time.Time
-	flushIntervalInMs      int64
+	cfg                    *config.ConsistentConfig
 	metricFlushLogDuration prometheus.Observer
+
+	flushIntervalInMs int64
 }
 
 // NewDisabledMetaManager creates a disabled Meta Manager.
@@ -74,47 +85,13 @@ func NewDisabledMetaManager() *metaManager {
 	}
 }
 
-// NewMetaManagerWithInit creates a new Manager and initializes the meta.
-func NewMetaManagerWithInit(
-	ctx context.Context, changefeedID model.ChangeFeedID,
-	cfg *config.ConsistentConfig, startTs model.Ts,
-) (*metaManager, error) {
-	m, err := NewMetaManager(ctx, changefeedID, 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, changefeedID model.ChangeFeedID,
-	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{
@@ -122,46 +99,80 @@ func NewMetaManager(ctx context.Context, changefeedID model.ChangeFeedID,
 		changeFeedID:      changefeedID,
 		uuidGenerator:     uuid.NewGenerator(),
 		enabled:           true,
-		flushIntervalInMs: cfg.FlushIntervalInMs,
+		cfg:               cfg,
+		startTs:           checkpoint,
+		flushIntervalInMs: cfg.MetaFlushIntervalInMs,
 	}
 
-	uri, err := storage.ParseRawURL(cfg.Storage)
-	if err != nil {
-		return nil, err
-	}
-	if redo.IsBlackholeStorage(uri.Scheme) {
-		return m, nil
+	if m.flushIntervalInMs < redo.MinFlushIntervalInMs {
+		log.Warn("redo flush interval is too small, use default value",
+			zap.Int64("interval", m.flushIntervalInMs))
+		m.flushIntervalInMs = redo.DefaultMetaFlushIntervalInMs
 	}
+	return m
+}
 
+// 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 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, _ ...chan<- error) 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, m.cfg.FlushIntervalInMs)
 	})
 	eg.Go(func() error {
 		return m.bgGC(egCtx)
 	})
+
+	m.running.Store(true)
 	return eg.Wait()
 }
 
@@ -194,9 +205,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 +215,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 +230,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 +266,16 @@ 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 +335,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 +423,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 +437,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)
 		}
 	}
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 337379f3b..57b15ce45 100644
--- a/cdc/redo/writer/memory/encoding_worker.go
+++ b/cdc/redo/writer/memory/encoding_worker.go
@@ -124,7 +124,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..66a418730 100644
--- a/cdc/redo/writer/memory/mem_log_writer.go
+++ b/cdc/redo/writer/memory/mem_log_writer.go
@@ -52,13 +52,14 @@ func NewLogWriter(
 		return nil, errors.WrapError(errors.ErrRedoConfigInvalid,
 			errors.New("invalid LogWriterConfig"))
 	}
+
 	// "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/agent/table.go b/cdc/scheduler/internal/v3/agent/table.go
index be1f7ec92..e6a8525af 100644
--- a/cdc/scheduler/internal/v3/agent/table.go
+++ b/cdc/scheduler/internal/v3/agent/table.go
@@ -75,7 +75,7 @@ func (t *tableSpan) getTableSpanStatus(collectStat bool) tablepb.TableStatus {
 
 func newAddTableResponseMessage(status tablepb.TableStatus) *schedulepb.Message {
 	if status.Checkpoint.ResolvedTs < status.Checkpoint.CheckpointTs {
-		log.Panic("schedulerv3: resolved ts should not less than checkpoint ts",
+		log.Warn("schedulerv3: resolved ts should not less than checkpoint ts",
 			zap.Any("tableStatus", status),
 			zap.Any("checkpoint", status.Checkpoint.CheckpointTs),
 			zap.Any("resolved", status.Checkpoint.ResolvedTs))
@@ -100,7 +100,7 @@ func newRemoveTableResponseMessage(status tablepb.TableStatus) *schedulepb.Messa
 			// Advance resolved ts to checkpoint ts if table is removed.
 			status.Checkpoint.ResolvedTs = status.Checkpoint.CheckpointTs
 		} else {
-			log.Panic("schedulerv3: resolved ts should not less than checkpoint ts",
+			log.Warn("schedulerv3: resolved ts should not less than checkpoint ts",
 				zap.Any("tableStatus", status),
 				zap.Any("checkpoint", status.Checkpoint.CheckpointTs),
 				zap.Any("resolved", status.Checkpoint.ResolvedTs))
@@ -311,7 +311,7 @@ func newTableSpanManager(
 func (tm *tableSpanManager) poll(ctx context.Context) ([]*schedulepb.Message, error) {
 	result := make([]*schedulepb.Message, 0)
 	var err error
-	toBeDropped := []tablepb.Span{}
+	var toBeDropped []tablepb.Span
 	tm.tables.Ascend(func(span tablepb.Span, table *tableSpan) bool {
 		message, err1 := table.poll(ctx)
 		if err != nil {
diff --git a/cdc/scheduler/internal/v3/coordinator.go b/cdc/scheduler/internal/v3/coordinator.go
index 2d536eb23..e27bdf49c 100644
--- a/cdc/scheduler/internal/v3/coordinator.go
+++ b/cdc/scheduler/internal/v3/coordinator.go
@@ -334,7 +334,7 @@ func (c *coordinator) poll(
 	currentSpans := c.reconciler.Reconcile(
 		ctx, &c.tableRanges, replications, c.captureM.Captures, c.compat)
 	allTasks := c.schedulerM.Schedule(
-		checkpointTs, currentSpans, c.captureM.Captures, replications, runningTasks, c.redoMetaManager)
+		checkpointTs, currentSpans, c.captureM.Captures, replications, runningTasks)
 
 	// Handle generated schedule tasks.
 	msgs, err = c.replicationM.HandleTasks(allTasks)
diff --git a/cdc/scheduler/internal/v3/replication/replication_manager.go b/cdc/scheduler/internal/v3/replication/replication_manager.go
index 416aaf4b2..0ec7c7a67 100644
--- a/cdc/scheduler/internal/v3/replication/replication_manager.go
+++ b/cdc/scheduler/internal/v3/replication/replication_manager.go
@@ -78,14 +78,14 @@ func (t MoveTable) String() string {
 
 // AddTable is a schedule task for adding a table.
 type AddTable struct {
-	Span       tablepb.Span
-	CaptureID  model.CaptureID
-	Checkpoint tablepb.Checkpoint
+	Span         tablepb.Span
+	CaptureID    model.CaptureID
+	CheckpointTs model.Ts
 }
 
 func (t AddTable) String() string {
-	return fmt.Sprintf("AddTable, span: %s, capture: %s, checkpointTs: %d, resolvedTs: %d",
-		t.Span.String(), t.CaptureID, t.Checkpoint.CheckpointTs, t.Checkpoint.ResolvedTs)
+	return fmt.Sprintf("AddTable, span: %s, capture: %s, checkpointTs: %d",
+		t.Span.String(), t.CaptureID, t.CheckpointTs)
 }
 
 // RemoveTable is a schedule task for removing a table.
@@ -200,12 +200,7 @@ func (r *Manager) HandleCaptureChanges(
 		}
 		var err error
 		spanStatusMap.Ascend(func(span tablepb.Span, status map[string]*tablepb.TableStatus) bool {
-			checkpoint := tablepb.Checkpoint{
-				CheckpointTs: checkpointTs,
-				// Note that the real resolved ts is stored in the status.
-				ResolvedTs: checkpointTs,
-			}
-			table, err1 := NewReplicationSet(span, checkpoint, status, r.changefeedID)
+			table, err1 := NewReplicationSet(span, checkpointTs, status, r.changefeedID)
 			if err1 != nil {
 				err = errors.Trace(err1)
 				return false
@@ -442,7 +437,7 @@ func (r *Manager) handleAddTableTask(
 	var err error
 	table, ok := r.spans.Get(task.Span)
 	if !ok {
-		table, err = NewReplicationSet(task.Span, task.Checkpoint, nil, r.changefeedID)
+		table, err = NewReplicationSet(task.Span, task.CheckpointTs, nil, r.changefeedID)
 		if err != nil {
 			return nil, errors.Trace(err)
 		}
diff --git a/cdc/scheduler/internal/v3/replication/replication_set.go b/cdc/scheduler/internal/v3/replication/replication_set.go
index f62752f27..fec432499 100644
--- a/cdc/scheduler/internal/v3/replication/replication_set.go
+++ b/cdc/scheduler/internal/v3/replication/replication_set.go
@@ -141,7 +141,7 @@ type ReplicationSet struct { //nolint:revive
 // NewReplicationSet returns a new replication set.
 func NewReplicationSet(
 	span tablepb.Span,
-	checkpoint tablepb.Checkpoint,
+	checkpoint model.Ts,
 	tableStatus map[model.CaptureID]*tablepb.TableStatus,
 	changefeed model.ChangeFeedID,
 ) (*ReplicationSet, error) {
@@ -149,7 +149,10 @@ func NewReplicationSet(
 		Changefeed: changefeed,
 		Span:       span,
 		Captures:   make(map[string]Role),
-		Checkpoint: checkpoint,
+		Checkpoint: tablepb.Checkpoint{
+			CheckpointTs: checkpoint,
+			ResolvedTs:   checkpoint,
+		},
 	}
 	// Count of captures that is in Stopping states.
 	stoppingCount := 0
@@ -159,9 +162,7 @@ func NewReplicationSet(
 			return nil, r.inconsistentError(table, captureID,
 				"schedulerv3: table id inconsistent")
 		}
-		if err := r.updateCheckpointAndStats(table.Checkpoint, table.Stats); err != nil {
-			return nil, errors.Trace(err)
-		}
+		r.updateCheckpointAndStats(table.Checkpoint, table.Stats)
 
 		switch table.State {
 		case tablepb.TableStateReplicating:
@@ -497,8 +498,8 @@ func (r *ReplicationSet) pollOnPrepare(
 		}
 	case tablepb.TableStateReplicating:
 		if r.Primary == captureID {
-			err := r.updateCheckpointAndStats(input.Checkpoint, input.Stats)
-			return nil, false, err
+			r.updateCheckpointAndStats(input.Checkpoint, input.Stats)
+			return nil, false, nil
 		}
 	case tablepb.TableStateStopping, tablepb.TableStateStopped:
 		if r.Primary == captureID {
@@ -611,9 +612,7 @@ func (r *ReplicationSet) pollOnCommit(
 
 	case tablepb.TableStateStopped, tablepb.TableStateAbsent:
 		if r.Primary == captureID {
-			if err := r.updateCheckpointAndStats(input.Checkpoint, input.Stats); err != nil {
-				return nil, false, errors.Trace(err)
-			}
+			r.updateCheckpointAndStats(input.Checkpoint, input.Stats)
 			original := r.Primary
 			r.clearPrimary()
 			if !r.hasRole(RoleSecondary) {
@@ -685,9 +684,7 @@ func (r *ReplicationSet) pollOnCommit(
 
 	case tablepb.TableStateReplicating:
 		if r.Primary == captureID {
-			if err := r.updateCheckpointAndStats(input.Checkpoint, input.Stats); err != nil {
-				return nil, false, errors.Trace(err)
-			}
+			r.updateCheckpointAndStats(input.Checkpoint, input.Stats)
 			if r.hasRole(RoleSecondary) {
 				// Original primary is not stopped, ask for stopping.
 				return &schedulepb.Message{
@@ -722,8 +719,8 @@ func (r *ReplicationSet) pollOnCommit(
 
 	case tablepb.TableStateStopping:
 		if r.Primary == captureID && r.hasRole(RoleSecondary) {
-			err := r.updateCheckpointAndStats(input.Checkpoint, input.Stats)
-			return nil, false, err
+			r.updateCheckpointAndStats(input.Checkpoint, input.Stats)
+			return nil, false, nil
 		} else if r.isInRole(captureID, RoleUndetermined) {
 			log.Info("schedulerv3: capture is stopping during Commit",
 				zap.String("namespace", r.Changefeed.Namespace),
@@ -752,8 +749,8 @@ func (r *ReplicationSet) pollOnReplicating(
 	switch input.State {
 	case tablepb.TableStateReplicating:
 		if r.Primary == captureID {
-			err := r.updateCheckpointAndStats(input.Checkpoint, input.Stats)
-			return nil, false, err
+			r.updateCheckpointAndStats(input.Checkpoint, input.Stats)
+			return nil, false, nil
 		}
 		return nil, false, r.multiplePrimaryError(
 			input, captureID, "schedulerv3: multiple primary")
@@ -764,10 +761,7 @@ func (r *ReplicationSet) pollOnReplicating(
 	case tablepb.TableStateStopping:
 	case tablepb.TableStateStopped:
 		if r.Primary == captureID {
-			if err := r.updateCheckpointAndStats(input.Checkpoint, input.Stats); err != nil {
-				return nil, false, errors.Trace(err)
-			}
-
+			r.updateCheckpointAndStats(input.Checkpoint, input.Stats)
 			// 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",
@@ -997,7 +991,7 @@ func (r *ReplicationSet) handleCaptureShutdown(
 
 func (r *ReplicationSet) updateCheckpointAndStats(
 	checkpoint tablepb.Checkpoint, stats tablepb.Stats,
-) error {
+) {
 	if checkpoint.ResolvedTs < checkpoint.CheckpointTs {
 		log.Warn("schedulerv3: resolved ts should not less than checkpoint ts",
 			zap.String("namespace", r.Changefeed.Namespace),
@@ -1025,11 +1019,8 @@ func (r *ReplicationSet) updateCheckpointAndStats(
 			zap.Any("replicationSet", r),
 			zap.Any("checkpointTs", r.Checkpoint.CheckpointTs),
 			zap.Any("resolvedTs", r.Checkpoint.ResolvedTs))
-		return errors.ErrInvalidCheckpointTs.GenWithStackByArgs(r.Checkpoint.CheckpointTs,
-			r.Checkpoint.ResolvedTs)
 	}
 	r.Stats = stats
-	return nil
 }
 
 // SetHeap is a max-heap, it implements heap.Interface.
diff --git a/cdc/scheduler/internal/v3/scheduler/scheduler.go b/cdc/scheduler/internal/v3/scheduler/scheduler.go
index fd117849b..96088ec52 100644
--- a/cdc/scheduler/internal/v3/scheduler/scheduler.go
+++ b/cdc/scheduler/internal/v3/scheduler/scheduler.go
@@ -24,7 +24,7 @@ import (
 type scheduler interface {
 	Name() string
 	Schedule(
-		checkpoint tablepb.Checkpoint,
+		checkpointTs model.Ts,
 		currentSpans []tablepb.Span,
 		aliveCaptures map[model.CaptureID]*member.CaptureStatus,
 		replications *spanz.BtreeMap[*replication.ReplicationSet],
diff --git a/cdc/scheduler/internal/v3/scheduler/scheduler_balance.go b/cdc/scheduler/internal/v3/scheduler/scheduler_balance.go
index 26dce7da2..21c642a91 100644
--- a/cdc/scheduler/internal/v3/scheduler/scheduler_balance.go
+++ b/cdc/scheduler/internal/v3/scheduler/scheduler_balance.go
@@ -56,7 +56,7 @@ func (b *balanceScheduler) Name() string {
 }
 
 func (b *balanceScheduler) Schedule(
-	_ tablepb.Checkpoint,
+	_ model.Ts,
 	_ []tablepb.Span,
 	captures map[model.CaptureID]*member.CaptureStatus,
 	replications *spanz.BtreeMap[*replication.ReplicationSet],
diff --git a/cdc/scheduler/internal/v3/scheduler/scheduler_basic.go b/cdc/scheduler/internal/v3/scheduler/scheduler_basic.go
index bb6e613f9..2efffdbda 100644
--- a/cdc/scheduler/internal/v3/scheduler/scheduler_basic.go
+++ b/cdc/scheduler/internal/v3/scheduler/scheduler_basic.go
@@ -49,7 +49,7 @@ func (b *basicScheduler) Name() string {
 }
 
 func (b *basicScheduler) Schedule(
-	checkpoint tablepb.Checkpoint,
+	checkpointTs model.Ts,
 	currentSpans []tablepb.Span,
 	captures map[model.CaptureID]*member.CaptureStatus,
 	replications *spanz.BtreeMap[*replication.ReplicationSet],
@@ -102,7 +102,7 @@ func (b *basicScheduler) Schedule(
 			return tasks
 		}
 		tasks = append(
-			tasks, newBurstAddTables(b.changefeedID, checkpoint, newSpans, captureIDs))
+			tasks, newBurstAddTables(b.changefeedID, checkpointTs, newSpans, captureIDs))
 	}
 
 	// Build remove table tasks.
@@ -140,16 +140,16 @@ func (b *basicScheduler) Schedule(
 // newBurstAddTables add each new table to captures in a round-robin way.
 func newBurstAddTables(
 	changefeedID model.ChangeFeedID,
-	checkpoint tablepb.Checkpoint, newSpans []tablepb.Span, captureIDs []model.CaptureID,
+	checkpointTs model.Ts, newSpans []tablepb.Span, captureIDs []model.CaptureID,
 ) *replication.ScheduleTask {
 	idx := 0
 	tables := make([]replication.AddTable, 0, len(newSpans))
 	for _, span := range newSpans {
 		targetCapture := captureIDs[idx]
 		tables = append(tables, replication.AddTable{
-			Span:       span,
-			CaptureID:  targetCapture,
-			Checkpoint: checkpoint,
+			Span:         span,
+			CaptureID:    targetCapture,
+			CheckpointTs: checkpointTs,
 		})
 		log.Info("schedulerv3: burst add table",
 			zap.String("namespace", changefeedID.Namespace),
diff --git a/cdc/scheduler/internal/v3/scheduler/scheduler_drain_capture.go b/cdc/scheduler/internal/v3/scheduler/scheduler_drain_capture.go
index a1995a585..66de141b7 100644
--- a/cdc/scheduler/internal/v3/scheduler/scheduler_drain_capture.go
+++ b/cdc/scheduler/internal/v3/scheduler/scheduler_drain_capture.go
@@ -71,7 +71,7 @@ func (d *drainCaptureScheduler) setTarget(target model.CaptureID) bool {
 }
 
 func (d *drainCaptureScheduler) Schedule(
-	_ tablepb.Checkpoint,
+	_ model.Ts,
 	_ []tablepb.Span,
 	captures map[model.CaptureID]*member.CaptureStatus,
 	replications *spanz.BtreeMap[*replication.ReplicationSet],
diff --git a/cdc/scheduler/internal/v3/scheduler/scheduler_manager.go b/cdc/scheduler/internal/v3/scheduler/scheduler_manager.go
index 21ab13b3e..76817b344 100644
--- a/cdc/scheduler/internal/v3/scheduler/scheduler_manager.go
+++ b/cdc/scheduler/internal/v3/scheduler/scheduler_manager.go
@@ -20,7 +20,6 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/processor/tablepb"
-	"github.com/pingcap/tiflow/cdc/redo"
 	"github.com/pingcap/tiflow/cdc/scheduler/internal/v3/member"
 	"github.com/pingcap/tiflow/cdc/scheduler/internal/v3/replication"
 	"github.com/pingcap/tiflow/pkg/config"
@@ -70,22 +69,7 @@ func (sm *Manager) Schedule(
 	aliveCaptures map[model.CaptureID]*member.CaptureStatus,
 	replications *spanz.BtreeMap[*replication.ReplicationSet],
 	runTasking *spanz.BtreeMap[*replication.ScheduleTask],
-	redoMetaManager redo.MetaManager,
 ) []*replication.ScheduleTask {
-	checkpoint := tablepb.Checkpoint{
-		CheckpointTs: checkpointTs,
-		ResolvedTs:   checkpointTs,
-	}
-	if redoMetaManager != nil && redoMetaManager.Enabled() {
-		flushedMeta := redoMetaManager.GetFlushedMeta()
-		if flushedMeta.ResolvedTs < checkpointTs {
-			log.Panic("schedulerv3: flushed resolved ts is less than checkpoint ts",
-				zap.Uint64("checkpointTs", checkpointTs),
-				zap.Any("flushedMeta", flushedMeta))
-		}
-		checkpoint.ResolvedTs = flushedMeta.ResolvedTs
-	}
-
 	for sid, scheduler := range sm.schedulers {
 		// Basic scheduler bypasses max task check, because it handles the most
 		// critical scheduling, e.g. add table via CREATE TABLE DDL.
@@ -96,7 +80,7 @@ func (sm *Manager) Schedule(
 				return nil
 			}
 		}
-		tasks := scheduler.Schedule(checkpoint, currentSpans, aliveCaptures, replications)
+		tasks := scheduler.Schedule(checkpointTs, currentSpans, aliveCaptures, replications)
 		for _, t := range tasks {
 			name := struct {
 				scheduler, task string
diff --git a/cdc/scheduler/internal/v3/scheduler/scheduler_move_table.go b/cdc/scheduler/internal/v3/scheduler/scheduler_move_table.go
index bf9ef491f..10b44de77 100644
--- a/cdc/scheduler/internal/v3/scheduler/scheduler_move_table.go
+++ b/cdc/scheduler/internal/v3/scheduler/scheduler_move_table.go
@@ -67,7 +67,7 @@ func (m *moveTableScheduler) addTask(span tablepb.Span, target model.CaptureID)
 }
 
 func (m *moveTableScheduler) Schedule(
-	_ tablepb.Checkpoint,
+	_ model.Ts,
 	currentSpans []tablepb.Span,
 	captures map[model.CaptureID]*member.CaptureStatus,
 	replications *spanz.BtreeMap[*replication.ReplicationSet],
diff --git a/cdc/scheduler/internal/v3/scheduler/scheduler_rebalance.go b/cdc/scheduler/internal/v3/scheduler/scheduler_rebalance.go
index 2bb4a4ecb..ecc02d37c 100644
--- a/cdc/scheduler/internal/v3/scheduler/scheduler_rebalance.go
+++ b/cdc/scheduler/internal/v3/scheduler/scheduler_rebalance.go
@@ -51,7 +51,7 @@ func (r *rebalanceScheduler) Name() string {
 }
 
 func (r *rebalanceScheduler) Schedule(
-	_ tablepb.Checkpoint,
+	_ model.Ts,
 	currentSpans []tablepb.Span,
 	captures map[model.CaptureID]*member.CaptureStatus,
 	replications *spanz.BtreeMap[*replication.ReplicationSet],
diff --git a/cdc/sink/dmlsink/mq/manager/kafka_manager.go b/cdc/sink/dmlsink/mq/manager/kafka_manager.go
index ea8c3991d..bd0af80fd 100644
--- a/cdc/sink/dmlsink/mq/manager/kafka_manager.go
+++ b/cdc/sink/dmlsink/mq/manager/kafka_manager.go
@@ -40,6 +40,8 @@ const (
 type kafkaTopicManager struct {
 	changefeedID model.ChangeFeedID
 
+	defaultTopic string
+
 	admin kafka.ClusterAdminClient
 
 	cfg *kafka.AutoCreateTopicConfig
@@ -55,11 +57,13 @@ type kafkaTopicManager struct {
 // NewKafkaTopicManager creates a new topic manager.
 func NewKafkaTopicManager(
 	ctx context.Context,
+	defaultTopic string,
 	changefeedID model.ChangeFeedID,
 	admin kafka.ClusterAdminClient,
 	cfg *kafka.AutoCreateTopicConfig,
 ) *kafkaTopicManager {
 	mgr := &kafkaTopicManager{
+		defaultTopic:      defaultTopic,
 		changefeedID:      changefeedID,
 		admin:             admin,
 		cfg:               cfg,
@@ -165,6 +169,15 @@ func (m *kafkaTopicManager) fetchAllTopicsPartitionsNum(
 		return nil, 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.
+	_, ok := numPartitions[m.defaultTopic]
+	if ok {
+		numPartitions[m.defaultTopic] = m.cfg.PartitionNum
+	}
+
 	log.Info(
 		"Kafka admin client describe topics success",
 		zap.String("namespace", m.changefeedID.Namespace),
@@ -271,8 +284,12 @@ func (m *kafkaTopicManager) CreateTopicAndWaitUntilVisible(
 		return 0, errors.Trace(err)
 	}
 	if detail, ok := topicDetails[topicName]; ok {
-		m.tryUpdatePartitionsAndLogging(topicName, detail.NumPartitions)
-		return detail.NumPartitions, nil
+		numPartition := detail.NumPartitions
+		if topicName == m.defaultTopic {
+			numPartition = m.cfg.PartitionNum
+		}
+		m.tryUpdatePartitionsAndLogging(topicName, numPartition)
+		return numPartition, nil
 	}
 
 	partitionNum, err := m.createTopic(ctx, topicName)
diff --git a/cdc/sink/util/helper.go b/cdc/sink/util/helper.go
index 4ff6db3e6..cec952cbe 100644
--- a/cdc/sink/util/helper.go
+++ b/cdc/sink/util/helper.go
@@ -97,7 +97,7 @@ func GetTopicManagerAndTryCreateTopic(
 	adminClient kafka.ClusterAdminClient,
 ) (manager.TopicManager, error) {
 	topicManager := manager.NewKafkaTopicManager(
-		ctx, changefeedID, adminClient, topicCfg,
+		ctx, topic, changefeedID, adminClient, topicCfg,
 	)
 
 	if _, err := topicManager.CreateTopicAndWaitUntilVisible(ctx, topic); err != nil {
diff --git a/cmd/kafka-consumer/main.go b/cmd/kafka-consumer/main.go
index b160d423b..48ad922e0 100644
--- a/cmd/kafka-consumer/main.go
+++ b/cmd/kafka-consumer/main.go
@@ -16,9 +16,12 @@ package main
 import (
 	"context"
 	"database/sql"
+	"errors"
 	"flag"
 	"fmt"
 	"math"
+	"net/http"
+	_ "net/http/pprof"
 	"net/url"
 	"os"
 	"os/signal"
@@ -31,8 +34,9 @@ import (
 	"time"
 
 	"github.com/IBM/sarama"
+	"github.com/edwingeng/deque"
 	"github.com/google/uuid"
-	"github.com/pingcap/errors"
+	cerror "github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/sink/ddlsink"
@@ -97,6 +101,8 @@ type consumerOption struct {
 
 	// upstreamTiDBDSN is the dsn of the upstream TiDB cluster
 	upstreamTiDBDSN string
+
+	enableProfiling bool
 }
 
 // Adjust the consumer option by the upstream uri passed in parameters.
@@ -193,10 +199,10 @@ func (o *consumerOption) Adjust(upstreamURI *url.URL, configFile string) error {
 		replicaConfig.Sink.Protocol = util.AddressOf(o.protocol.String())
 		err := cmdUtil.StrictDecodeFile(configFile, "kafka consumer", replicaConfig)
 		if err != nil {
-			return errors.Trace(err)
+			return cerror.Trace(err)
 		}
 		if _, err := filter.VerifyTableRules(replicaConfig.Filter); err != nil {
-			return errors.Trace(err)
+			return cerror.Trace(err)
 		}
 		o.replicaConfig = replicaConfig
 	}
@@ -238,6 +244,7 @@ func main() {
 	flag.StringVar(&consumerOption.ca, "ca", "", "CA certificate path for Kafka SSL connection")
 	flag.StringVar(&consumerOption.cert, "cert", "", "Certificate path for Kafka SSL connection")
 	flag.StringVar(&consumerOption.key, "key", "", "Private key path for Kafka SSL connection")
+	flag.BoolVar(&consumerOption.enableProfiling, "enable-profiling", false, "enable pprof profiling")
 	flag.Parse()
 
 	err := logutil.InitLogger(&logutil.Config{
@@ -293,12 +300,22 @@ func main() {
 		log.Panic("Error creating consumer group client", zap.Error(err))
 	}
 
-	wg := &sync.WaitGroup{}
+	var wg sync.WaitGroup
+	if consumerOption.enableProfiling {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if err := http.ListenAndServe(":6060", nil); err != nil {
+				log.Panic("Error starting pprof", zap.Error(err))
+			}
+		}()
+	}
+
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
 		for {
-			// `Consume` should be called inside an infinite loop, when a
+			// `consume` should be called inside an infinite loop, when a
 			// server-side rebalance happens, the consumer session will need to be
 			// recreated to get the new claims
 			if err := client.Consume(ctx, strings.Split(consumerOption.topic, ","), consumer); err != nil {
@@ -342,19 +359,19 @@ func getPartitionNum(address []string, topic string, cfg *sarama.Config) (int32,
 	// get partition number or create topic automatically
 	admin, err := sarama.NewClusterAdmin(address, cfg)
 	if err != nil {
-		return 0, errors.Trace(err)
+		return 0, cerror.Trace(err)
 	}
 	topics, err := admin.ListTopics()
 	if err != nil {
-		return 0, errors.Trace(err)
+		return 0, cerror.Trace(err)
 	}
 	err = admin.Close()
 	if err != nil {
-		return 0, errors.Trace(err)
+		return 0, cerror.Trace(err)
 	}
 	topicDetail, exist := topics[topic]
 	if !exist {
-		return 0, errors.Errorf("can not find topic %s", topic)
+		return 0, cerror.Errorf("can not find topic %s", topic)
 	}
 	log.Info("get partition number of topic",
 		zap.String("topic", topic),
@@ -365,13 +382,13 @@ func getPartitionNum(address []string, topic string, cfg *sarama.Config) (int32,
 func waitTopicCreated(address []string, topic string, cfg *sarama.Config) error {
 	admin, err := sarama.NewClusterAdmin(address, cfg)
 	if err != nil {
-		return errors.Trace(err)
+		return cerror.Trace(err)
 	}
 	defer admin.Close()
 	for i := 0; i <= 30; i++ {
 		topics, err := admin.ListTopics()
 		if err != nil {
-			return errors.Trace(err)
+			return cerror.Trace(err)
 		}
 		if _, ok := topics[topic]; ok {
 			return nil
@@ -379,7 +396,7 @@ func waitTopicCreated(address []string, topic string, cfg *sarama.Config) error
 		log.Info("wait the topic created", zap.String("topic", topic))
 		time.Sleep(1 * time.Second)
 	}
-	return errors.Errorf("wait the topic(%s) created timeout", topic)
+	return cerror.Errorf("wait the topic(%s) created timeout", topic)
 }
 
 func newSaramaConfig(o *consumerOption) (*sarama.Config, error) {
@@ -387,7 +404,7 @@ func newSaramaConfig(o *consumerOption) (*sarama.Config, error) {
 
 	version, err := sarama.ParseKafkaVersion(o.version)
 	if err != nil {
-		return nil, errors.Trace(err)
+		return nil, cerror.Trace(err)
 	}
 
 	config.ClientID = "ticdc_kafka_sarama_consumer"
@@ -406,7 +423,7 @@ func newSaramaConfig(o *consumerOption) (*sarama.Config, error) {
 			KeyPath:  o.key,
 		}).ToTLSConfig()
 		if err != nil {
-			return nil, errors.Trace(err)
+			return nil, cerror.Trace(err)
 		}
 	}
 
@@ -419,6 +436,8 @@ type partitionSinks struct {
 	tableSinksMap     sync.Map
 	// resolvedTs record the maximum timestamp of the received event
 	resolvedTs uint64
+
+	flowController *flowController
 }
 
 // Consumer represents a Sarama consumer group consumer
@@ -450,6 +469,10 @@ type Consumer struct {
 	upstreamTiDB *sql.DB
 }
 
+const (
+	defaultMemoryQuotaInBytes = 2 * 1024 * 1024 * 1024 // 2GB
+)
+
 // NewConsumer creates a new cdc kafka consumer
 func NewConsumer(ctx context.Context, o *consumerOption) (*Consumer, error) {
 	c := new(Consumer)
@@ -457,7 +480,7 @@ func NewConsumer(ctx context.Context, o *consumerOption) (*Consumer, error) {
 
 	tz, err := util.GetTimezone(o.timezone)
 	if err != nil {
-		return nil, errors.Annotate(err, "can not load timezone")
+		return nil, cerror.Annotate(err, "can not load timezone")
 	}
 	config.GetGlobalServerConfig().TZ = o.timezone
 	c.tz = tz
@@ -488,7 +511,7 @@ func NewConsumer(ctx context.Context, o *consumerOption) (*Consumer, error) {
 	if o.replicaConfig != nil {
 		eventRouter, err := dispatcher.NewEventRouter(o.replicaConfig, o.protocol, o.topic, "kafka")
 		if err != nil {
-			return nil, errors.Trace(err)
+			return nil, cerror.Trace(err)
 		}
 		c.eventRouter = eventRouter
 	}
@@ -496,21 +519,28 @@ func NewConsumer(ctx context.Context, o *consumerOption) (*Consumer, error) {
 	c.sinks = make([]*partitionSinks, o.partitionNum)
 	ctx, cancel := context.WithCancel(ctx)
 	errChan := make(chan error, 1)
+
+	memoryQuotaPerPartition := defaultMemoryQuotaInBytes / int(o.partitionNum)
 	for i := 0; i < int(o.partitionNum); i++ {
-		c.sinks[i] = &partitionSinks{}
+		c.sinks[i] = &partitionSinks{
+			flowController: newFlowController(uint64(memoryQuotaPerPartition)),
+		}
 	}
+	log.Info("flow controller created for each partition",
+		zap.Int32("partitionNum", o.partitionNum),
+		zap.Int("quota", memoryQuotaPerPartition))
 
 	changefeedID := model.DefaultChangeFeedID("kafka-consumer")
 	f, err := eventsinkfactory.New(ctx, changefeedID, o.downstreamURI, config.GetDefaultReplicaConfig(), errChan)
 	if err != nil {
 		cancel()
-		return nil, errors.Trace(err)
+		return nil, cerror.Trace(err)
 	}
 	c.sinkFactory = f
 
 	go func() {
 		err := <-errChan
-		if errors.Cause(err) != context.Canceled {
+		if !errors.Is(cerror.Cause(err), context.Canceled) {
 			log.Error("error on running consumer", zap.Error(err))
 		} else {
 			log.Info("consumer exited")
@@ -521,7 +551,7 @@ func NewConsumer(ctx context.Context, o *consumerOption) (*Consumer, error) {
 	ddlSink, err := ddlsinkfactory.New(ctx, changefeedID, o.downstreamURI, config.GetDefaultReplicaConfig())
 	if err != nil {
 		cancel()
-		return nil, errors.Trace(err)
+		return nil, cerror.Trace(err)
 	}
 	c.ddlSink = ddlSink
 	c.ready = make(chan bool)
@@ -595,14 +625,14 @@ func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram
 	case config.ProtocolAvro:
 		schemaM, err := avro.NewConfluentSchemaManager(ctx, c.option.schemaRegistryURI, nil)
 		if err != nil {
-			return errors.Trace(err)
+			return cerror.Trace(err)
 		}
 		decoder = avro.NewDecoder(c.codecConfig, schemaM, c.option.topic, c.tz)
 	default:
 		log.Panic("Protocol not supported", zap.Any("Protocol", c.codecConfig.Protocol))
 	}
 	if err != nil {
-		return errors.Trace(err)
+		return cerror.Trace(err)
 	}
 
 	log.Info("start consume claim",
@@ -613,7 +643,7 @@ func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram
 	for message := range claim.Messages() {
 		if err := decoder.AddKeyValue(message.Key, message.Value); err != nil {
 			log.Error("add key value to the decoder failed", zap.Error(err))
-			return errors.Trace(err)
+			return cerror.Trace(err)
 		}
 
 		counter := 0
@@ -664,7 +694,7 @@ func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram
 				if c.eventRouter != nil {
 					target, _, err := c.eventRouter.GetPartitionForRowChange(row, c.option.partitionNum)
 					if err != nil {
-						return errors.Trace(err)
+						return cerror.Trace(err)
 					}
 					if partition != target {
 						log.Panic("RowChangedEvent dispatched to wrong partition",
@@ -706,6 +736,16 @@ func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram
 				group.Append(row)
 				// todo: mark the offset after the DDL is fully synced to the downstream mysql.
 				session.MarkMessage(message, "")
+
+				size := uint64(row.ApproximateBytes())
+				err = sink.flowController.consume(row.CommitTs, size)
+				if err != nil {
+					if errors.Is(err, errFlowControllerAborted) {
+						log.Info("flow control aborted")
+						return nil
+					}
+					return cerror.Trace(err)
+				}
 			case model.MessageTypeResolved:
 				ts, err := decoder.NextResolvedEvent()
 				if err != nil {
@@ -773,10 +813,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.
@@ -818,7 +859,7 @@ func (c *Consumer) forEachSink(fn func(sink *partitionSinks) error) error {
 	defer c.sinksMu.Unlock()
 	for _, sink := range c.sinks {
 		if err := fn(sink); err != nil {
-			return errors.Trace(err)
+			return cerror.Trace(err)
 		}
 	}
 	return nil
@@ -849,7 +890,7 @@ func (c *Consumer) Run(ctx context.Context) error {
 
 		minPartitionResolvedTs, err := c.getMinPartitionResolvedTs()
 		if err != nil {
-			return errors.Trace(err)
+			return cerror.Trace(err)
 		}
 
 		// handle DDL
@@ -859,12 +900,12 @@ func (c *Consumer) Run(ctx context.Context) error {
 			if err := c.forEachSink(func(sink *partitionSinks) error {
 				return syncFlushRowChangedEvents(ctx, sink, todoDDL.CommitTs)
 			}); err != nil {
-				return errors.Trace(err)
+				return cerror.Trace(err)
 			}
 
 			// DDL can be executed, do it first.
 			if err := c.ddlSink.WriteDDLEvent(ctx, todoDDL); err != nil {
-				return errors.Trace(err)
+				return cerror.Trace(err)
 			}
 			c.popDDL()
 
@@ -890,7 +931,7 @@ func (c *Consumer) Run(ctx context.Context) error {
 		if err := c.forEachSink(func(sink *partitionSinks) error {
 			return syncFlushRowChangedEvents(ctx, sink, c.globalResolvedTs)
 		}); err != nil {
-			return errors.Trace(err)
+			return cerror.Trace(err)
 		}
 	}
 }
@@ -914,9 +955,11 @@ func syncFlushRowChangedEvents(ctx context.Context, sink *partitionSinks, resolv
 				log.Error("Failed to update resolved ts", zap.Error(err))
 				return false
 			}
-			if !tableSink.(tablesink.TableSink).GetCheckpointTs().EqualOrGreater(resolvedTs) {
+			checkpoint := tableSink.(tablesink.TableSink).GetCheckpointTs()
+			if !checkpoint.EqualOrGreater(resolvedTs) {
 				flushedResolvedTs = false
 			}
+			sink.flowController.release(checkpoint.Ts)
 			return true
 		})
 		if flushedResolvedTs {
@@ -950,7 +993,7 @@ func openDB(ctx context.Context, dsn string) (*sql.DB, error) {
 	db, err := sql.Open("mysql", dsn)
 	if err != nil {
 		log.Error("open db failed", zap.Error(err))
-		return nil, errors.Trace(err)
+		return nil, cerror.Trace(err)
 	}
 
 	db.SetMaxOpenConns(10)
@@ -961,8 +1004,146 @@ func openDB(ctx context.Context, dsn string) (*sql.DB, error) {
 	defer cancel()
 	if err = db.PingContext(ctx); err != nil {
 		log.Error("ping db failed", zap.Error(err))
-		return nil, errors.Trace(err)
+		return nil, cerror.Trace(err)
 	}
 	log.Info("open db success", zap.String("dsn", dsn))
 	return db, nil
 }
+
+var (
+	errFlowControllerLargerThanQuota = errors.New("flow controller request memory larger than quota")
+	errFlowControllerAborted         = errors.New("flow controller aborted")
+)
+
+type memoryQuota struct {
+	quota uint64 // should not be changed once initialized
+
+	isAborted atomic.Bool
+
+	consumed struct {
+		sync.Mutex
+		bytes uint64
+	}
+
+	consumedCond *sync.Cond
+}
+
+// newMemoryQuota creates a new memoryQuota
+// quota: max advised memory consumption in bytes.
+func newMemoryQuota(quota uint64) *memoryQuota {
+	ret := &memoryQuota{
+		quota: quota,
+	}
+
+	ret.consumedCond = sync.NewCond(&ret.consumed)
+	return ret
+}
+
+// consumeWithBlocking is called when a hard-limit is needed. The method will
+// block until enough memory has been freed up by release.
+// blockCallBack will be called if the function will block.
+// Should be used with care to prevent deadlock.
+func (c *memoryQuota) consumeWithBlocking(nBytes uint64) error {
+	if nBytes >= c.quota {
+		return errFlowControllerLargerThanQuota
+	}
+
+	c.consumed.Lock()
+	defer c.consumed.Unlock()
+
+	for {
+		if c.isAborted.Load() {
+			return errFlowControllerAborted
+		}
+
+		newConsumed := c.consumed.bytes + nBytes
+		if newConsumed < c.quota {
+			break
+		}
+		c.consumedCond.Wait()
+	}
+
+	c.consumed.bytes += nBytes
+	return nil
+}
+
+// release is called when a chuck of memory is done being used.
+func (c *memoryQuota) release(nBytes uint64) {
+	c.consumed.Lock()
+
+	if c.consumed.bytes < nBytes {
+		c.consumed.Unlock()
+		log.Panic("memoryQuota: releasing more than consumed, report a bug",
+			zap.Uint64("consumed", c.consumed.bytes),
+			zap.Uint64("released", nBytes))
+	}
+
+	c.consumed.bytes -= nBytes
+	if c.consumed.bytes < c.quota {
+		c.consumed.Unlock()
+		c.consumedCond.Signal()
+		return
+	}
+
+	c.consumed.Unlock()
+}
+
+type flowController struct {
+	memoryQuota *memoryQuota
+
+	queueMu struct {
+		sync.Mutex
+		queue deque.Deque
+	}
+}
+
+type entry struct {
+	commitTs uint64
+	size     uint64
+}
+
+func newFlowController(quota uint64) *flowController {
+	return &flowController{
+		memoryQuota: newMemoryQuota(quota),
+		queueMu: struct {
+			sync.Mutex
+			queue deque.Deque
+		}{
+			queue: deque.NewDeque(),
+		},
+	}
+}
+
+func (c *flowController) consume(commitTs uint64, size uint64) error {
+	err := c.memoryQuota.consumeWithBlocking(size)
+	if err != nil {
+		return cerror.Trace(err)
+	}
+
+	c.queueMu.Lock()
+	defer c.queueMu.Unlock()
+
+	c.queueMu.queue.PushBack(&entry{
+		commitTs: commitTs,
+		size:     size,
+	})
+
+	return nil
+}
+
+func (c *flowController) release(resolvedTs uint64) {
+	var nBytesToRelease uint64
+
+	c.queueMu.Lock()
+	for c.queueMu.queue.Len() > 0 {
+		if peeked := c.queueMu.queue.Front().(*entry); peeked.commitTs <= resolvedTs {
+			nBytesToRelease += peeked.size
+			c.queueMu.queue.PopFront()
+		} else {
+			break
+		}
+	}
+	c.queueMu.Unlock()
+
+	c.memoryQuota.release(nBytesToRelease)
+}
diff --git a/dm/pkg/binlog/event/common.go b/dm/pkg/binlog/event/common.go
index e84dc8400..c5b6dfcb1 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.CheckGTIDSetEmpty(gSet) {
 		return nil, terror.ErrBinlogEmptyGTID.Generate()
 	}
 
diff --git a/dm/pkg/binlog/event/event.go b/dm/pkg/binlog/event/event.go
index daab72ff7..3273722a0 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.CheckGTIDSetEmpty(gSet) {
 		return nil, terror.ErrBinlogEmptyGTID.Generate()
 	}
 
diff --git a/dm/pkg/binlog/position.go b/dm/pkg/binlog/position.go
index 77ac1f990..ad548c8b9 100644
--- a/dm/pkg/binlog/position.go
+++ b/dm/pkg/binlog/position.go
@@ -365,18 +365,9 @@ func IsFreshPosition(location Location, flavor string, cmpGTID bool) bool {
 //	-1, true if gSet1 is less than gSet2
 //
 // but if can't compare gSet1 and gSet2, will returns 0, false.
-var (
-	emptyMySQLGTIDSet, _   = gmysql.ParseMysqlGTIDSet("")
-	emptyMariaDBGTIDSet, _ = gmysql.ParseMariadbGTIDSet("")
-)
-
-func CheckGTIDSetEmpty(gSet gmysql.GTIDSet) bool {
-	return gSet == nil || gSet.Equal(emptyMySQLGTIDSet) || gSet.Equal(emptyMariaDBGTIDSet)
-}
-
 func CompareGTID(gSet1, gSet2 gmysql.GTIDSet) (int, bool) {
-	gSetIsEmpty1 := CheckGTIDSetEmpty(gSet1)
-	gSetIsEmpty2 := CheckGTIDSetEmpty(gSet2)
+	gSetIsEmpty1 := gtid.CheckGTIDSetEmpty(gSet1)
+	gSetIsEmpty2 := gtid.CheckGTIDSetEmpty(gSet2)
 
 	switch {
 	case gSetIsEmpty1 && gSetIsEmpty2:
diff --git a/dm/pkg/gtid/gtid.go b/dm/pkg/gtid/gtid.go
index 98075a52d..4f0a51cd8 100644
--- a/dm/pkg/gtid/gtid.go
+++ b/dm/pkg/gtid/gtid.go
@@ -21,6 +21,16 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 )
 
+var (
+	emptyMySQLGTIDSet, _   = mysql.ParseMysqlGTIDSet("")
+	emptyMariaDBGTIDSet, _ = mysql.ParseMariadbGTIDSet("")
+)
+
+// CheckGTIDSetEmpty is used to check whether a GTID set is zero.
+func CheckGTIDSetEmpty(gSet mysql.GTIDSet) bool {
+	return gSet == nil || gSet.Equal(emptyMySQLGTIDSet) || gSet.Equal(emptyMariaDBGTIDSet)
+}
+
 // 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 +52,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 +88,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 +101,23 @@ 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"
+}
diff --git a/dm/syncer/checkpoint.go b/dm/syncer/checkpoint.go
index babaf9dd7..2128e986a 100644
--- a/dm/syncer/checkpoint.go
+++ b/dm/syncer/checkpoint.go
@@ -675,7 +675,7 @@ func (cp *RemoteCheckPoint) IsOlderThanTablePoint(table *filter.Table, location
 	}
 	oldLocation := point.MySQLLocation()
 	// if we update enable-gtid = false to true, we need to compare binlog position instead of GTID before we save table point
-	cmpGTID := cp.cfg.EnableGTID && !(binlog.CheckGTIDSetEmpty(oldLocation.GetGTID()) && binlog.ComparePosition(oldLocation.Position, binlog.MinPosition) > 0)
+	cmpGTID := cp.cfg.EnableGTID && !(gtid.CheckGTIDSetEmpty(oldLocation.GetGTID()) && binlog.ComparePosition(oldLocation.Position, binlog.MinPosition) > 0)
 	cp.logCtx.L().Debug("compare table location whether is newer", zap.Stringer("location", location), zap.Stringer("old location", oldLocation), zap.Bool("cmpGTID", cmpGTID))
 
 	return binlog.CompareLocation(location, oldLocation, cmpGTID) <= 0
diff --git a/dm/syncer/ddl.go b/dm/syncer/ddl.go
index a292824e0..5060aba78 100644
--- a/dm/syncer/ddl.go
+++ b/dm/syncer/ddl.go
@@ -1102,10 +1102,10 @@ func (ddl *DDLWorker) handleModifyColumn(qec *queryEventContext, info *ddlInfo,
 	switch {
 	case mysql.HasAutoIncrementFlag(oldCol.GetFlag()) && !mysql.HasAutoIncrementFlag(newCol.GetFlag()):
 		return bf.RemoveAutoIncrement, nil
-	case mysql.HasPriKeyFlag(oldCol.GetFlag()) != mysql.HasPriKeyFlag(newCol.GetFlag()):
-		return bf.ModifyPK, nil
-	case mysql.HasUniKeyFlag(oldCol.GetFlag()) != mysql.HasUniKeyFlag(newCol.GetFlag()):
-		return bf.ModifyUK, nil
+	case mysql.HasPriKeyFlag(oldCol.GetFlag()) && !mysql.HasPriKeyFlag(newCol.GetFlag()):
+		return bf.DropPrimaryKey, nil
+	case mysql.HasUniKeyFlag(oldCol.GetFlag()) && !mysql.HasUniKeyFlag(newCol.GetFlag()):
+		return bf.DropUniqueKey, nil
 	case oldCol.GetDefaultValue() != newCol.GetDefaultValue():
 		return bf.ModifyDefaultValue, nil
 	case oldCol.GetCharset() != newCol.GetCharset():
@@ -1115,7 +1115,7 @@ func (ddl *DDLWorker) handleModifyColumn(qec *queryEventContext, info *ddlInfo,
 	case spec.Position != nil && spec.Position.Tp != ast.ColumnPositionNone:
 		return bf.ModifyColumnsOrder, nil
 	case oldCol.Name.L != newCol.Name.L:
-		return bf.Rename, nil
+		return bf.RenameColumn, nil
 	default:
 		return bf.AlterTable, nil
 	}
@@ -1142,14 +1142,22 @@ func (ddl *DDLWorker) AstToDDLEvent(qec *queryEventContext, info *ddlInfo) (et b
 					ddl.logger.Warn("handle modify column failed", zap.Error(err))
 				}
 				return et
-			case ast.AlterTableRenameColumn, ast.AlterTableRenameIndex, ast.AlterTableRenameTable:
-				return bf.Rename
-			case ast.AlterTableDropColumn, ast.AlterTableDropIndex, ast.AlterTableDropPartition:
-				return bf.Drop
+			case ast.AlterTableRenameColumn:
+				return bf.RenameColumn
+			case ast.AlterTableRenameIndex:
+				return bf.RenameIndex
+			case ast.AlterTableRenameTable:
+				return bf.RenameTable
+			case ast.AlterTableDropColumn:
+				return bf.DropColumn
+			case ast.AlterTableDropIndex:
+				return bf.DropIndex
+			case ast.AlterTableDropPartition:
+				return bf.DropTablePartition
 			case ast.AlterTableDropPrimaryKey:
-				return bf.ModifyPK
+				return bf.DropPrimaryKey
 			case ast.AlterTableTruncatePartition:
-				return bf.Truncate
+				return bf.TruncateTablePartition
 			case ast.AlterTableAlterColumn:
 				return bf.ModifyDefaultValue
 			case ast.AlterTableAddConstraint:
diff --git a/dm/syncer/syncer.go b/dm/syncer/syncer.go
index 5e3e41405..ce24b532f 100644
--- a/dm/syncer/syncer.go
+++ b/dm/syncer/syncer.go
@@ -50,6 +50,7 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/conn"
 	tcontext "github.com/pingcap/tiflow/dm/pkg/context"
 	fr "github.com/pingcap/tiflow/dm/pkg/func-rollback"
+	"github.com/pingcap/tiflow/dm/pkg/gtid"
 	"github.com/pingcap/tiflow/dm/pkg/ha"
 	"github.com/pingcap/tiflow/dm/pkg/log"
 	parserpkg "github.com/pingcap/tiflow/dm/pkg/parser"
@@ -3497,7 +3498,7 @@ func (s *Syncer) adjustGlobalPointGTID(tctx *tcontext.Context) (bool, error) {
 	// 2. location already has GTID position
 	// 3. location is totally new, has no position info
 	// 4. location is too early thus not a COMMIT location, which happens when it's reset by other logic
-	if !s.cfg.EnableGTID || !binlog.CheckGTIDSetEmpty(location.GetGTID()) || location.Position.Name == "" || location.Position.Pos == 4 {
+	if !s.cfg.EnableGTID || !gtid.CheckGTIDSetEmpty(location.GetGTID()) || location.Position.Name == "" || location.Position.Pos == 4 {
 		return false, nil
 	}
 	// set enableGTID to false for new streamerController
diff --git a/go.mod b/go.mod
index ee2f2b3ac..adee3545f 100644
--- a/go.mod
+++ b/go.mod
@@ -69,7 +69,7 @@ require (
 	github.com/pingcap/kvproto v0.0.0-20230925123611-87bebcc0d071
 	github.com/pingcap/log v1.1.1-0.20230317032135-a0d097d16e22
 	github.com/pingcap/tidb v1.1.0-beta.0.20231010115255-ec2731b8f539
-	github.com/pingcap/tidb-tools v0.0.0-20231010140451-6189b1bea2fd
+	github.com/pingcap/tidb-tools v0.0.0-20231031074420-a020d2cdfc5a
 	github.com/pingcap/tidb/parser v0.0.0-20231010115255-ec2731b8f539
 	github.com/prometheus/client_golang v1.16.0
 	github.com/prometheus/client_model v0.4.0
diff --git a/go.sum b/go.sum
index 6c06a6def..cd211db13 100644
--- a/go.sum
+++ b/go.sum
@@ -1047,8 +1047,8 @@ github.com/pingcap/sysutil v1.0.1-0.20230407040306-fb007c5aff21/go.mod h1:QYnjfA
 github.com/pingcap/tidb v1.1.0-beta.0.20220511160835-98c31070d958/go.mod h1:luW4sIZoLHY3bCWuKqyqk2QgMvF+/M7nWOXf/me0+fY=
 github.com/pingcap/tidb v1.1.0-beta.0.20231010115255-ec2731b8f539 h1:Pw+tqB5BeNxn8uRQh6XZgU/MmdaSQiXtioStsxiu7M4=
 github.com/pingcap/tidb v1.1.0-beta.0.20231010115255-ec2731b8f539/go.mod h1:K9GzvaQZ9ht52B81kDXqvtW8p5rkQmP8rgJ8AW8yXCo=
-github.com/pingcap/tidb-tools v0.0.0-20231010140451-6189b1bea2fd h1:0BfNSfxiQgTD+YA4a3dmb7zR7nyVMOZB4qknn1y8t9s=
-github.com/pingcap/tidb-tools v0.0.0-20231010140451-6189b1bea2fd/go.mod h1:Awem9vQiEdvfdGLEbO6HWjR1jh2vFOCjKkwZdoHXfVI=
+github.com/pingcap/tidb-tools v0.0.0-20231031074420-a020d2cdfc5a h1:9MXDt1cSyG8eLDZj0dv1NZrT86BKjU8KxPalIS7KQRQ=
+github.com/pingcap/tidb-tools v0.0.0-20231031074420-a020d2cdfc5a/go.mod h1:Awem9vQiEdvfdGLEbO6HWjR1jh2vFOCjKkwZdoHXfVI=
 github.com/pingcap/tidb/parser v0.0.0-20211011031125-9b13dc409c5e/go.mod h1:e1MGCA9Sg3T8jid8PKAEq5eYVuMMCq4n8gJ+Kqp4Plg=
 github.com/pingcap/tidb/parser v0.0.0-20220511160835-98c31070d958/go.mod h1:ElJiub4lRy6UZDb+0JHDkGEdr6aOli+ykhyej7VCLoI=
 github.com/pingcap/tidb/parser v0.0.0-20231010115255-ec2731b8f539 h1:q9m7W3fJRAgmIwRckzxxrqWX/QRuQrXnVJSc5do1New=
diff --git a/pkg/config/consistent.go b/pkg/config/consistent.go
index 636edcf86..4a521f98d 100644
--- a/pkg/config/consistent.go
+++ b/pkg/config/consistent.go
@@ -24,11 +24,12 @@ import (
 
 // ConsistentConfig represents replication consistency config for a changefeed.
 type ConsistentConfig struct {
-	Level             string `toml:"level" json:"level"`
-	MaxLogSize        int64  `toml:"max-log-size" json:"max-log-size"`
-	FlushIntervalInMs int64  `toml:"flush-interval" json:"flush-interval"`
-	Storage           string `toml:"storage" json:"storage"`
-	UseFileBackend    bool   `toml:"use-file-backend" json:"use-file-backend"`
+	Level                 string `toml:"level" json:"level"`
+	MaxLogSize            int64  `toml:"max-log-size" json:"max-log-size"`
+	FlushIntervalInMs     int64  `toml:"flush-interval" json:"flush-interval"`
+	MetaFlushIntervalInMs int64  `toml:"meta-flush-interval" json:"meta-flush-interval"`
+	Storage               string `toml:"storage" json:"storage"`
+	UseFileBackend        bool   `toml:"use-file-backend" json:"use-file-backend"`
 }
 
 // ValidateAndAdjust validates the consistency config and adjusts it if necessary.
@@ -50,6 +51,15 @@ func (c *ConsistentConfig) ValidateAndAdjust() error {
 				c.FlushIntervalInMs, redo.MinFlushIntervalInMs))
 	}
 
+	if c.MetaFlushIntervalInMs == 0 {
+		c.MetaFlushIntervalInMs = redo.DefaultMetaFlushIntervalInMs
+	}
+	if c.MetaFlushIntervalInMs < redo.MinFlushIntervalInMs {
+		return cerror.ErrInvalidReplicaConfig.FastGenByArgs(
+			fmt.Sprintf("The consistent.meta-flush-interval:%d must be equal or greater than %d",
+				c.MetaFlushIntervalInMs, redo.MinFlushIntervalInMs))
+	}
+
 	uri, err := storage.ParseRawURL(c.Storage)
 	if err != nil {
 		return cerror.ErrInvalidReplicaConfig.GenWithStackByArgs(
diff --git a/pkg/config/replica_config.go b/pkg/config/replica_config.go
index dcf193c94..301d38547 100644
--- a/pkg/config/replica_config.go
+++ b/pkg/config/replica_config.go
@@ -23,6 +23,7 @@ import (
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
+	"github.com/pingcap/tidb/parser/mysql"
 	"github.com/pingcap/tiflow/pkg/config/outdated"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/integrity"
@@ -38,6 +39,12 @@ const (
 	// minSyncPointRetention is the minimum of SyncPointRetention can be set.
 	minSyncPointRetention           = time.Hour * 1
 	minChangeFeedErrorStuckDuration = time.Minute * 30
+	// The default SQL Mode of TiDB: "ONLY_FULL_GROUP_BY,
+	// STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,
+	// NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION"
+	// Note: The SQL Mode of TiDB is not the same as ORACLE.
+	// If you want to use the same SQL Mode as ORACLE, you need to add "ORACLE" to the SQL Mode.
+	defaultSQLMode = mysql.DefaultSQLMode
 )
 
 var defaultReplicaConfig = &ReplicaConfig{
@@ -72,11 +79,12 @@ var defaultReplicaConfig = &ReplicaConfig{
 		AdvanceTimeoutInSec:              util.AddressOf(DefaultAdvanceTimeoutInSec),
 	},
 	Consistent: &ConsistentConfig{
-		Level:             "none",
-		MaxLogSize:        redo.DefaultMaxLogSize,
-		FlushIntervalInMs: redo.DefaultFlushIntervalInMs,
-		Storage:           "",
-		UseFileBackend:    false,
+		Level:                 "none",
+		MaxLogSize:            redo.DefaultMaxLogSize,
+		FlushIntervalInMs:     redo.DefaultFlushIntervalInMs,
+		MetaFlushIntervalInMs: redo.DefaultMetaFlushIntervalInMs,
+		Storage:               "",
+		UseFileBackend:        false,
 	},
 	Scheduler: &ChangefeedSchedulerConfig{
 		EnableTableAcrossNodes: false,
@@ -88,6 +96,7 @@ var defaultReplicaConfig = &ReplicaConfig{
 		CorruptionHandleLevel: integrity.CorruptionHandleLevelWarn,
 	},
 	ChangefeedErrorStuckDuration: util.AddressOf(time.Minute * 30),
+	SQLMode:                      defaultSQLMode,
 }
 
 // GetDefaultReplicaConfig returns the default replica config.
@@ -139,6 +148,7 @@ type replicaConfig struct {
 	// Integrity is only available when the downstream is MQ.
 	Integrity                    *integrity.Config `toml:"integrity" json:"integrity"`
 	ChangefeedErrorStuckDuration *time.Duration    `toml:"changefeed-error-stuck-duration" json:"changefeed-error-stuck-duration,omitempty"`
+	SQLMode                      string            `toml:"sql-mode" json:"sql-mode"`
 }
 
 // Value implements the driver.Valuer interface
diff --git a/pkg/config/sink_protocol.go b/pkg/config/sink_protocol.go
index 0b4d41380..87c99c55b 100644
--- a/pkg/config/sink_protocol.go
+++ b/pkg/config/sink_protocol.go
@@ -38,6 +38,7 @@ const (
 	ProtocolCraft
 	ProtocolOpen
 	ProtocolCsv
+	ProtocolSimple
 )
 
 // IsBatchEncode returns whether the protocol is a batch encoder.
@@ -66,6 +67,8 @@ func ParseSinkProtocolFromString(protocol string) (Protocol, error) {
 		return ProtocolOpen, nil
 	case "csv":
 		return ProtocolCsv, nil
+	case "simple":
+		return ProtocolSimple, nil
 	default:
 		return ProtocolUnknown, cerror.ErrSinkUnknownProtocol.GenWithStackByArgs(protocol)
 	}
@@ -90,6 +93,8 @@ func (p Protocol) String() string {
 		return "open-protocol"
 	case ProtocolCsv:
 		return "csv"
+	case ProtocolSimple:
+		return "simple"
 	default:
 		panic("unreachable")
 	}
diff --git a/pkg/filter/expr_filter.go b/pkg/filter/expr_filter.go
index 8bae5d2e9..6015dda85 100644
--- a/pkg/filter/expr_filter.go
+++ b/pkg/filter/expr_filter.go
@@ -14,6 +14,7 @@
 package filter
 
 import (
+	"fmt"
 	"strings"
 	"sync"
 
@@ -21,6 +22,7 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/expression"
 	"github.com/pingcap/tidb/parser"
+	"github.com/pingcap/tidb/parser/mysql"
 	"github.com/pingcap/tidb/planner/core"
 	"github.com/pingcap/tidb/sessionctx"
 	"github.com/pingcap/tidb/types"
@@ -77,10 +79,16 @@ func newExprFilterRule(
 
 // verifyAndInitRule will verify and init the rule.
 // It should only be called in dmlExprFilter's verify method.
-func (r *dmlExprFilterRule) verify(tableInfos []*model.TableInfo) error {
+func (r *dmlExprFilterRule) verify(tableInfos []*model.TableInfo, sqlMode string) error {
 	// verify expression filter rule syntax.
 	p := parser.New()
-	_, _, err := p.ParseSQL(completeExpression(r.config.IgnoreInsertValueExpr))
+	mode, err := mysql.GetSQLMode(sqlMode)
+	if err != nil {
+		log.Error("failed to get sql mode", zap.Error(err))
+		return cerror.ErrInvalidReplicaConfig.FastGenByArgs(fmt.Sprintf("invalid sqlMode %s", sqlMode))
+	}
+	p.SetSQLMode(mode)
+	_, _, err = p.ParseSQL(completeExpression(r.config.IgnoreInsertValueExpr))
 	if err != nil {
 		log.Error("failed to parse expression", zap.Error(err))
 		return cerror.ErrExpressionParseFailed.
@@ -347,14 +355,18 @@ func getColumnFromError(err error) string {
 
 // dmlExprFilter is a filter that filters DML events by SQL expression.
 type dmlExprFilter struct {
-	rules []*dmlExprFilterRule
+	rules   []*dmlExprFilterRule
+	sqlMODE string
 }
 
 func newExprFilter(
 	timezone string,
 	cfg *config.FilterConfig,
+	sqlMODE string,
 ) (*dmlExprFilter, error) {
-	res := &dmlExprFilter{}
+	res := &dmlExprFilter{
+		sqlMODE: sqlMODE,
+	}
 	sessCtx := utils.NewSessionCtx(map[string]string{
 		"time_zone": timezone,
 	})
@@ -382,7 +394,7 @@ func (f *dmlExprFilter) addRule(
 // verify checks if all rules in this filter is valid.
 func (f *dmlExprFilter) verify(tableInfos []*model.TableInfo) error {
 	for _, rule := range f.rules {
-		err := rule.verify(tableInfos)
+		err := rule.verify(tableInfos, f.sqlMODE)
 		if err != nil {
 			log.Error("failed to verify expression filter rule", zap.Error(err))
 			return errors.Trace(err)
diff --git a/pkg/filter/filter.go b/pkg/filter/filter.go
index d5674475f..a964c68e6 100644
--- a/pkg/filter/filter.go
+++ b/pkg/filter/filter.go
@@ -102,11 +102,11 @@ func NewFilter(cfg *config.ReplicaConfig, tz string) (Filter, error) {
 		f = tfilter.CaseInsensitive(f)
 	}
 
-	dmlExprFilter, err := newExprFilter(tz, cfg.Filter)
+	dmlExprFilter, err := newExprFilter(tz, cfg.Filter, cfg.SQLMode)
 	if err != nil {
 		return nil, err
 	}
-	sqlEventFilter, err := newSQLEventFilter(cfg.Filter)
+	sqlEventFilter, err := newSQLEventFilter(cfg.Filter, cfg.SQLMode)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/filter/sql_event_filter.go b/pkg/filter/sql_event_filter.go
index d9e80d623..fdf3ab885 100644
--- a/pkg/filter/sql_event_filter.go
+++ b/pkg/filter/sql_event_filter.go
@@ -14,6 +14,7 @@
 package filter
 
 import (
+	"fmt"
 	"sync"
 
 	"github.com/pingcap/errors"
@@ -21,6 +22,7 @@ import (
 	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb/parser"
 	timodel "github.com/pingcap/tidb/parser/model"
+	"github.com/pingcap/tidb/parser/mysql"
 	tfilter "github.com/pingcap/tidb/util/table-filter"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/pkg/config"
@@ -107,9 +109,17 @@ type sqlEventFilter struct {
 	rules     []*sqlEventRule
 }
 
-func newSQLEventFilter(cfg *config.FilterConfig) (*sqlEventFilter, error) {
+func newSQLEventFilter(cfg *config.FilterConfig, sqlMode string) (*sqlEventFilter, error) {
+	p := parser.New()
+	mode, err := mysql.GetSQLMode(sqlMode)
+	if err != nil {
+		log.Error("failed to get sql mode", zap.Error(err))
+		return nil, cerror.ErrInvalidReplicaConfig.FastGenByArgs(fmt.Sprintf("invalid sqlMode %s", sqlMode))
+	}
+	p.SetSQLMode(mode)
+
 	res := &sqlEventFilter{
-		ddlParser: parser.New(),
+		ddlParser: p,
 	}
 	for _, rule := range cfg.EventFilters {
 		if err := res.addRule(rule); err != nil {
diff --git a/pkg/redo/config.go b/pkg/redo/config.go
index fbcdb3f1e..d6b66de45 100644
--- a/pkg/redo/config.go
+++ b/pkg/redo/config.go
@@ -44,6 +44,8 @@ const (
 	FlushWarnDuration = time.Second * 20
 	// DefaultFlushIntervalInMs is the default flush interval for redo log.
 	DefaultFlushIntervalInMs = 2000
+	// DefaultMetaFlushIntervalInMs is the default flush interval for redo meta.
+	DefaultMetaFlushIntervalInMs = 200
 	// MinFlushIntervalInMs is the minimum flush interval for redo log.
 	MinFlushIntervalInMs = 50
 
@@ -172,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.
@@ -220,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/codec/avro/confluent_schema_registry.go b/pkg/sink/codec/avro/confluent_schema_registry.go
index 9f06fe645..c7e085181 100644
--- a/pkg/sink/codec/avro/confluent_schema_registry.go
+++ b/pkg/sink/codec/avro/confluent_schema_registry.go
@@ -210,20 +210,12 @@ func (m *confluentSchemaManager) Lookup(
 	m.cacheRWLock.RLock()
 	entry, exists := m.cache[schemaName]
 	if exists && entry.schemaID.confluentSchemaID == schemaID.confluentSchemaID {
-		log.Debug("Avro schema lookup cache hit",
-			zap.String("key", schemaName),
-			zap.Int("schemaID", entry.schemaID.confluentSchemaID))
 		m.cacheRWLock.RUnlock()
 		return entry.codec, nil
 	}
 	m.cacheRWLock.RUnlock()
 
-	log.Info("Avro schema lookup cache miss",
-		zap.String("key", schemaName),
-		zap.Int("schemaID", schemaID.confluentSchemaID))
-
 	uri := m.registryURL + "/schemas/ids/" + strconv.Itoa(schemaID.confluentSchemaID)
-	log.Debug("Querying for latest schema", zap.String("uri", uri))
 
 	req, err := http.NewRequestWithContext(ctx, "GET", uri, nil)
 	if err != nil {
@@ -289,11 +281,6 @@ func (m *confluentSchemaManager) Lookup(
 	m.cacheRWLock.Lock()
 	m.cache[schemaName] = cacheEntry
 	m.cacheRWLock.Unlock()
-
-	log.Info("Avro schema lookup successful with cache miss",
-		zap.Int("schemaID", cacheEntry.schemaID.confluentSchemaID),
-		zap.String("schema", cacheEntry.codec.Schema()))
-
 	return cacheEntry.codec, nil
 }
 
diff --git a/pkg/sink/codec/simple/decoder.go b/pkg/sink/codec/simple/decoder.go
index ef8e4598b..2913a5083 100644
--- a/pkg/sink/codec/simple/decoder.go
+++ b/pkg/sink/codec/simple/decoder.go
@@ -1,4 +1,4 @@
-// Copyright 2022 PingCAP, Inc.
+// Copyright 2023 PingCAP, Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -14,33 +14,66 @@
 package simple
 
 import (
+	"encoding/json"
+
 	"github.com/pingcap/tiflow/cdc/model"
+	cerror "github.com/pingcap/tiflow/pkg/errors"
 )
 
-type decoder struct{}
+type decoder struct {
+	value []byte
+
+	msg *message
+}
 
 // NewDecoder returns a new decoder
 func NewDecoder() *decoder {
-	// TODO implement me
-	panic("implement me")
+	return &decoder{}
 }
 
 // AddKeyValue add the received key and values to the decoder,
-func (d *decoder) AddKeyValue(key, value []byte) error {
-	// TODO implement me
-	panic("implement me")
+func (d *decoder) AddKeyValue(_, value []byte) error {
+	if d.value != nil {
+		return cerror.ErrDecodeFailed.GenWithStack(
+			"decoder value already exists, not consumed yet")
+	}
+	d.value = value
+	return nil
 }
 
 // HasNext returns whether there is any event need to be consumed
 func (d *decoder) HasNext() (model.MessageType, bool, error) {
-	// TODO implement me
-	panic("implement me")
+	if d.value == nil {
+		return model.MessageTypeUnknown, false, nil
+	}
+
+	var m message
+	if err := json.Unmarshal(d.value, &m); err != nil {
+		return model.MessageTypeUnknown, false, cerror.WrapError(cerror.ErrDecodeFailed, err)
+	}
+	d.msg = &m
+	d.value = nil
+
+	if d.msg.Type == WatermarkType {
+		return model.MessageTypeResolved, true, nil
+	}
+	if d.msg.Data != nil || d.msg.Old != nil {
+		return model.MessageTypeRow, true, nil
+	}
+	return model.MessageTypeDDL, true, nil
 }
 
 // NextResolvedEvent returns the next resolved event if exists
 func (d *decoder) NextResolvedEvent() (uint64, error) {
-	// TODO implement me
-	panic("implement me")
+	if d.msg.Type != WatermarkType {
+		return 0, cerror.ErrCodecDecode.GenWithStack(
+			"not found resolved event message")
+	}
+
+	ts := d.msg.CommitTs
+	d.msg = nil
+
+	return ts, nil
 }
 
 // NextRowChangedEvent returns the next row changed event if exists
diff --git a/pkg/sink/codec/simple/encoder.go b/pkg/sink/codec/simple/encoder.go
index babf9d8cf..be0b34aa8 100644
--- a/pkg/sink/codec/simple/encoder.go
+++ b/pkg/sink/codec/simple/encoder.go
@@ -15,8 +15,12 @@ package simple
 
 import (
 	"context"
+	"encoding/json"
 
 	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/pingcap/tiflow/pkg/config"
+	cerror "github.com/pingcap/tiflow/pkg/errors"
+	"github.com/pingcap/tiflow/pkg/sink/codec"
 	"github.com/pingcap/tiflow/pkg/sink/codec/common"
 )
 
@@ -27,8 +31,12 @@ type builder struct{}
 
 // NewBuilder returns a new builder
 func NewBuilder() *builder {
-	// TODO implement me
-	panic("implement me")
+	return &builder{}
+}
+
+// Build implement the RowEventEncoderBuilder interface
+func (b *builder) Build() codec.RowEventEncoder {
+	return &encoder{}
 }
 
 // AppendRowChangedEvent implement the RowEventEncoder interface
@@ -53,8 +61,12 @@ func (e *encoder) Build() []*common.Message {
 //
 //nolint:unused
 func (e *encoder) EncodeCheckpointEvent(ts uint64) (*common.Message, error) {
-	// TODO implement me
-	panic("implement me")
+	message := newResolvedMessage(ts)
+	value, err := json.Marshal(message)
+	if err != nil {
+		return nil, cerror.WrapError(cerror.ErrEncodeFailed, err)
+	}
+	return common.NewResolvedMsg(config.ProtocolSimple, nil, value, ts), nil
 }
 
 // EncodeDDLEvent implement the DDLEventBatchEncoder interface
diff --git a/pkg/sink/codec/simple/message.go b/pkg/sink/codec/simple/message.go
new file mode 100644
index 000000000..6e5cb8fc3
--- /dev/null
+++ b/pkg/sink/codec/simple/message.go
@@ -0,0 +1,46 @@
+// Copyright 2023 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package simple
+
+const (
+	defaultVersion = 0
+)
+
+// EventType describes the type of the event.
+type EventType string
+
+const (
+	// WatermarkType is the type of the watermark event.
+	WatermarkType EventType = "WATERMARK"
+)
+
+type message struct {
+	Version int `json:"version"`
+	// Scheme and Table is empty for the resolved ts event.
+	Schema   string    `json:"schema,omitempty"`
+	Table    string    `json:"table,omitempty"`
+	Type     EventType `json:"type"`
+	CommitTs uint64    `json:"commitTs"`
+	// Data is available for the row changed event.
+	Data map[string]interface{} `json:"data,omitempty"`
+	Old  map[string]interface{} `json:"old,omitempty"`
+}
+
+func newResolvedMessage(ts uint64) *message {
+	return &message{
+		Version:  defaultVersion,
+		Type:     WatermarkType,
+		CommitTs: ts,
+	}
+}
diff --git a/pkg/txnutil/lock_resolver.go b/pkg/txnutil/lock_resolver.go
index 9826bbcc1..dfdbf4f94 100644
--- a/pkg/txnutil/lock_resolver.go
+++ b/pkg/txnutil/lock_resolver.go
@@ -50,16 +50,32 @@ func NewLockerResolver(
 
 const scanLockLimit = 1024
 
-func (r *resolver) Resolve(ctx context.Context, regionID uint64, maxVersion uint64) error {
-	// TODO test whether this function will kill active transaction
+func (r *resolver) Resolve(ctx context.Context, regionID uint64, maxVersion uint64) (err error) {
+	var lockCount int = 0
+
+	log.Info("resolve lock starts",
+		zap.Uint64("regionID", regionID),
+		zap.Uint64("maxVersion", maxVersion),
+		zap.String("namespace", r.changefeed.Namespace),
+		zap.String("changefeed", r.changefeed.ID))
 
+	defer func() {
+		log.Info("resolve lock finishes",
+			zap.Uint64("regionID", regionID),
+			zap.Int("lockCount", lockCount),
+			zap.Uint64("maxVersion", maxVersion),
+			zap.String("namespace", r.changefeed.Namespace),
+			zap.String("changefeed", r.changefeed.ID),
+			zap.Error(err))
+	}()
+
+	// TODO test whether this function will kill active transaction
 	req := tikvrpc.NewRequest(tikvrpc.CmdScanLock, &kvrpcpb.ScanLockRequest{
 		MaxVersion: maxVersion,
 		Limit:      scanLockLimit,
 	})
 
 	bo := tikv.NewGcResolveLockMaxBackoffer(ctx)
-	var lockCount int
 	var loc *tikv.KeyLocation
 	var key []byte
 	flushRegion := func() error {
@@ -128,11 +144,5 @@ func (r *resolver) Resolve(ctx context.Context, regionID uint64, maxVersion uint
 		}
 		bo = tikv.NewGcResolveLockMaxBackoffer(ctx)
 	}
-	log.Info("resolve lock successfully",
-		zap.Uint64("regionID", regionID),
-		zap.Int("lockCount", lockCount),
-		zap.Uint64("maxVersion", maxVersion),
-		zap.String("namespace", r.changefeed.Namespace),
-		zap.String("changefeed", r.changefeed.ID))
 	return nil
 }
diff --git a/scripts/check-diff-line-width.sh b/scripts/check-diff-line-width.sh
index af65760cc..a034d0a2b 100755
--- a/scripts/check-diff-line-width.sh
+++ b/scripts/check-diff-line-width.sh
@@ -27,6 +27,7 @@ git --no-pager diff $BASE_HASH -U0 -- cdc pkg cmd \
 	-- ':(exclude)*_gen.go' \
 	-- ':(exclude)*_gen_test.go' \
 	-- ':(exclude)*_mock.go' \
+	-- ':(exclude)*_test_data.go' \
 	-- ':(exclude)*.pb.go' |
 	grep -E '^\+' | grep -vE '^\+\+\+' | grep -vE 'json:' | grep -vE 'toml:' |
 	sed 's/\t/    /g' |


Test patch:
diff --git a/cdc/api/v2/model_test.go b/cdc/api/v2/model_test.go
index 8e3b697e4..8a33d3a38 100644
--- a/cdc/api/v2/model_test.go
+++ b/cdc/api/v2/model_test.go
@@ -60,11 +60,12 @@ var defaultAPIConfig = &ReplicaConfig{
 		AdvanceTimeoutInSec:              util.AddressOf(uint(150)),
 	},
 	Consistent: &ConsistentConfig{
-		Level:             "none",
-		MaxLogSize:        64,
-		FlushIntervalInMs: redo.DefaultFlushIntervalInMs,
-		Storage:           "",
-		UseFileBackend:    false,
+		Level:                 "none",
+		MaxLogSize:            64,
+		FlushIntervalInMs:     redo.DefaultFlushIntervalInMs,
+		MetaFlushIntervalInMs: redo.DefaultMetaFlushIntervalInMs,
+		Storage:               "",
+		UseFileBackend:        false,
 	},
 	Scheduler: &ChangefeedSchedulerConfig{
 		EnableTableAcrossNodes: config.GetDefaultReplicaConfig().
@@ -80,6 +81,7 @@ var defaultAPIConfig = &ReplicaConfig{
 	},
 	ChangefeedErrorStuckDuration: &JSONDuration{*config.
 		GetDefaultReplicaConfig().ChangefeedErrorStuckDuration},
+	SQLMode: config.GetDefaultReplicaConfig().SQLMode,
 }
 
 func TestDefaultReplicaConfig(t *testing.T) {
diff --git a/cdc/entry/mounter_test.go b/cdc/entry/mounter_test.go
index 9fed0d270..e1df5b2a4 100644
--- a/cdc/entry/mounter_test.go
+++ b/cdc/entry/mounter_test.go
@@ -1601,9 +1601,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))
@@ -1615,7 +1647,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/kv/shared_client_test.go b/cdc/kv/shared_client_test.go
index 5a28d6e7b..4764fd1cb 100644
--- a/cdc/kv/shared_client_test.go
+++ b/cdc/kv/shared_client_test.go
@@ -38,17 +38,22 @@ import (
 )
 
 func TestRequestedStreamRequestedRegions(t *testing.T) {
-	stream := &requestedStream{streamID: 100, requests: chann.NewAutoDrainChann[singleRegionInfo]()}
-	defer stream.requests.CloseAndDrain()
-	stream.requestedRegions.m = make(map[SubscriptionID]map[uint64]*regionFeedState)
+	stream := newRequestedStream(100)
 
 	require.Nil(t, stream.getState(1, 2))
 	require.Nil(t, stream.takeState(1, 2))
 
-	stream.setState(1, 2, &regionFeedState{sri: singleRegionInfo{requestedTable: &requestedTable{}}})
+	stream.setState(1, 2, &regionFeedState{})
 	require.NotNil(t, stream.getState(1, 2))
 	require.NotNil(t, stream.takeState(1, 2))
 	require.Nil(t, stream.getState(1, 2))
+	require.Equal(t, 0, len(stream.requestedRegions.m))
+
+	stream.setState(1, 2, &regionFeedState{})
+	require.NotNil(t, stream.getState(1, 2))
+	require.NotNil(t, stream.takeState(1, 2))
+	require.Nil(t, stream.getState(1, 2))
+	require.Equal(t, 0, len(stream.requestedRegions.m))
 }
 
 func TestRequestedTable(t *testing.T) {
diff --git a/cdc/owner/changefeed_test.go b/cdc/owner/changefeed_test.go
index ffb4725f0..8c9f9ddf8 100644
--- a/cdc/owner/changefeed_test.go
+++ b/cdc/owner/changefeed_test.go
@@ -537,9 +537,10 @@ func TestRemoveChangefeed(t *testing.T) {
 	dir := t.TempDir()
 	info.SinkURI = "mysql://"
 	info.Config.Consistent = &config.ConsistentConfig{
-		Level:             "eventual",
-		Storage:           filepath.Join("nfs://", dir),
-		FlushIntervalInMs: redo.DefaultFlushIntervalInMs,
+		Level:                 "eventual",
+		Storage:               filepath.Join("nfs://", dir),
+		FlushIntervalInMs:     redo.DefaultFlushIntervalInMs,
+		MetaFlushIntervalInMs: redo.DefaultMetaFlushIntervalInMs,
 	}
 	ctx = cdcContext.WithChangefeedVars(ctx, &cdcContext.ChangefeedVars{
 		ID:   ctx.ChangefeedVars().ID,
diff --git a/cdc/owner/ddl_sink_test.go b/cdc/owner/ddl_sink_test.go
index be3c3ee58..52883f2e1 100644
--- a/cdc/owner/ddl_sink_test.go
+++ b/cdc/owner/ddl_sink_test.go
@@ -23,6 +23,7 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/sink/ddlsink"
+	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/retry"
 	"github.com/stretchr/testify/require"
@@ -61,7 +62,13 @@ func (m *mockSink) GetDDL() *model.DDLEvent {
 
 func newDDLSink4Test(reportErr func(err error), reportWarn func(err error)) (DDLSink, *mockSink) {
 	mockSink := &mockSink{}
-	ddlSink := newDDLSink(model.DefaultChangeFeedID("changefeed-test"), &model.ChangeFeedInfo{}, reportErr, reportWarn)
+	ddlSink := newDDLSink(
+		model.DefaultChangeFeedID("changefeed-test"),
+		&model.ChangeFeedInfo{
+			Config: config.GetDefaultReplicaConfig(),
+		},
+		reportErr,
+		reportWarn)
 	ddlSink.(*ddlSinkImpl).sinkInitHandler = func(ctx context.Context, s *ddlSinkImpl) error {
 		s.sink = mockSink
 		return nil
@@ -545,6 +552,9 @@ func TestAddSpecialComment(t *testing.T) {
 	}
 
 	s := &ddlSinkImpl{}
+	s.info = &model.ChangeFeedInfo{
+		Config: config.GetDefaultReplicaConfig(),
+	}
 	for _, ca := range testCase {
 		re, err := s.addSpecialComment(ca.event)
 		require.Nil(t, err)
diff --git a/cdc/processor/processor_test.go b/cdc/processor/processor_test.go
index 4b2e006f6..64ab9d9f4 100644
--- a/cdc/processor/processor_test.go
+++ b/cdc/processor/processor_test.go
@@ -71,14 +71,14 @@ func newProcessor4Test(
 		} else {
 			tmpDir := t.TempDir()
 			redoDir := fmt.Sprintf("%s/%s", tmpDir, changefeedID)
-			dmlMgr, err := redo.NewDMLManager(ctx, changefeedID, &config.ConsistentConfig{
-				Level:             string(redoPkg.ConsistentLevelEventual),
-				MaxLogSize:        redoPkg.DefaultMaxLogSize,
-				FlushIntervalInMs: redoPkg.DefaultFlushIntervalInMs,
-				Storage:           "file://" + redoDir,
-				UseFileBackend:    false,
+			dmlMgr := redo.NewDMLManager(changefeedID, &config.ConsistentConfig{
+				Level:                 string(redoPkg.ConsistentLevelEventual),
+				MaxLogSize:            redoPkg.DefaultMaxLogSize,
+				FlushIntervalInMs:     redoPkg.DefaultFlushIntervalInMs,
+				MetaFlushIntervalInMs: redoPkg.DefaultMetaFlushIntervalInMs,
+				Storage:               "file://" + redoDir,
+				UseFileBackend:        false,
 			})
-			require.NoError(t, err)
 			p.redo.r = dmlMgr
 		}
 		p.redo.name = "RedoManager"
diff --git a/cdc/processor/sinkmanager/manager_test.go b/cdc/processor/sinkmanager/manager_test.go
index 16806580c..b70e17fc3 100644
--- a/cdc/processor/sinkmanager/manager_test.go
+++ b/cdc/processor/sinkmanager/manager_test.go
@@ -117,7 +117,7 @@ func TestAddTable(t *testing.T) {
 	require.True(t, ok)
 	require.NotNil(t, tableSink)
 	require.Equal(t, 0, manager.sinkProgressHeap.len(), "Not started table shout not in progress heap")
-	err := manager.StartTable(span, 1, 1)
+	err := manager.StartTable(span, 1)
 	require.NoError(t, err)
 	require.Equal(t, uint64(0x7ffffffffffbffff), tableSink.(*tableSinkWrapper).replicateTs)
 
@@ -144,7 +144,7 @@ func TestRemoveTable(t *testing.T) {
 	tableSink, ok := manager.tableSinks.Load(span)
 	require.True(t, ok)
 	require.NotNil(t, tableSink)
-	err := manager.StartTable(span, 0, 0)
+	err := manager.StartTable(span, 0)
 	require.NoError(t, err)
 	addTableAndAddEventsToSortEngine(t, e, span)
 	manager.UpdateBarrierTs(4, nil)
@@ -191,7 +191,7 @@ func TestGenerateTableSinkTaskWithBarrierTs(t *testing.T) {
 	manager.UpdateBarrierTs(4, nil)
 	manager.UpdateReceivedSorterResolvedTs(span, 5)
 	manager.schemaStorage.AdvanceResolvedTs(5)
-	err := manager.StartTable(span, 0, 0)
+	err := manager.StartTable(span, 0)
 	require.NoError(t, err)
 
 	require.Eventually(t, func() bool {
@@ -222,7 +222,7 @@ func TestGenerateTableSinkTaskWithResolvedTs(t *testing.T) {
 	manager.UpdateBarrierTs(4, nil)
 	manager.UpdateReceivedSorterResolvedTs(span, 3)
 	manager.schemaStorage.AdvanceResolvedTs(4)
-	err := manager.StartTable(span, 0, 0)
+	err := manager.StartTable(span, 0)
 	require.NoError(t, err)
 
 	require.Eventually(t, func() bool {
@@ -252,7 +252,7 @@ func TestGetTableStatsToReleaseMemQuota(t *testing.T) {
 	manager.UpdateBarrierTs(4, nil)
 	manager.UpdateReceivedSorterResolvedTs(span, 5)
 	manager.schemaStorage.AdvanceResolvedTs(5)
-	err := manager.StartTable(span, 0, 0)
+	err := manager.StartTable(span, 0)
 	require.NoError(t, err)
 
 	require.Eventually(t, func() bool {
@@ -341,7 +341,7 @@ func TestSinkManagerRunWithErrors(t *testing.T) {
 
 	source.AddTable(span, "test", 100)
 	manager.AddTable(span, 100, math.MaxUint64)
-	manager.StartTable(span, 100, 0)
+	manager.StartTable(span, 100)
 	source.Add(span, model.NewResolvedPolymorphicEvent(0, 101))
 	manager.UpdateReceivedSorterResolvedTs(span, 101)
 	manager.UpdateBarrierTs(101, nil)
diff --git a/cdc/redo/manager_test.go b/cdc/redo/manager_test.go
index 0a85e01ba..a07eb39fe 100644
--- a/cdc/redo/manager_test.go
+++ b/cdc/redo/manager_test.go
@@ -26,7 +26,6 @@ import (
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/processor/tablepb"
 	"github.com/pingcap/tiflow/cdc/redo/writer"
-	"github.com/pingcap/tiflow/cdc/redo/writer/blackhole"
 	"github.com/pingcap/tiflow/pkg/config"
 	"github.com/pingcap/tiflow/pkg/redo"
 	"github.com/pingcap/tiflow/pkg/spanz"
@@ -115,21 +114,18 @@ func TestLogManagerInProcessor(t *testing.T) {
 	testWriteDMLs := func(storage string, useFileBackend bool) {
 		ctx, cancel := context.WithCancel(ctx)
 		cfg := &config.ConsistentConfig{
-			Level:             string(redo.ConsistentLevelEventual),
-			MaxLogSize:        redo.DefaultMaxLogSize,
-			Storage:           storage,
-			FlushIntervalInMs: redo.MinFlushIntervalInMs,
-			UseFileBackend:    useFileBackend,
+			Level:                 string(redo.ConsistentLevelEventual),
+			MaxLogSize:            redo.DefaultMaxLogSize,
+			Storage:               storage,
+			FlushIntervalInMs:     redo.MinFlushIntervalInMs,
+			MetaFlushIntervalInMs: redo.MinFlushIntervalInMs,
+			UseFileBackend:        useFileBackend,
 		}
-		dmlMgr, err := NewDMLManager(ctx, model.DefaultChangeFeedID("test"), cfg)
-		require.NoError(t, err)
-		wg := sync.WaitGroup{}
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			dmlMgr.Run(ctx)
-		}()
-
+		dmlMgr := NewDMLManager(model.DefaultChangeFeedID("test"), cfg)
+		var eg errgroup.Group
+		eg.Go(func() error {
+			return dmlMgr.Run(ctx)
+		})
 		// check emit row changed events can move forward resolved ts
 		spans := []tablepb.Span{
 			spanz.TableIDToComparableSpan(53),
@@ -202,7 +198,7 @@ func TestLogManagerInProcessor(t *testing.T) {
 		checkResolvedTs(t, dmlMgr.logManager, flushResolvedTs)
 
 		cancel()
-		wg.Wait()
+		require.ErrorIs(t, eg.Wait(), context.Canceled)
 	}
 
 	testWriteDMLs("blackhole://", true)
@@ -226,25 +222,24 @@ func TestLogManagerInOwner(t *testing.T) {
 	testWriteDDLs := func(storage string, useFileBackend bool) {
 		ctx, cancel := context.WithCancel(ctx)
 		cfg := &config.ConsistentConfig{
-			Level:             string(redo.ConsistentLevelEventual),
-			MaxLogSize:        redo.DefaultMaxLogSize,
-			Storage:           storage,
-			FlushIntervalInMs: redo.MinFlushIntervalInMs,
-			UseFileBackend:    useFileBackend,
+			Level:                 string(redo.ConsistentLevelEventual),
+			MaxLogSize:            redo.DefaultMaxLogSize,
+			Storage:               storage,
+			FlushIntervalInMs:     redo.MinFlushIntervalInMs,
+			MetaFlushIntervalInMs: redo.DefaultMetaFlushIntervalInMs,
+			UseFileBackend:        useFileBackend,
 		}
 		startTs := model.Ts(10)
-		ddlMgr, err := NewDDLManager(ctx, model.DefaultChangeFeedID("test"), cfg, startTs)
-		require.NoError(t, err)
-		wg := sync.WaitGroup{}
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			ddlMgr.Run(ctx)
-		}()
+		ddlMgr := NewDDLManager(model.DefaultChangeFeedID("test"), cfg, startTs)
+
+		var eg errgroup.Group
+		eg.Go(func() error {
+			return ddlMgr.Run(ctx)
+		})
 
 		require.Equal(t, startTs, ddlMgr.GetResolvedTs())
 		ddl := &model.DDLEvent{StartTs: 100, CommitTs: 120, Query: "CREATE TABLE `TEST.T1`"}
-		err = ddlMgr.EmitDDLEvent(ctx, ddl)
+		err := ddlMgr.EmitDDLEvent(ctx, ddl)
 		require.NoError(t, err)
 		require.Equal(t, startTs, ddlMgr.GetResolvedTs())
 
@@ -252,7 +247,7 @@ func TestLogManagerInOwner(t *testing.T) {
 		checkResolvedTs(t, ddlMgr.logManager, ddl.CommitTs)
 
 		cancel()
-		wg.Wait()
+		require.ErrorIs(t, eg.Wait(), context.Canceled)
 	}
 
 	testWriteDDLs("blackhole://", true)
@@ -273,25 +268,17 @@ func TestLogManagerError(t *testing.T) {
 	defer cancel()
 
 	cfg := &config.ConsistentConfig{
-		Level:             string(redo.ConsistentLevelEventual),
-		MaxLogSize:        redo.DefaultMaxLogSize,
-		Storage:           "blackhole://",
-		FlushIntervalInMs: redo.MinFlushIntervalInMs,
+		Level:                 string(redo.ConsistentLevelEventual),
+		MaxLogSize:            redo.DefaultMaxLogSize,
+		Storage:               "blackhole-invalid://",
+		FlushIntervalInMs:     redo.MinFlushIntervalInMs,
+		MetaFlushIntervalInMs: redo.MinFlushIntervalInMs,
 	}
-	logMgr, err := NewDMLManager(ctx, model.DefaultChangeFeedID("test"), cfg)
-	require.NoError(t, err)
-	err = logMgr.writer.Close()
-	require.NoError(t, err)
-	logMgr.writer = blackhole.NewInvalidLogWriter(logMgr.writer)
-
-	wg := sync.WaitGroup{}
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		err := logMgr.Run(ctx)
-		require.Regexp(t, ".*invalid black hole writer.*", err)
-		require.Regexp(t, ".*WriteLog.*", err)
-	}()
+	logMgr := NewDMLManager(model.DefaultChangeFeedID("test"), cfg)
+	var eg errgroup.Group
+	eg.Go(func() error {
+		return logMgr.Run(ctx)
+	})
 
 	testCases := []struct {
 		span tablepb.Span
@@ -310,7 +297,10 @@ func TestLogManagerError(t *testing.T) {
 		err := logMgr.emitRedoEvents(ctx, tc.span, nil, tc.rows...)
 		require.NoError(t, err)
 	}
-	wg.Wait()
+
+	err := eg.Wait()
+	require.Regexp(t, ".*invalid black hole writer.*", err)
+	require.Regexp(t, ".*WriteLog.*", err)
 }
 
 func BenchmarkBlackhole(b *testing.B) {
@@ -330,15 +320,15 @@ func BenchmarkFileWriter(b *testing.B) {
 func runBenchTest(b *testing.B, storage string, useFileBackend bool) {
 	ctx, cancel := context.WithCancel(context.Background())
 	cfg := &config.ConsistentConfig{
-		Level:             string(redo.ConsistentLevelEventual),
-		MaxLogSize:        redo.DefaultMaxLogSize,
-		Storage:           storage,
-		FlushIntervalInMs: redo.MinFlushIntervalInMs,
-		UseFileBackend:    useFileBackend,
+		Level:                 string(redo.ConsistentLevelEventual),
+		MaxLogSize:            redo.DefaultMaxLogSize,
+		Storage:               storage,
+		FlushIntervalInMs:     redo.MinFlushIntervalInMs,
+		MetaFlushIntervalInMs: redo.MinFlushIntervalInMs,
+		UseFileBackend:        useFileBackend,
 	}
-	dmlMgr, err := NewDMLManager(ctx, model.DefaultChangeFeedID("test"), 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)
 	})
@@ -366,7 +356,7 @@ func runBenchTest(b *testing.B, storage string, useFileBackend bool) {
 		go func(span tablepb.Span) {
 			defer wg.Done()
 			maxCommitTs := maxTsMap.GetV(span)
-			rows := []*model.RowChangedEvent{}
+			var rows []*model.RowChangedEvent
 			for i := 0; i < maxRowCount; i++ {
 				if i%100 == 0 {
 					// prepare new row change events
@@ -409,6 +399,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 cb110f5ba..9143f02cd 100644
--- a/cdc/redo/meta_manager_test.go
+++ b/cdc/redo/meta_manager_test.go
@@ -50,17 +50,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)
@@ -71,16 +72,28 @@ func TestInitAndWriteMeta(t *testing.T) {
 
 	startTs := uint64(10)
 	cfg := &config.ConsistentConfig{
-		Level:             string(redo.ConsistentLevelEventual),
-		MaxLogSize:        redo.DefaultMaxLogSize,
-		Storage:           uri.String(),
-		FlushIntervalInMs: redo.MinFlushIntervalInMs,
+		Level:                 string(redo.ConsistentLevelEventual),
+		MaxLogSize:            redo.DefaultMaxLogSize,
+		Storage:               uri.String(),
+		FlushIntervalInMs:     redo.MinFlushIntervalInMs,
+		MetaFlushIntervalInMs: redo.MinFlushIntervalInMs,
 	}
-	m, err := NewMetaManagerWithInit(ctx, changefeedID, 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)
@@ -91,7 +104,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) {
@@ -114,7 +130,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)
@@ -135,29 +151,42 @@ func TestPreCleanupAndWriteMeta(t *testing.T) {
 
 	startTs := uint64(10)
 	cfg := &config.ConsistentConfig{
-		Level:             string(redo.ConsistentLevelEventual),
-		MaxLogSize:        redo.DefaultMaxLogSize,
-		Storage:           uri.String(),
-		FlushIntervalInMs: redo.MinFlushIntervalInMs,
+		Level:                 string(redo.ConsistentLevelEventual),
+		MaxLogSize:            redo.DefaultMaxLogSize,
+		Storage:               uri.String(),
+		FlushIntervalInMs:     redo.MinFlushIntervalInMs,
+		MetaFlushIntervalInMs: redo.MinFlushIntervalInMs,
 	}
-	m, err := NewMetaManagerWithInit(ctx, changefeedID, 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) {
@@ -174,14 +203,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)
@@ -207,9 +232,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) {
@@ -260,20 +282,27 @@ func TestGCAndCleanup(t *testing.T) {
 
 	startTs := uint64(3)
 	cfg := &config.ConsistentConfig{
-		Level:             string(redo.ConsistentLevelEventual),
-		MaxLogSize:        redo.DefaultMaxLogSize,
-		Storage:           uri.String(),
-		FlushIntervalInMs: redo.MinFlushIntervalInMs,
+		Level:                 string(redo.ConsistentLevelEventual),
+		MaxLogSize:            redo.DefaultMaxLogSize,
+		Storage:               uri.String(),
+		FlushIntervalInMs:     redo.MinFlushIntervalInMs,
+		MetaFlushIntervalInMs: redo.MinFlushIntervalInMs,
 	}
-	m, err := NewMetaManagerWithInit(ctx, changefeedID, 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 60ea189a8..bba18c806 100644
--- a/cdc/redo/writer/memory/mem_log_writer_test.go
+++ b/cdc/redo/writer/memory/mem_log_writer_test.go
@@ -61,7 +61,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/agent/main_test.go b/cdc/scheduler/internal/v3/agent/main_test.go
deleted file mode 100644
index cb7775cac..000000000
--- a/cdc/scheduler/internal/v3/agent/main_test.go
+++ /dev/null
@@ -1,24 +0,0 @@
-// Copyright 2022 PingCAP, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package agent
-
-import (
-	"testing"
-
-	"github.com/pingcap/tiflow/pkg/leakutil"
-)
-
-func TestMain(m *testing.M) {
-	leakutil.SetUpLeakTest(m)
-}
diff --git a/cdc/scheduler/internal/v3/compat/main_test.go b/cdc/scheduler/internal/v3/compat/main_test.go
deleted file mode 100644
index d8ad350ef..000000000
--- a/cdc/scheduler/internal/v3/compat/main_test.go
+++ /dev/null
@@ -1,24 +0,0 @@
-// Copyright 2022 PingCAP, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package compat
-
-import (
-	"testing"
-
-	"github.com/pingcap/tiflow/pkg/leakutil"
-)
-
-func TestMain(m *testing.M) {
-	leakutil.SetUpLeakTest(m)
-}
diff --git a/cdc/scheduler/internal/v3/coordinator_bench_test.go b/cdc/scheduler/internal/v3/coordinator_bench_test.go
index cdf3adc86..75094ba1d 100644
--- a/cdc/scheduler/internal/v3/coordinator_bench_test.go
+++ b/cdc/scheduler/internal/v3/coordinator_bench_test.go
@@ -149,7 +149,7 @@ func BenchmarkCoordinatorHeartbeatResponse(b *testing.B) {
 			captureID := fmt.Sprint(i % captureCount)
 			span := tablepb.Span{TableID: tableID}
 			rep, err := replication.NewReplicationSet(
-				span, tablepb.Checkpoint{}, map[string]*tablepb.TableStatus{
+				span, 0, map[string]*tablepb.TableStatus{
 					captureID: {
 						Span:  tablepb.Span{TableID: tableID},
 						State: tablepb.TableStateReplicating,
diff --git a/cdc/scheduler/internal/v3/keyspan/main_test.go b/cdc/scheduler/internal/v3/keyspan/main_test.go
deleted file mode 100644
index da1223583..000000000
--- a/cdc/scheduler/internal/v3/keyspan/main_test.go
+++ /dev/null
@@ -1,24 +0,0 @@
-// Copyright 2022 PingCAP, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package keyspan
-
-import (
-	"testing"
-
-	"github.com/pingcap/tiflow/pkg/leakutil"
-)
-
-func TestMain(m *testing.M) {
-	leakutil.SetUpLeakTest(m)
-}
diff --git a/cdc/scheduler/internal/v3/member/main_test.go b/cdc/scheduler/internal/v3/member/main_test.go
deleted file mode 100644
index 02b77a221..000000000
--- a/cdc/scheduler/internal/v3/member/main_test.go
+++ /dev/null
@@ -1,24 +0,0 @@
-// Copyright 2022 PingCAP, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package member
-
-import (
-	"testing"
-
-	"github.com/pingcap/tiflow/pkg/leakutil"
-)
-
-func TestMain(m *testing.M) {
-	leakutil.SetUpLeakTest(m)
-}
diff --git a/cdc/scheduler/internal/v3/replication/main_test.go b/cdc/scheduler/internal/v3/replication/main_test.go
deleted file mode 100644
index cefa385ed..000000000
--- a/cdc/scheduler/internal/v3/replication/main_test.go
+++ /dev/null
@@ -1,24 +0,0 @@
-// Copyright 2022 PingCAP, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package replication
-
-import (
-	"testing"
-
-	"github.com/pingcap/tiflow/pkg/leakutil"
-)
-
-func TestMain(m *testing.M) {
-	leakutil.SetUpLeakTest(m)
-}
diff --git a/cdc/scheduler/internal/v3/replication/replication_manager_test.go b/cdc/scheduler/internal/v3/replication/replication_manager_test.go
index 6cbfab083..632a0835a 100644
--- a/cdc/scheduler/internal/v3/replication/replication_manager_test.go
+++ b/cdc/scheduler/internal/v3/replication/replication_manager_test.go
@@ -35,12 +35,7 @@ func TestReplicationManagerHandleAddTableTask(t *testing.T) {
 	// Absent -> Prepare
 	msgs, err := r.HandleTasks([]*ScheduleTask{{
 		AddTable: &AddTable{
-			Span:      spanz.TableIDToComparableSpan(1),
-			CaptureID: "1",
-			Checkpoint: tablepb.Checkpoint{
-				CheckpointTs: 1,
-				ResolvedTs:   1,
-			},
+			Span: spanz.TableIDToComparableSpan(1), CaptureID: "1", CheckpointTs: 1,
 		},
 		Accept: func() {
 			addTableCh <- 1
@@ -155,15 +150,9 @@ func TestReplicationManagerRemoveTable(t *testing.T) {
 
 	// Add the table.
 	span := spanz.TableIDToComparableSpan(1)
-	tbl, err := NewReplicationSet(span,
-		tablepb.Checkpoint{
-			CheckpointTs: 0,
-			ResolvedTs:   0,
-		},
-		map[string]*tablepb.TableStatus{
-			"1": {Span: span, State: tablepb.TableStateReplicating},
-		},
-		model.ChangeFeedID{})
+	tbl, err := NewReplicationSet(span, 0, map[string]*tablepb.TableStatus{
+		"1": {Span: span, State: tablepb.TableStateReplicating},
+	}, model.ChangeFeedID{})
 	require.Nil(t, err)
 	require.Equal(t, ReplicationSetStateReplicating, tbl.State)
 	r.spans.ReplaceOrInsert(spanz.TableIDToComparableSpan(1), tbl)
@@ -257,15 +246,9 @@ func TestReplicationManagerMoveTable(t *testing.T) {
 
 	// Add the table.
 	span := spanz.TableIDToComparableSpan(1)
-	tbl, err := NewReplicationSet(span,
-		tablepb.Checkpoint{
-			CheckpointTs: 0,
-			ResolvedTs:   0,
-		},
-		map[string]*tablepb.TableStatus{
-			source: {Span: span, State: tablepb.TableStateReplicating},
-		},
-		model.ChangeFeedID{})
+	tbl, err := NewReplicationSet(span, 0, map[string]*tablepb.TableStatus{
+		source: {Span: span, State: tablepb.TableStateReplicating},
+	}, model.ChangeFeedID{})
 	require.Nil(t, err)
 	require.Equal(t, ReplicationSetStateReplicating, tbl.State)
 	r.spans.ReplaceOrInsert(spanz.TableIDToComparableSpan(1), tbl)
@@ -394,23 +377,19 @@ func TestReplicationManagerBurstBalance(t *testing.T) {
 	r := NewReplicationManager(1, model.ChangeFeedID{})
 	balanceTableCh := make(chan int, 1)
 
-	checkpoint := tablepb.Checkpoint{
-		CheckpointTs: 1,
-		ResolvedTs:   1,
-	}
 	// Burst balance is not limited by maxTaskConcurrency.
 	msgs, err := r.HandleTasks([]*ScheduleTask{{
 		AddTable: &AddTable{
-			Span: spanz.TableIDToComparableSpan(1), CaptureID: "0", Checkpoint: checkpoint,
+			Span: spanz.TableIDToComparableSpan(1), CaptureID: "0", CheckpointTs: 1,
 		},
 	}, {
 		BurstBalance: &BurstBalance{
 			AddTables: []AddTable{{
-				Span: spanz.TableIDToComparableSpan(1), CaptureID: "1", Checkpoint: checkpoint,
+				Span: spanz.TableIDToComparableSpan(1), CaptureID: "1", CheckpointTs: 1,
 			}, {
-				Span: spanz.TableIDToComparableSpan(2), CaptureID: "2", Checkpoint: checkpoint,
+				Span: spanz.TableIDToComparableSpan(2), CaptureID: "2", CheckpointTs: 1,
 			}, {
-				Span: spanz.TableIDToComparableSpan(3), CaptureID: "3", Checkpoint: checkpoint,
+				Span: spanz.TableIDToComparableSpan(3), CaptureID: "3", CheckpointTs: 1,
 			}},
 		},
 		Accept: func() {
@@ -445,25 +424,19 @@ func TestReplicationManagerBurstBalance(t *testing.T) {
 
 	// Add a new table.
 	span := spanz.TableIDToComparableSpan(5)
-	table5, err := NewReplicationSet(span,
-		tablepb.Checkpoint{},
-		map[string]*tablepb.TableStatus{
-			"5": {Span: span, State: tablepb.TableStateReplicating},
-		}, model.ChangeFeedID{})
+	table5, err := NewReplicationSet(span, 0, map[string]*tablepb.TableStatus{
+		"5": {Span: span, State: tablepb.TableStateReplicating},
+	}, model.ChangeFeedID{})
 	require.Nil(t, err)
 	r.spans.ReplaceOrInsert(span, table5)
 
-	checkpoint = tablepb.Checkpoint{
-		CheckpointTs: 2,
-		ResolvedTs:   2,
-	}
 	// More burst balance is still allowed.
 	msgs, err = r.HandleTasks([]*ScheduleTask{{
 		BurstBalance: &BurstBalance{
 			AddTables: []AddTable{{
-				Span: spanz.TableIDToComparableSpan(4), CaptureID: "4", Checkpoint: checkpoint,
+				Span: spanz.TableIDToComparableSpan(4), CaptureID: "4", CheckpointTs: 2,
 			}, {
-				Span: spanz.TableIDToComparableSpan(1), CaptureID: "0", Checkpoint: checkpoint,
+				Span: spanz.TableIDToComparableSpan(1), CaptureID: "0", CheckpointTs: 2,
 			}},
 			RemoveTables: []RemoveTable{{
 				Span: spanz.TableIDToComparableSpan(5), CaptureID: "5",
@@ -516,13 +489,13 @@ func TestReplicationManagerBurstBalanceMoveTables(t *testing.T) {
 	var err error
 	// Two tables in "1".
 	span := spanz.TableIDToComparableSpan(1)
-	table, err := NewReplicationSet(span, tablepb.Checkpoint{}, map[string]*tablepb.TableStatus{
+	table, err := NewReplicationSet(span, 0, map[string]*tablepb.TableStatus{
 		"1": {Span: span, State: tablepb.TableStateReplicating},
 	}, model.ChangeFeedID{})
 	require.Nil(t, err)
 	r.spans.ReplaceOrInsert(span, table)
 	span2 := spanz.TableIDToComparableSpan(2)
-	table2, err := NewReplicationSet(span2, tablepb.Checkpoint{}, map[string]*tablepb.TableStatus{
+	table2, err := NewReplicationSet(span2, 0, map[string]*tablepb.TableStatus{
 		"1": {
 			Span: span2, State: tablepb.TableStateReplicating,
 			Checkpoint: tablepb.Checkpoint{CheckpointTs: 1, ResolvedTs: 1},
@@ -631,15 +604,15 @@ func (m *mockRedoMetaManager) Enabled() bool {
 	return m.enable
 }
 
+func (m *mockRedoMetaManager) Running() bool {
+	return true
+}
+
 func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 	t.Parallel()
 	r := NewReplicationManager(1, model.ChangeFeedID{})
 	span := spanz.TableIDToComparableSpan(1)
-	rs, err := NewReplicationSet(span,
-		tablepb.Checkpoint{
-			CheckpointTs: 10,
-			ResolvedTs:   10,
-		},
+	rs, err := NewReplicationSet(span, model.Ts(10),
 		map[model.CaptureID]*tablepb.TableStatus{
 			"1": {
 				Span:  spanz.TableIDToComparableSpan(1),
@@ -654,11 +627,7 @@ func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 	r.spans.ReplaceOrInsert(span, rs)
 
 	span2 := spanz.TableIDToComparableSpan(2)
-	rs, err = NewReplicationSet(span2,
-		tablepb.Checkpoint{
-			CheckpointTs: 15,
-			ResolvedTs:   15,
-		},
+	rs, err = NewReplicationSet(span2, model.Ts(15),
 		map[model.CaptureID]*tablepb.TableStatus{
 			"2": {
 				Span:  spanz.TableIDToComparableSpan(2),
@@ -694,11 +663,7 @@ func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 	require.Equal(t, checkpointCannotProceed, resolved)
 
 	span3 := spanz.TableIDToComparableSpan(3)
-	rs, err = NewReplicationSet(span3,
-		tablepb.Checkpoint{
-			CheckpointTs: 5,
-			ResolvedTs:   5,
-		},
+	rs, err = NewReplicationSet(span3, model.Ts(5),
 		map[model.CaptureID]*tablepb.TableStatus{
 			"1": {
 				Span:  spanz.TableIDToComparableSpan(3),
@@ -725,11 +690,7 @@ func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 
 	currentTables.UpdateTables([]model.TableID{1, 2, 3, 4})
 	span4 := spanz.TableIDToComparableSpan(4)
-	rs, err = NewReplicationSet(span4,
-		tablepb.Checkpoint{
-			CheckpointTs: 3,
-			ResolvedTs:   3,
-		},
+	rs, err = NewReplicationSet(span4, model.Ts(3),
 		map[model.CaptureID]*tablepb.TableStatus{
 			"1": {
 				Span:  spanz.TableIDToComparableSpan(4),
@@ -753,11 +714,7 @@ func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 	span5_2 := spanz.TableIDToComparableSpan(5)
 	span5_2.StartKey = append(span5_2.StartKey, 0)
 	for _, span := range []tablepb.Span{span5_1, span5_2} {
-		rs, err = NewReplicationSet(span,
-			tablepb.Checkpoint{
-				CheckpointTs: 3,
-				ResolvedTs:   3,
-			},
+		rs, err = NewReplicationSet(span, model.Ts(3),
 			map[model.CaptureID]*tablepb.TableStatus{
 				"1": {
 					Span:  span,
@@ -791,11 +748,7 @@ func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 	// redo is enabled
 	currentTables.UpdateTables([]model.TableID{4})
 	spanRedo := spanz.TableIDToComparableSpan(4)
-	rs, err = NewReplicationSet(spanRedo,
-		tablepb.Checkpoint{
-			CheckpointTs: 3,
-			ResolvedTs:   3,
-		},
+	rs, err = NewReplicationSet(spanRedo, model.Ts(3),
 		map[model.CaptureID]*tablepb.TableStatus{
 			"1": {
 				Span:  spanz.TableIDToComparableSpan(4),
@@ -821,11 +774,7 @@ func TestReplicationManagerAdvanceCheckpointWithRedoEnabled(t *testing.T) {
 	t.Parallel()
 	r := NewReplicationManager(1, model.ChangeFeedID{})
 	span := spanz.TableIDToComparableSpan(1)
-	rs, err := NewReplicationSet(span,
-		tablepb.Checkpoint{
-			CheckpointTs: 10,
-			ResolvedTs:   10,
-		},
+	rs, err := NewReplicationSet(span, model.Ts(10),
 		map[model.CaptureID]*tablepb.TableStatus{
 			"1": {
 				Span:  spanz.TableIDToComparableSpan(1),
@@ -840,11 +789,7 @@ func TestReplicationManagerAdvanceCheckpointWithRedoEnabled(t *testing.T) {
 	r.spans.ReplaceOrInsert(span, rs)
 
 	span2 := spanz.TableIDToComparableSpan(2)
-	rs, err = NewReplicationSet(span2,
-		tablepb.Checkpoint{
-			CheckpointTs: 15,
-			ResolvedTs:   15,
-		},
+	rs, err = NewReplicationSet(span2, model.Ts(15),
 		map[model.CaptureID]*tablepb.TableStatus{
 			"2": {
 				Span:  spanz.TableIDToComparableSpan(2),
diff --git a/cdc/scheduler/internal/v3/replication/replication_set_test.go b/cdc/scheduler/internal/v3/replication/replication_set_test.go
index 1da88a557..da417438b 100644
--- a/cdc/scheduler/internal/v3/replication/replication_set_test.go
+++ b/cdc/scheduler/internal/v3/replication/replication_set_test.go
@@ -220,10 +220,7 @@ func TestNewReplicationSet(t *testing.T) {
 	for id, tc := range testcases {
 		set := tc.set
 		status := tc.tableStatus
-		checkpoint := tablepb.Checkpoint{
-			CheckpointTs: tc.checkpoint,
-			ResolvedTs:   tc.checkpoint,
-		}
+		checkpoint := tc.checkpoint
 
 		span := tablepb.Span{TableID: 0}
 		output, err := NewReplicationSet(span, checkpoint, status, model.ChangeFeedID{})
@@ -271,7 +268,7 @@ func TestReplicationSetPoll(t *testing.T) {
 			}
 		}
 		span := tablepb.Span{TableID: 1}
-		r, _ := NewReplicationSet(span, tablepb.Checkpoint{}, status, model.ChangeFeedID{})
+		r, _ := NewReplicationSet(span, 0, status, model.ChangeFeedID{})
 		var tableStates []int
 		for state := range tablepb.TableState_name {
 			tableStates = append(tableStates, int(state))
@@ -303,7 +300,7 @@ func TestReplicationSetPollUnknownCapture(t *testing.T) {
 
 	tableID := model.TableID(1)
 	span := tablepb.Span{TableID: tableID}
-	r, err := NewReplicationSet(span, tablepb.Checkpoint{}, map[model.CaptureID]*tablepb.TableStatus{
+	r, err := NewReplicationSet(span, 0, map[model.CaptureID]*tablepb.TableStatus{
 		"1": {
 			Span:       tablepb.Span{TableID: tableID},
 			State:      tablepb.TableStateReplicating,
@@ -340,7 +337,7 @@ func TestReplicationSetAddTable(t *testing.T) {
 	from := "1"
 	tableID := model.TableID(1)
 	span := tablepb.Span{TableID: tableID}
-	r, err := NewReplicationSet(span, tablepb.Checkpoint{}, nil, model.ChangeFeedID{})
+	r, err := NewReplicationSet(span, 0, nil, model.ChangeFeedID{})
 	require.Nil(t, err)
 
 	// Absent -> Prepare
@@ -485,7 +482,7 @@ func TestReplicationSetRemoveTable(t *testing.T) {
 	from := "1"
 	tableID := model.TableID(1)
 	span := tablepb.Span{TableID: tableID}
-	r, err := NewReplicationSet(span, tablepb.Checkpoint{}, nil, model.ChangeFeedID{})
+	r, err := NewReplicationSet(span, 0, nil, model.ChangeFeedID{})
 	require.Nil(t, err)
 
 	// Ignore removing table if it's not in replicating.
@@ -566,7 +563,7 @@ func TestReplicationSetMoveTable(t *testing.T) {
 
 	tableID := model.TableID(1)
 	span := tablepb.Span{TableID: tableID}
-	r, err := NewReplicationSet(span, tablepb.Checkpoint{}, nil, model.ChangeFeedID{})
+	r, err := NewReplicationSet(span, 0, nil, model.ChangeFeedID{})
 	require.Nil(t, err)
 
 	source := "1"
@@ -798,7 +795,7 @@ func TestReplicationSetCaptureShutdown(t *testing.T) {
 	from := "1"
 	tableID := model.TableID(1)
 	span := tablepb.Span{TableID: tableID}
-	r, err := NewReplicationSet(span, tablepb.Checkpoint{}, nil, model.ChangeFeedID{})
+	r, err := NewReplicationSet(span, 0, nil, model.ChangeFeedID{})
 	require.Nil(t, err)
 
 	// Add table, Absent -> Prepare
@@ -1104,7 +1101,7 @@ func TestReplicationSetCaptureShutdownAfterReconstructCommitState(t *testing.T)
 		from: {Span: tablepb.Span{TableID: tableID}, State: tablepb.TableStatePrepared},
 	}
 	span := tablepb.Span{TableID: tableID}
-	r, err := NewReplicationSet(span, tablepb.Checkpoint{}, tableStatus, model.ChangeFeedID{})
+	r, err := NewReplicationSet(span, 0, tableStatus, model.ChangeFeedID{})
 	require.Nil(t, err)
 	require.Equal(t, ReplicationSetStateCommit, r.State)
 	require.Equal(t, "", r.Primary)
@@ -1125,7 +1122,7 @@ func TestReplicationSetMoveTableWithHeartbeatResponse(t *testing.T) {
 
 	tableID := model.TableID(1)
 	span := tablepb.Span{TableID: tableID}
-	r, err := NewReplicationSet(span, tablepb.Checkpoint{}, nil, model.ChangeFeedID{})
+	r, err := NewReplicationSet(span, 0, nil, model.ChangeFeedID{})
 	require.Nil(t, err)
 
 	source := "1"
@@ -1213,7 +1210,7 @@ func TestReplicationSetMoveTableSameDestCapture(t *testing.T) {
 
 	tableID := model.TableID(1)
 	span := tablepb.Span{TableID: tableID}
-	r, err := NewReplicationSet(span, tablepb.Checkpoint{}, nil, model.ChangeFeedID{})
+	r, err := NewReplicationSet(span, 0, nil, model.ChangeFeedID{})
 	require.Nil(t, err)
 
 	source := "1"
@@ -1246,7 +1243,7 @@ func TestReplicationSetCommitRestart(t *testing.T) {
 		},
 	}
 	span := tablepb.Span{TableID: 0}
-	r, err := NewReplicationSet(span, tablepb.Checkpoint{}, tableStatus, model.ChangeFeedID{})
+	r, err := NewReplicationSet(span, 0, tableStatus, model.ChangeFeedID{})
 	require.Nil(t, err)
 	require.Equal(t, ReplicationSetStateCommit, r.State)
 	require.EqualValues(t, RoleSecondary, r.Captures["1"])
@@ -1329,7 +1326,7 @@ func TestReplicationSetRemoveRestart(t *testing.T) {
 		},
 	}
 	span := tablepb.Span{TableID: 0}
-	r, err := NewReplicationSet(span, tablepb.Checkpoint{}, tableStatus, model.ChangeFeedID{})
+	r, err := NewReplicationSet(span, 0, tableStatus, model.ChangeFeedID{})
 	require.Nil(t, err)
 	require.Equal(t, ReplicationSetStateRemoving, r.State)
 	require.False(t, r.hasRole(RoleSecondary))
@@ -1458,3 +1455,29 @@ func TestReplicationSetHeap_MinK(t *testing.T) {
 	require.Equal(t, expectedTables, tables)
 	require.Equal(t, 0, h.Len())
 }
+
+func TestUpdateCheckpointAndStats(t *testing.T) {
+	cases := []struct {
+		checkpoint tablepb.Checkpoint
+		stats      tablepb.Stats
+	}{
+		{
+			checkpoint: tablepb.Checkpoint{
+				CheckpointTs: 1,
+				ResolvedTs:   2,
+			},
+			stats: tablepb.Stats{},
+		},
+		{
+			checkpoint: tablepb.Checkpoint{
+				CheckpointTs: 2,
+				ResolvedTs:   1,
+			},
+			stats: tablepb.Stats{},
+		},
+	}
+	r := &ReplicationSet{}
+	for _, c := range cases {
+		r.updateCheckpointAndStats(c.checkpoint, c.stats)
+	}
+}
diff --git a/cdc/scheduler/internal/v3/scheduler/main_test.go b/cdc/scheduler/internal/v3/scheduler/main_test.go
deleted file mode 100644
index 47da658bf..000000000
--- a/cdc/scheduler/internal/v3/scheduler/main_test.go
+++ /dev/null
@@ -1,24 +0,0 @@
-// Copyright 2022 PingCAP, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package scheduler
-
-import (
-	"testing"
-
-	"github.com/pingcap/tiflow/pkg/leakutil"
-)
-
-func TestMain(m *testing.M) {
-	leakutil.SetUpLeakTest(m)
-}
diff --git a/cdc/scheduler/internal/v3/scheduler/scheduler_balance_test.go b/cdc/scheduler/internal/v3/scheduler/scheduler_balance_test.go
index a66b52b32..a8efb07ad 100644
--- a/cdc/scheduler/internal/v3/scheduler/scheduler_balance_test.go
+++ b/cdc/scheduler/internal/v3/scheduler/scheduler_balance_test.go
@@ -18,7 +18,6 @@ import (
 	"time"
 
 	"github.com/pingcap/tiflow/cdc/model"
-	"github.com/pingcap/tiflow/cdc/processor/tablepb"
 	"github.com/pingcap/tiflow/cdc/scheduler/internal/v3/member"
 	"github.com/pingcap/tiflow/cdc/scheduler/internal/v3/replication"
 	"github.com/pingcap/tiflow/pkg/spanz"
@@ -28,7 +27,6 @@ import (
 func TestSchedulerBalanceCaptureOnline(t *testing.T) {
 	t.Parallel()
 
-	var checkpoint tablepb.Checkpoint
 	sched := newBalanceScheduler(time.Duration(0), 3)
 	sched.random = nil
 
@@ -39,14 +37,14 @@ func TestSchedulerBalanceCaptureOnline(t *testing.T) {
 		1: {State: replication.ReplicationSetStateReplicating, Primary: "a"},
 		2: {State: replication.ReplicationSetStateReplicating, Primary: "a"},
 	})
-	tasks := sched.Schedule(checkpoint, currentTables, captures, replications)
+	tasks := sched.Schedule(0, currentTables, captures, replications)
 	require.Len(t, tasks, 1)
 	require.NotNil(t, tasks[0].MoveTable)
 	require.Equal(t, tasks[0].MoveTable.Span.TableID, model.TableID(1))
 
 	// New capture "b" online, but this time has capture is stopping
 	captures["a"].State = member.CaptureStateStopping
-	tasks = sched.Schedule(checkpoint, currentTables, captures, replications)
+	tasks = sched.Schedule(0, currentTables, captures, replications)
 	require.Len(t, tasks, 0)
 
 	// New capture "b" online, it keeps balancing, even though it has not pass
@@ -58,7 +56,7 @@ func TestSchedulerBalanceCaptureOnline(t *testing.T) {
 		1: {State: replication.ReplicationSetStateReplicating, Primary: "a"},
 		2: {State: replication.ReplicationSetStateReplicating, Primary: "a"},
 	})
-	tasks = sched.Schedule(checkpoint, currentTables, captures, replications)
+	tasks = sched.Schedule(0, currentTables, captures, replications)
 	require.Len(t, tasks, 1)
 
 	// New capture "b" online, but this time it not pass check balance interval.
@@ -70,14 +68,13 @@ func TestSchedulerBalanceCaptureOnline(t *testing.T) {
 		1: {State: replication.ReplicationSetStateReplicating, Primary: "a"},
 		2: {State: replication.ReplicationSetStateReplicating, Primary: "a"},
 	})
-	tasks = sched.Schedule(checkpoint, currentTables, captures, replications)
+	tasks = sched.Schedule(0, currentTables, captures, replications)
 	require.Len(t, tasks, 0)
 }
 
 func TestSchedulerBalanceTaskLimit(t *testing.T) {
 	t.Parallel()
 
-	var checkpoint tablepb.Checkpoint
 	sched := newBalanceScheduler(time.Duration(0), 2)
 	sched.random = nil
 
@@ -90,10 +87,10 @@ func TestSchedulerBalanceTaskLimit(t *testing.T) {
 		3: {State: replication.ReplicationSetStateReplicating, Primary: "a"},
 		4: {State: replication.ReplicationSetStateReplicating, Primary: "a"},
 	})
-	tasks := sched.Schedule(checkpoint, currentTables, captures, replications)
+	tasks := sched.Schedule(0, currentTables, captures, replications)
 	require.Len(t, tasks, 2)
 
 	sched = newBalanceScheduler(time.Duration(0), 1)
-	tasks = sched.Schedule(checkpoint, currentTables, captures, replications)
+	tasks = sched.Schedule(0, currentTables, captures, replications)
 	require.Len(t, tasks, 1)
 }
diff --git a/cdc/scheduler/internal/v3/scheduler/scheduler_basic_test.go b/cdc/scheduler/internal/v3/scheduler/scheduler_basic_test.go
index ad8ecd63a..83d00ed36 100644
--- a/cdc/scheduler/internal/v3/scheduler/scheduler_basic_test.go
+++ b/cdc/scheduler/internal/v3/scheduler/scheduler_basic_test.go
@@ -36,7 +36,6 @@ func mapToSpanMap[T any](in map[model.TableID]T) *spanz.BtreeMap[T] {
 func TestSchedulerBasic(t *testing.T) {
 	t.Parallel()
 
-	var checkpoint tablepb.Checkpoint
 	captures := map[model.CaptureID]*member.CaptureStatus{"a": {}, "b": {}}
 	currentTables := spanz.ArrayToSpan([]model.TableID{1, 2, 3, 4})
 
@@ -47,7 +46,7 @@ func TestSchedulerBasic(t *testing.T) {
 
 	// one capture stopping, another one is initialized
 	captures["a"].State = member.CaptureStateStopping
-	tasks := b.Schedule(checkpoint, currentTables, captures, replications)
+	tasks := b.Schedule(0, currentTables, captures, replications)
 	require.Len(t, tasks, 1)
 	require.Len(t, tasks[0].BurstBalance.AddTables, 2)
 	require.Equal(t, tasks[0].BurstBalance.AddTables[0].CaptureID, "b")
@@ -55,12 +54,12 @@ func TestSchedulerBasic(t *testing.T) {
 
 	// all capture's stopping, cannot add table
 	captures["b"].State = member.CaptureStateStopping
-	tasks = b.Schedule(checkpoint, currentTables, captures, replications)
+	tasks = b.Schedule(0, currentTables, captures, replications)
 	require.Len(t, tasks, 0)
 
 	captures["a"].State = member.CaptureStateInitialized
 	captures["b"].State = member.CaptureStateInitialized
-	tasks = b.Schedule(checkpoint, currentTables, captures, replications)
+	tasks = b.Schedule(0, currentTables, captures, replications)
 	require.Len(t, tasks, 1)
 	require.Len(t, tasks[0].BurstBalance.AddTables, 2)
 	require.Equal(t, tasks[0].BurstBalance.AddTables[0].Span.TableID, model.TableID(1))
@@ -89,11 +88,10 @@ func TestSchedulerBasic(t *testing.T) {
 		},
 		4: {State: replication.ReplicationSetStateAbsent},
 	})
-	checkpoint1 := tablepb.Checkpoint{CheckpointTs: 1, ResolvedTs: 1}
-	tasks = b.Schedule(checkpoint1, currentTables, captures, replications)
+	tasks = b.Schedule(1, currentTables, captures, replications)
 	require.Len(t, tasks, 1)
 	require.Equal(t, tasks[0].BurstBalance.AddTables[0].Span.TableID, model.TableID(4))
-	require.Equal(t, tasks[0].BurstBalance.AddTables[0].Checkpoint, checkpoint1)
+	require.Equal(t, tasks[0].BurstBalance.AddTables[0].CheckpointTs, model.Ts(1))
 
 	// DDL CREATE/DROP/TRUNCATE TABLE.
 	// AddTable 4, and RemoveTable 5.
@@ -123,16 +121,15 @@ func TestSchedulerBasic(t *testing.T) {
 			},
 		},
 	})
-	checkpoint2 := tablepb.Checkpoint{CheckpointTs: 2, ResolvedTs: 2}
-	tasks = b.Schedule(checkpoint2, currentTables, captures, replications)
+	tasks = b.Schedule(2, currentTables, captures, replications)
 	require.Len(t, tasks, 2)
 	if tasks[0].BurstBalance.AddTables != nil {
 		require.Equal(t, tasks[0].BurstBalance.AddTables[0].Span.TableID, model.TableID(4))
-		require.Equal(t, tasks[0].BurstBalance.AddTables[0].Checkpoint, checkpoint2)
+		require.Equal(t, tasks[0].BurstBalance.AddTables[0].CheckpointTs, model.Ts(2))
 		require.Equal(t, tasks[1].BurstBalance.RemoveTables[0].Span.TableID, model.TableID(5))
 	} else {
 		require.Equal(t, tasks[1].BurstBalance.AddTables[0].Span.TableID, model.TableID(4))
-		require.Equal(t, tasks[0].BurstBalance.AddTables[0].Checkpoint, checkpoint2)
+		require.Equal(t, tasks[0].BurstBalance.AddTables[0].CheckpointTs, model.Ts(2))
 		require.Equal(t, tasks[0].BurstBalance.RemoveTables[0].Span.TableID, model.TableID(5))
 	}
 
@@ -169,8 +166,7 @@ func TestSchedulerBasic(t *testing.T) {
 			},
 		},
 	})
-	checkpoint3 := tablepb.Checkpoint{CheckpointTs: 3, ResolvedTs: 3}
-	tasks = b.Schedule(checkpoint3, currentTables, captures, replications)
+	tasks = b.Schedule(3, currentTables, captures, replications)
 	require.Len(t, tasks, 1)
 	require.Equal(t, tasks[0].BurstBalance.RemoveTables[0].Span.TableID, model.TableID(5))
 }
@@ -197,13 +193,12 @@ func benchmarkSchedulerBalance(
 	),
 ) {
 	size := 16384
-	var checkpoint tablepb.Checkpoint
 	for total := 1; total <= size; total *= 2 {
 		name, currentTables, captures, replications, sched := factory(total)
 		b.ResetTimer()
 		b.Run(name, func(b *testing.B) {
 			for i := 0; i < b.N; i++ {
-				sched.Schedule(checkpoint, currentTables, captures, replications)
+				sched.Schedule(0, currentTables, captures, replications)
 			}
 		})
 		b.StopTimer()
diff --git a/cdc/scheduler/internal/v3/scheduler/scheduler_drain_capture_test.go b/cdc/scheduler/internal/v3/scheduler/scheduler_drain_capture_test.go
index 81af850e2..03f1a9919 100644
--- a/cdc/scheduler/internal/v3/scheduler/scheduler_drain_capture_test.go
+++ b/cdc/scheduler/internal/v3/scheduler/scheduler_drain_capture_test.go
@@ -30,18 +30,18 @@ func TestDrainCapture(t *testing.T) {
 	scheduler := newDrainCaptureScheduler(10, model.ChangeFeedID{})
 	require.Equal(t, "drain-capture-scheduler", scheduler.Name())
 
-	var checkpoint tablepb.Checkpoint
+	var checkpointTs model.Ts
 	captures := make(map[model.CaptureID]*member.CaptureStatus)
 	currentTables := make([]tablepb.Span, 0)
 	replications := mapToSpanMap(make(map[model.TableID]*replication.ReplicationSet))
 
-	tasks := scheduler.Schedule(checkpoint, currentTables, captures, replications)
+	tasks := scheduler.Schedule(checkpointTs, currentTables, captures, replications)
 	require.Len(t, tasks, 0)
 
 	ok := scheduler.setTarget("a")
 	require.True(t, ok)
 
-	tasks = scheduler.Schedule(checkpoint, currentTables, captures, replications)
+	tasks = scheduler.Schedule(checkpointTs, currentTables, captures, replications)
 	require.Len(t, tasks, 0)
 	// the target capture has no table at the beginning, so reset the target
 	require.Equal(t, captureIDNotDraining, scheduler.target)
@@ -50,7 +50,7 @@ func TestDrainCapture(t *testing.T) {
 	ok = scheduler.setTarget("b")
 	require.True(t, ok)
 
-	tasks = scheduler.Schedule(checkpoint, currentTables, captures, replications)
+	tasks = scheduler.Schedule(checkpointTs, currentTables, captures, replications)
 	require.Len(t, tasks, 0)
 	// the target capture cannot be found in the latest captures
 	require.Equal(t, captureIDNotDraining, scheduler.target)
@@ -105,7 +105,7 @@ func TestDrainCapture(t *testing.T) {
 	ok = scheduler.setTarget("a")
 	require.True(t, ok)
 	// not all table is replicating, skip this tick.
-	tasks = scheduler.Schedule(checkpoint, currentTables, captures, replications)
+	tasks = scheduler.Schedule(checkpointTs, currentTables, captures, replications)
 	require.Equal(t, "a", scheduler.target)
 	require.Len(t, tasks, 0)
 
@@ -118,13 +118,13 @@ func TestDrainCapture(t *testing.T) {
 		7: {State: replication.ReplicationSetStateReplicating, Primary: "b"},
 	})
 
-	tasks = scheduler.Schedule(checkpoint, currentTables, captures, replications)
+	tasks = scheduler.Schedule(checkpointTs, currentTables, captures, replications)
 	require.Equal(t, "a", scheduler.target)
 	require.Len(t, tasks, 3)
 
 	scheduler = newDrainCaptureScheduler(1, model.ChangeFeedID{})
 	require.True(t, scheduler.setTarget("a"))
-	tasks = scheduler.Schedule(checkpoint, currentTables, captures, replications)
+	tasks = scheduler.Schedule(checkpointTs, currentTables, captures, replications)
 	require.Equal(t, "a", scheduler.target)
 	require.Len(t, tasks, 1)
 }
@@ -132,13 +132,13 @@ func TestDrainCapture(t *testing.T) {
 func TestDrainStoppingCapture(t *testing.T) {
 	t.Parallel()
 
-	var checkpoint tablepb.Checkpoint
+	var checkpointTs model.Ts
 	captures := make(map[model.CaptureID]*member.CaptureStatus)
 	currentTables := make([]tablepb.Span, 0)
 	replications := mapToSpanMap(make(map[model.TableID]*replication.ReplicationSet))
 	scheduler := newDrainCaptureScheduler(10, model.ChangeFeedID{})
 
-	tasks := scheduler.Schedule(checkpoint, currentTables, captures, replications)
+	tasks := scheduler.Schedule(checkpointTs, currentTables, captures, replications)
 	require.Empty(t, tasks)
 
 	captures["a"] = &member.CaptureStatus{}
@@ -147,7 +147,7 @@ func TestDrainStoppingCapture(t *testing.T) {
 		1: {State: replication.ReplicationSetStateReplicating, Primary: "a"},
 		2: {State: replication.ReplicationSetStateReplicating, Primary: "b"},
 	})
-	tasks = scheduler.Schedule(checkpoint, currentTables, captures, replications)
+	tasks = scheduler.Schedule(checkpointTs, currentTables, captures, replications)
 	require.Len(t, tasks, 1)
 	require.EqualValues(t, 2, tasks[0].MoveTable.Span.TableID)
 	require.EqualValues(t, "a", tasks[0].MoveTable.DestCapture)
@@ -157,7 +157,7 @@ func TestDrainStoppingCapture(t *testing.T) {
 func TestDrainSkipOwner(t *testing.T) {
 	t.Parallel()
 
-	var checkpoint tablepb.Checkpoint
+	var checkpointTs model.Ts
 	currentTables := make([]tablepb.Span, 0)
 	captures := map[model.CaptureID]*member.CaptureStatus{
 		"a": {},
@@ -168,7 +168,7 @@ func TestDrainSkipOwner(t *testing.T) {
 		2: {State: replication.ReplicationSetStateReplicating, Primary: "b"},
 	})
 	scheduler := newDrainCaptureScheduler(10, model.ChangeFeedID{})
-	tasks := scheduler.Schedule(checkpoint, currentTables, captures, replications)
+	tasks := scheduler.Schedule(checkpointTs, currentTables, captures, replications)
 	require.Len(t, tasks, 0)
 	require.EqualValues(t, captureIDNotDraining, scheduler.getTarget())
 }
@@ -176,7 +176,7 @@ func TestDrainSkipOwner(t *testing.T) {
 func TestDrainImbalanceCluster(t *testing.T) {
 	t.Parallel()
 
-	var checkpoint tablepb.Checkpoint
+	var checkpointTs model.Ts
 	currentTables := make([]tablepb.Span, 0)
 	captures := map[model.CaptureID]*member.CaptureStatus{
 		"a": {State: member.CaptureStateInitialized},
@@ -188,7 +188,7 @@ func TestDrainImbalanceCluster(t *testing.T) {
 	})
 	scheduler := newDrainCaptureScheduler(10, model.ChangeFeedID{})
 	scheduler.setTarget("a")
-	tasks := scheduler.Schedule(checkpoint, currentTables, captures, replications)
+	tasks := scheduler.Schedule(checkpointTs, currentTables, captures, replications)
 	require.Len(t, tasks, 2)
 	require.EqualValues(t, "a", scheduler.getTarget())
 }
@@ -196,7 +196,7 @@ func TestDrainImbalanceCluster(t *testing.T) {
 func TestDrainEvenlyDistributedTables(t *testing.T) {
 	t.Parallel()
 
-	var checkpoint tablepb.Checkpoint
+	var checkpointTs model.Ts
 	currentTables := make([]tablepb.Span, 0)
 	captures := map[model.CaptureID]*member.CaptureStatus{
 		"a": {State: member.CaptureStateInitialized},
@@ -211,7 +211,7 @@ func TestDrainEvenlyDistributedTables(t *testing.T) {
 	})
 	scheduler := newDrainCaptureScheduler(10, model.ChangeFeedID{})
 	scheduler.setTarget("a")
-	tasks := scheduler.Schedule(checkpoint, currentTables, captures, replications)
+	tasks := scheduler.Schedule(checkpointTs, currentTables, captures, replications)
 	require.Len(t, tasks, 3)
 	taskMap := make(map[model.CaptureID]int)
 	for _, t := range tasks {
diff --git a/cdc/scheduler/internal/v3/scheduler/scheduler_manager_test.go b/cdc/scheduler/internal/v3/scheduler/scheduler_manager_test.go
index 613273b81..e7b3fb83a 100644
--- a/cdc/scheduler/internal/v3/scheduler/scheduler_manager_test.go
+++ b/cdc/scheduler/internal/v3/scheduler/scheduler_manager_test.go
@@ -18,7 +18,6 @@ import (
 
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/processor/tablepb"
-	"github.com/pingcap/tiflow/cdc/redo"
 	"github.com/pingcap/tiflow/cdc/scheduler/internal/v3/member"
 	"github.com/pingcap/tiflow/cdc/scheduler/internal/v3/replication"
 	"github.com/pingcap/tiflow/pkg/config"
@@ -54,16 +53,14 @@ func TestSchedulerManagerScheduler(t *testing.T) {
 	// schedulerPriorityBasic bypasses task check.
 	replications := mapToSpanMap(map[model.TableID]*replication.ReplicationSet{})
 	runningTasks := mapToSpanMap(map[model.TableID]*replication.ScheduleTask{1: {}})
-
-	redoMetaManager := redo.NewDisabledMetaManager()
-	tasks := m.Schedule(0, currentSpans, captures, replications, runningTasks, redoMetaManager)
+	tasks := m.Schedule(0, currentSpans, captures, replications, runningTasks)
 	require.Len(t, tasks, 1)
 
 	// No more task.
 	replications = mapToSpanMap(map[model.TableID]*replication.ReplicationSet{
 		1: {State: replication.ReplicationSetStateReplicating, Primary: "a"},
 	})
-	tasks = m.Schedule(0, currentSpans, captures, replications, runningTasks, redoMetaManager)
+	tasks = m.Schedule(0, currentSpans, captures, replications, runningTasks)
 	require.Len(t, tasks, 0)
 
 	// Move table is dropped because of running tasks.
@@ -71,7 +68,7 @@ func TestSchedulerManagerScheduler(t *testing.T) {
 	replications = mapToSpanMap(map[model.TableID]*replication.ReplicationSet{
 		1: {State: replication.ReplicationSetStateReplicating, Primary: "a"},
 	})
-	tasks = m.Schedule(0, currentSpans, captures, replications, runningTasks, redoMetaManager)
+	tasks = m.Schedule(0, currentSpans, captures, replications, runningTasks)
 	require.Len(t, tasks, 0)
 
 	// Move table can proceed after clean up tasks.
@@ -80,6 +77,6 @@ func TestSchedulerManagerScheduler(t *testing.T) {
 		1: {State: replication.ReplicationSetStateReplicating, Primary: "a"},
 	})
 	runningTasks = spanz.NewBtreeMap[*replication.ScheduleTask]()
-	tasks = m.Schedule(0, currentSpans, captures, replications, runningTasks, redoMetaManager)
+	tasks = m.Schedule(0, currentSpans, captures, replications, runningTasks)
 	require.Len(t, tasks, 1)
 }
diff --git a/cdc/scheduler/internal/v3/scheduler/scheduler_move_table_test.go b/cdc/scheduler/internal/v3/scheduler/scheduler_move_table_test.go
index 3adc047bf..e485db9eb 100644
--- a/cdc/scheduler/internal/v3/scheduler/scheduler_move_table_test.go
+++ b/cdc/scheduler/internal/v3/scheduler/scheduler_move_table_test.go
@@ -27,7 +27,7 @@ import (
 func TestSchedulerMoveTable(t *testing.T) {
 	t.Parallel()
 
-	var checkpoint tablepb.Checkpoint
+	var checkpointTs model.Ts
 	captures := map[model.CaptureID]*member.CaptureStatus{"a": {
 		State: member.CaptureStateInitialized,
 	}, "b": {
@@ -43,38 +43,38 @@ func TestSchedulerMoveTable(t *testing.T) {
 	require.Equal(t, "move-table-scheduler", scheduler.Name())
 
 	tasks := scheduler.Schedule(
-		checkpoint, currentTables, map[model.CaptureID]*member.CaptureStatus{}, replications)
+		checkpointTs, currentTables, map[model.CaptureID]*member.CaptureStatus{}, replications)
 	require.Len(t, tasks, 0)
 
 	scheduler.addTask(tablepb.Span{TableID: 0}, "a")
 	tasks = scheduler.Schedule(
-		checkpoint, currentTables, map[model.CaptureID]*member.CaptureStatus{}, replications)
+		checkpointTs, currentTables, map[model.CaptureID]*member.CaptureStatus{}, replications)
 	require.Len(t, tasks, 0)
 
 	// move a not exist table
 	scheduler.addTask(tablepb.Span{TableID: 0}, "a")
-	tasks = scheduler.Schedule(checkpoint, currentTables, captures, replications)
+	tasks = scheduler.Schedule(checkpointTs, currentTables, captures, replications)
 	require.Len(t, tasks, 0)
 
 	// move table to a not exist capture
 	scheduler.addTask(tablepb.Span{TableID: 1}, "c")
-	tasks = scheduler.Schedule(checkpoint, currentTables, captures, replications)
+	tasks = scheduler.Schedule(checkpointTs, currentTables, captures, replications)
 	require.Len(t, tasks, 0)
 
 	// move table not replicating
 	scheduler.addTask(tablepb.Span{TableID: 1}, "b")
 	tasks = scheduler.Schedule(
-		checkpoint, currentTables, captures, spanz.NewBtreeMap[*replication.ReplicationSet]())
+		checkpointTs, currentTables, captures, spanz.NewBtreeMap[*replication.ReplicationSet]())
 	require.Len(t, tasks, 0)
 
 	scheduler.addTask(tablepb.Span{TableID: 1}, "b")
 	replications.GetV(tablepb.Span{TableID: 1}).State = replication.ReplicationSetStatePrepare
-	tasks = scheduler.Schedule(checkpoint, currentTables, captures, replications)
+	tasks = scheduler.Schedule(checkpointTs, currentTables, captures, replications)
 	require.Len(t, tasks, 0)
 
 	scheduler.addTask(tablepb.Span{TableID: 1}, "b")
 	replications.GetV(tablepb.Span{TableID: 1}).State = replication.ReplicationSetStateReplicating
-	tasks = scheduler.Schedule(checkpoint, currentTables, captures, replications)
+	tasks = scheduler.Schedule(checkpointTs, currentTables, captures, replications)
 	require.Len(t, tasks, 1)
 	require.Equal(t, model.TableID(1), tasks[0].MoveTable.Span.TableID)
 	require.Equal(t, "b", tasks[0].MoveTable.DestCapture)
@@ -83,7 +83,7 @@ func TestSchedulerMoveTable(t *testing.T) {
 	// the target capture is stopping
 	scheduler.addTask(tablepb.Span{TableID: 1}, "b")
 	captures["b"].State = member.CaptureStateStopping
-	tasks = scheduler.Schedule(checkpoint, currentTables, captures, replications)
+	tasks = scheduler.Schedule(checkpointTs, currentTables, captures, replications)
 	require.Len(t, tasks, 0)
 	require.False(t, scheduler.tasks.Has(tablepb.Span{TableID: 1}))
 }
diff --git a/cdc/scheduler/internal/v3/scheduler/scheduler_rebalance_test.go b/cdc/scheduler/internal/v3/scheduler/scheduler_rebalance_test.go
index c2b3ea9c8..e6b1ff53d 100644
--- a/cdc/scheduler/internal/v3/scheduler/scheduler_rebalance_test.go
+++ b/cdc/scheduler/internal/v3/scheduler/scheduler_rebalance_test.go
@@ -28,7 +28,7 @@ import (
 func TestSchedulerRebalance(t *testing.T) {
 	t.Parallel()
 
-	var checkpoint tablepb.Checkpoint
+	var checkpointTs model.Ts
 	captures := map[model.CaptureID]*member.CaptureStatus{"a": {}, "b": {}}
 	currentTables := spanz.ArrayToSpan([]model.TableID{1, 2, 3, 4})
 
@@ -57,22 +57,22 @@ func TestSchedulerRebalance(t *testing.T) {
 	scheduler := newRebalanceScheduler(model.ChangeFeedID{})
 	require.Equal(t, "rebalance-scheduler", scheduler.Name())
 	// rebalance is not triggered
-	tasks := scheduler.Schedule(checkpoint, currentTables, captures, replications)
+	tasks := scheduler.Schedule(checkpointTs, currentTables, captures, replications)
 	require.Len(t, tasks, 0)
 
 	atomic.StoreInt32(&scheduler.rebalance, 1)
 	// no captures
 	tasks = scheduler.Schedule(
-		checkpoint, currentTables, map[model.CaptureID]*member.CaptureStatus{}, replications)
+		checkpointTs, currentTables, map[model.CaptureID]*member.CaptureStatus{}, replications)
 	require.Len(t, tasks, 0)
 
 	// table not in the replication set,
 	tasks = scheduler.Schedule(
-		checkpoint, spanz.ArrayToSpan([]model.TableID{0}), captures, replications)
+		checkpointTs, spanz.ArrayToSpan([]model.TableID{0}), captures, replications)
 	require.Len(t, tasks, 0)
 
 	// not all tables are replicating,
-	tasks = scheduler.Schedule(checkpoint, currentTables, captures, replications)
+	tasks = scheduler.Schedule(checkpointTs, currentTables, captures, replications)
 	require.Len(t, tasks, 0)
 
 	// table distribution is balanced, should have no task.
@@ -82,7 +82,7 @@ func TestSchedulerRebalance(t *testing.T) {
 		3: {State: replication.ReplicationSetStateReplicating, Primary: "b"},
 		4: {State: replication.ReplicationSetStateReplicating, Primary: "b"},
 	})
-	tasks = scheduler.Schedule(checkpoint, currentTables, captures, replications)
+	tasks = scheduler.Schedule(checkpointTs, currentTables, captures, replications)
 	require.Len(t, tasks, 0)
 
 	// Imbalance.
@@ -97,14 +97,14 @@ func TestSchedulerRebalance(t *testing.T) {
 
 	// capture is stopping, ignore the request
 	captures["a"].State = member.CaptureStateStopping
-	tasks = scheduler.Schedule(checkpoint, currentTables, captures, replications)
+	tasks = scheduler.Schedule(checkpointTs, currentTables, captures, replications)
 	require.Len(t, tasks, 0)
 	require.Equal(t, atomic.LoadInt32(&scheduler.rebalance), int32(0))
 
 	captures["a"].State = member.CaptureStateInitialized
 	atomic.StoreInt32(&scheduler.rebalance, 1)
 	scheduler.random = nil // disable random to make test easier.
-	tasks = scheduler.Schedule(checkpoint, currentTables, captures, replications)
+	tasks = scheduler.Schedule(checkpointTs, currentTables, captures, replications)
 	require.Len(t, tasks, 1)
 	require.Contains(t, tasks[0].BurstBalance.MoveTables, replication.MoveTable{
 		Span: tablepb.Span{TableID: 1}, DestCapture: "b",
@@ -115,6 +115,6 @@ func TestSchedulerRebalance(t *testing.T) {
 	require.EqualValues(t, 0, atomic.LoadInt32(&scheduler.rebalance))
 
 	// pending task is not consumed yet, this turn should have no tasks.
-	tasks = scheduler.Schedule(checkpoint, currentTables, captures, replications)
+	tasks = scheduler.Schedule(checkpointTs, currentTables, captures, replications)
 	require.Len(t, tasks, 0)
 }
diff --git a/cdc/scheduler/internal/v3/transport/main_test.go b/cdc/scheduler/internal/v3/transport/main_test.go
deleted file mode 100644
index bdb388e4c..000000000
--- a/cdc/scheduler/internal/v3/transport/main_test.go
+++ /dev/null
@@ -1,24 +0,0 @@
-// Copyright 2022 PingCAP, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package transport
-
-import (
-	"testing"
-
-	"github.com/pingcap/tiflow/pkg/leakutil"
-)
-
-func TestMain(m *testing.M) {
-	leakutil.SetUpLeakTest(m)
-}
diff --git a/cdc/sink/ddlsink/mq/mq_ddl_sink_test.go b/cdc/sink/ddlsink/mq/mq_ddl_sink_test.go
index 4403128cd..6408a1a7e 100644
--- a/cdc/sink/ddlsink/mq/mq_ddl_sink_test.go
+++ b/cdc/sink/ddlsink/mq/mq_ddl_sink_test.go
@@ -59,8 +59,9 @@ func TestWriteDDLEventToAllPartitions(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
+	// 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, "127.0.0.1:9092", kafka.DefaultMockTopicName)
 
@@ -89,10 +90,9 @@ func TestWriteDDLEventToAllPartitions(t *testing.T) {
 	err = s.WriteDDLEvent(ctx, ddl)
 	require.NoError(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("mock_topic", 0), 1)
 	require.Len(t, s.producer.(*ddlproducer.MockDDLProducer).GetEvents("mock_topic", 1), 1)
-	require.Len(t, s.producer.(*ddlproducer.MockDDLProducer).GetEvents("mock_topic", 2), 1)
 }
 
 func TestWriteDDLEventToZeroPartition(t *testing.T) {
@@ -144,8 +144,9 @@ func TestWriteCheckpointTsToDefaultTopic(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
+	// 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, "127.0.0.1:9092", kafka.DefaultMockTopicName)
@@ -169,10 +170,9 @@ 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("mock_topic", 0), 1)
 	require.Len(t, s.producer.(*ddlproducer.MockDDLProducer).GetEvents("mock_topic", 1), 1)
-	require.Len(t, s.producer.(*ddlproducer.MockDDLProducer).GetEvents("mock_topic", 2), 1)
 }
 
 func TestWriteCheckpointTsToTableTopics(t *testing.T) {
@@ -233,10 +233,8 @@ func TestWriteCheckpointTsToTableTopics(t *testing.T) {
 	require.NoError(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("mock_topic", 0), 1)
-	require.Len(t, s.producer.(*ddlproducer.MockDDLProducer).GetEvents("mock_topic", 1), 1)
-	require.Len(t, s.producer.(*ddlproducer.MockDDLProducer).GetEvents("mock_topic", 2), 1)
 	require.Len(t, s.producer.(*ddlproducer.MockDDLProducer).GetEvents("cdc_person", 0), 1)
 	require.Len(t, s.producer.(*ddlproducer.MockDDLProducer).GetEvents("cdc_person1", 0), 1)
 	require.Len(t, s.producer.(*ddlproducer.MockDDLProducer).GetEvents("cdc_person2", 0), 1)
diff --git a/cdc/sink/dmlsink/mq/manager/kafka_manager_test.go b/cdc/sink/dmlsink/mq/manager/kafka_manager_test.go
index 837ca1139..d4fb88550 100644
--- a/cdc/sink/dmlsink/mq/manager/kafka_manager_test.go
+++ b/cdc/sink/dmlsink/mq/manager/kafka_manager_test.go
@@ -22,28 +22,6 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestPartitions(t *testing.T) {
-	t.Parallel()
-
-	adminClient := kafka.NewClusterAdminClientMockImpl()
-	defer adminClient.Close()
-	cfg := &kafka.AutoCreateTopicConfig{
-		AutoCreate:        true,
-		PartitionNum:      2,
-		ReplicationFactor: 1,
-	}
-
-	ctx := context.Background()
-	manager := NewKafkaTopicManager(ctx, model.DefaultChangeFeedID("test"), adminClient, cfg)
-	defer manager.Close()
-
-	partitionsNum, err := manager.GetPartitionNum(
-		ctx,
-		kafka.DefaultMockTopicName)
-	require.Nil(t, err)
-	require.Equal(t, int32(3), partitionsNum)
-}
-
 func TestCreateTopic(t *testing.T) {
 	t.Parallel()
 
@@ -55,23 +33,24 @@ func TestCreateTopic(t *testing.T) {
 		ReplicationFactor: 1,
 	}
 
+	changefeedID := model.DefaultChangeFeedID("test")
 	ctx := context.Background()
-	manager := NewKafkaTopicManager(ctx, model.DefaultChangeFeedID("test"), adminClient, cfg)
+	manager := NewKafkaTopicManager(ctx, kafka.DefaultMockTopicName, changefeedID, adminClient, cfg)
 	defer manager.Close()
 	partitionNum, err := manager.CreateTopicAndWaitUntilVisible(ctx, 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(ctx, "new-topic")
-	require.Nil(t, err)
+	require.NoError(t, err)
 	require.Equal(t, int32(2), partitionNum)
 	partitionsNum, err := manager.GetPartitionNum(ctx, "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, model.DefaultChangeFeedID("test"), adminClient, cfg)
+	manager = NewKafkaTopicManager(ctx, "new-topic2", changefeedID, adminClient, cfg)
 	defer manager.Close()
 	_, err = manager.CreateTopicAndWaitUntilVisible(ctx, "new-topic2")
 	require.Regexp(
@@ -80,6 +59,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 = &kafka.AutoCreateTopicConfig{
@@ -87,9 +67,9 @@ func TestCreateTopic(t *testing.T) {
 		PartitionNum:      2,
 		ReplicationFactor: 4,
 	}
-	manager = NewKafkaTopicManager(ctx, model.DefaultChangeFeedID("test"), adminClient, cfg)
+	manager = NewKafkaTopicManager(ctx, topic, changefeedID, adminClient, cfg)
 	defer manager.Close()
-	_, err = manager.CreateTopicAndWaitUntilVisible(ctx, "new-topic-failed")
+	_, err = manager.CreateTopicAndWaitUntilVisible(ctx, topic)
 	require.Regexp(
 		t,
 		"kafka create topic failed: kafka server: Replication-factor is invalid",
@@ -108,14 +88,16 @@ func TestCreateTopicWithDelay(t *testing.T) {
 		ReplicationFactor: 1,
 	}
 
+	topic := "new_topic"
+	changefeedID := model.DefaultChangeFeedID("test")
 	ctx := context.Background()
-	manager := NewKafkaTopicManager(ctx, model.DefaultChangeFeedID("test"), adminClient, cfg)
+	manager := NewKafkaTopicManager(ctx, topic, changefeedID, adminClient, cfg)
 	defer manager.Close()
-	partitionNum, err := manager.createTopic(ctx, "new_topic")
-	require.Nil(t, err)
-	err = adminClient.SetRemainingFetchesUntilTopicVisible("new_topic", 3)
-	require.Nil(t, err)
-	err = manager.waitUntilTopicVisible(ctx, "new_topic")
-	require.Nil(t, err)
+	partitionNum, err := manager.createTopic(ctx, topic)
+	require.NoError(t, err)
+	err = adminClient.SetRemainingFetchesUntilTopicVisible(topic, 3)
+	require.NoError(t, err)
+	err = manager.waitUntilTopicVisible(ctx, topic)
+	require.NoError(t, err)
 	require.Equal(t, int32(2), partitionNum)
 }
diff --git a/cdc/sink/util/helper_test.go b/cdc/sink/util/helper_test.go
index ff07f9c2d..31f9f7cc8 100644
--- a/cdc/sink/util/helper_test.go
+++ b/cdc/sink/util/helper_test.go
@@ -14,12 +14,56 @@
 package util
 
 import (
+	"context"
 	"net/url"
 	"testing"
 
+	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/pingcap/tiflow/pkg/sink/kafka"
 	"github.com/stretchr/testify/require"
 )
 
+func TestPartition(t *testing.T) {
+	t.Parallel()
+
+	adminClient := kafka.NewClusterAdminClientMockImpl()
+	defer adminClient.Close()
+
+	cfg := &kafka.AutoCreateTopicConfig{
+		AutoCreate:        true,
+		PartitionNum:      2,
+		ReplicationFactor: 1,
+	}
+
+	changefeedID := model.DefaultChangeFeedID("test")
+	ctx := context.Background()
+
+	manager, err := GetTopicManagerAndTryCreateTopic(ctx, changefeedID, kafka.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(ctx, kafka.DefaultMockTopicName)
+	require.NoError(t, err)
+	require.Equal(t, int32(2), partitionsNum)
+
+	// new topic, create it with partition number as 2.
+	partitionsNum, err = manager.GetPartitionNum(ctx, "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(ctx, &kafka.TopicDetail{
+		Name:          "new-topic-2",
+		NumPartitions: 3,
+	}, false)
+	require.NoError(t, err)
+
+	partitionsNum, err = manager.GetPartitionNum(ctx, "new-topic-2")
+	require.NoError(t, err)
+	require.Equal(t, int32(3), partitionsNum)
+}
+
 func TestGetTopic(t *testing.T) {
 	t.Parallel()
 
diff --git a/dm/pkg/gtid/gtid_test.go b/dm/pkg/gtid/gtid_test.go
index 275b82527..79abed6e0 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 TestCheckGTIDSetEmpty(t *testing.T) {
+	t.Parallel()
+
+	testCases := []struct {
+		gsetStr string
+		isEmpty 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.isEmpty, CheckGTIDSetEmpty(gset))
+	}
+}
diff --git a/dm/tests/incompatible_ddl_changes/conf/dm-task1.yaml b/dm/tests/incompatible_ddl_changes/conf/dm-task1.yaml
index 7b409c406..f2de75176 100644
--- a/dm/tests/incompatible_ddl_changes/conf/dm-task1.yaml
+++ b/dm/tests/incompatible_ddl_changes/conf/dm-task1.yaml
@@ -23,14 +23,22 @@ filters:
     table-pattern: "*"
     events:
       [
+        "drop database",
+        "drop table",
+        "drop index",
+        "rename table",
+        "truncate table",
+        "drop table partition",
+        "truncate table partition",
         "value range decrease",
         "precision decrease",
         "modify column",
-        "rename",
-        "drop",
-        "truncate",
-        "modify pk",
-        "modify uk",
+        "rename column",
+        "rename index",
+        "drop column",
+        "truncate table partition",
+        "drop primary key",
+        "drop unique key",
         "modify default value",
         "modify constaints",
         "modify columns order",
@@ -38,11 +46,11 @@ filters:
         "modify collation",
         "remove auto increment",
         "modify storage engine",
-        "reorganize partition",
-        "coalesce partition",
-        "split partition",
-        "exchange partition",
-        "rebuild partition",
+        "reorganize table partition",
+        "coalesce table partition",
+        "split table partition",
+        "exchange table partition",
+        "rebuild table partition",
       ]
     sql-pattern: []
     action: Error
diff --git a/dm/tests/incompatible_ddl_changes/run.sh b/dm/tests/incompatible_ddl_changes/run.sh
index 538a168f7..90710ff44 100755
--- a/dm/tests/incompatible_ddl_changes/run.sh
+++ b/dm/tests/incompatible_ddl_changes/run.sh
@@ -74,23 +74,23 @@ function incompatible_ddl() {
 	run_sql_source1 "alter table incompatible_ddl_changes.t1 change c_mediumint c_mediumint_new mediumint(7);"
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"query-status test" \
-		"event rename" 1
+		"event rename column" 1
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"binlog skip test" \
 		"\"result\": true" 2
 	run_sql_source1 "alter table incompatible_ddl_changes.t1 change c_mediumint_new c_mediumint mediumint(7);"
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"query-status test" \
-		"event rename" 1
+		"event rename column" 1
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"binlog skip test" \
 		"\"result\": true" 2
 
-	# drop
+	# drop column
 	run_sql_source1 "alter table incompatible_ddl_changes.t1 drop column c_json;"
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"query-status test" \
-		"event drop" 1
+		"event drop column" 1
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"binlog skip test" \
 		"\"result\": true" 2
@@ -105,11 +105,11 @@ function incompatible_ddl() {
 		"binlog skip test" \
 		"\"result\": true" 2
 
-	# modify pk
+	# drop pk
 	run_sql_source1 "alter table incompatible_ddl_changes.t1 drop primary key"
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"query-status test" \
-		"event modify pk" 1
+		"event drop primary key" 1
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"binlog skip test" \
 		"\"result\": true" 2
@@ -132,18 +132,19 @@ function incompatible_ddl() {
 		"binlog skip test" \
 		"\"result\": true" 2
 
-	# modify uk
+	# drop uk
 	run_sql_source1 "alter table incompatible_ddl_changes.t1 change c_int c_int int"
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"query-status test" \
-		"modify uk" 1
+		"drop unique key" 1
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"binlog skip test" \
 		"\"result\": true" 2
+	# drop index
 	run_sql_source1 "alter table incompatible_ddl_changes.t1 drop index c_int_unique"
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"query-status test" \
-		"event drop" 1
+		"event drop index" 1
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"binlog skip test" \
 		"\"result\": true" 2
@@ -189,16 +190,16 @@ function incompatible_ddl() {
 	run_sql_source1 "alter table incompatible_ddl_changes.t1 reorganize partition p0 into ( partition n0 values less than (5), partition n1 values less than (100000));"
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"query-status test" \
-		"event reorganize partition" 1
+		"event reorganize table partition" 1
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"binlog skip test" \
 		"\"result\": true" 2
 
-	# truncate
+	# truncate partition
 	run_sql_source1 "alter table incompatible_ddl_changes.t1 truncate partition n0"
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"query-status test" \
-		"event truncate" 1
+		"event truncate table partition" 1
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"binlog skip test" \
 		"\"result\": true" 2
@@ -207,18 +208,18 @@ function incompatible_ddl() {
 	run_sql_source1 "alter table incompatible_ddl_changes.t1 rebuild partition n0,n1;"
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"query-status test" \
-		"event rebuild partition" 1
+		"event rebuild table partition" 1
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"binlog skip test" \
 		"\"result\": true" 2
 
 	# exchange partition
-	run_sql_source1 "create table incompatible_ddl_changes.tb1(id int) partition by range(id)(partition p0 values less than (100000));"
+	run_sql_source1 "create table incompatible_ddl_changes.tb1(id int) partition by range(id)(partition p0 values less than (100000), partition p1 values less than(1000000));"
 	run_sql_source1 "create table incompatible_ddl_changes.tb2(id int);"
 	run_sql_source1 "alter table incompatible_ddl_changes.tb1 exchange partition p0 with table incompatible_ddl_changes.tb2"
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"query-status test" \
-		"event exchange partition" 1
+		"event exchange table partition" 1
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"binlog skip test" \
 		"\"result\": true" 2
@@ -228,7 +229,68 @@ function incompatible_ddl() {
 	run_sql_source1 "alter table incompatible_ddl_changes.tb3 coalesce partition 2;"
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"query-status test" \
-		"event coalesce partition" 1
+		"event coalesce table partition" 1
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"binlog skip test" \
+		"\"result\": true" 2
+
+	# rename index
+	run_sql_source1 "alter table incompatible_ddl_changes.tb3 add index idx(id);"
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"query-status test" \
+		"event modify constaints" 1
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"binlog skip test" \
+		"\"result\": true" 2
+	run_sql_source1 "alter table incompatible_ddl_changes.tb3 rename index idx to idx1;"
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"query-status test" \
+		"event rename index" 1
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"binlog skip test" \
+		"\"result\": true" 2
+
+	# drop partition
+	run_sql_source1 "alter table incompatible_ddl_changes.tb1 drop partition p0;"
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"query-status test" \
+		"event drop table partition" 1
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"binlog skip test" \
+		"\"result\": true" 2
+
+	# rename table
+	run_sql_source1 "rename table incompatible_ddl_changes.tb3 to incompatible_ddl_changes.tb4;"
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"query-status test" \
+		"event rename table" 1
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"binlog skip test" \
+		"\"result\": true" 2
+
+	# truncate table
+	run_sql_source1 "truncate table incompatible_ddl_changes.tb4;"
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"query-status test" \
+		"event truncate table" 1
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"binlog skip test" \
+		"\"result\": true" 2
+
+	# drop table
+	run_sql_source1 "drop table incompatible_ddl_changes.tb4;"
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"query-status test" \
+		"event drop table" 1
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"binlog skip test" \
+		"\"result\": true" 2
+
+	# drop database
+	run_sql_source1 "drop database incompatible_ddl_changes;"
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"query-status test" \
+		"event drop database" 1
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"binlog skip test" \
 		"\"result\": true" 2
diff --git a/pkg/config/config_test_data.go b/pkg/config/config_test_data.go
index 18d803b30..b28c28395 100644
--- a/pkg/config/config_test_data.go
+++ b/pkg/config/config_test_data.go
@@ -70,6 +70,7 @@ const (
     "level": "none",
     "max-log-size": 64,
     "flush-interval": 2000,
+    "meta-flush-interval": 200,
     "storage": "",
     "use-file-backend": false
   },
@@ -81,7 +82,8 @@ const (
     "integrity-check-level": "none",
     "corruption-handle-level": "warn"
  },
-  "changefeed-error-stuck-duration": 1800000000000
+  "changefeed-error-stuck-duration": 1800000000000,
+  "sql-mode":"ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION"
 }`
 
 	testCfgTestServerConfigMarshal = `{
@@ -303,6 +305,7 @@ const (
     "level": "none",
     "max-log-size": 64,
     "flush-interval": 2000,
+    "meta-flush-interval": 200,
     "storage": "",
     "use-file-backend": false
   },
@@ -317,7 +320,8 @@ const (
     "integrity-check-level": "none",
     "corruption-handle-level": "warn"
   },
-  "changefeed-error-stuck-duration": 1800000000000
+  "changefeed-error-stuck-duration": 1800000000000,
+  "sql-mode":"ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION"
 }`
 
 	testCfgTestReplicaConfigMarshal2 = `{
@@ -450,6 +454,7 @@ const (
     "level": "none",
     "max-log-size": 64,
     "flush-interval": 2000,
+    "meta-flush-interval": 200,
     "storage": "",
     "use-file-backend": false
   },
@@ -462,6 +467,7 @@ const (
     "integrity-check-level": "none",
     "corruption-handle-level": "warn"
   },
-  "changefeed-error-stuck-duration": 1800000000000
+  "changefeed-error-stuck-duration": 1800000000000,
+  "sql-mode":"ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION"
 }`
 )
diff --git a/pkg/filter/expr_filter_bench_test.go b/pkg/filter/expr_filter_bench_test.go
index 91c3fe903..62263e8e2 100644
--- a/pkg/filter/expr_filter_bench_test.go
+++ b/pkg/filter/expr_filter_bench_test.go
@@ -57,7 +57,7 @@ func BenchmarkSkipDML(b *testing.B) {
 	sessCtx := utils.NewSessionCtx(map[string]string{
 		"time_zone": "",
 	})
-	f, err := newExprFilter("", cfg)
+	f, err := newExprFilter("", cfg, config.GetDefaultReplicaConfig().SQLMode)
 	require.Nil(b, err)
 
 	type innerCase struct {
diff --git a/pkg/filter/expr_filter_test.go b/pkg/filter/expr_filter_test.go
index 372ace442..66fe1f02a 100644
--- a/pkg/filter/expr_filter_test.go
+++ b/pkg/filter/expr_filter_test.go
@@ -324,7 +324,7 @@ func TestShouldSkipDMLBasic(t *testing.T) {
 
 	for _, tc := range testCases {
 		tableInfo := helper.execDDL(tc.ddl)
-		f, err := newExprFilter("", tc.cfg)
+		f, err := newExprFilter("", tc.cfg, config.GetDefaultReplicaConfig().SQLMode)
 		require.Nil(t, err)
 		for _, c := range tc.cases {
 			rowDatums, err := utils.AdjustBinaryProtocolForDatum(sessCtx, c.row, tableInfo.Columns)
@@ -441,7 +441,7 @@ func TestShouldSkipDMLError(t *testing.T) {
 
 	for _, tc := range testCases {
 		tableInfo := helper.execDDL(tc.ddl)
-		f, err := newExprFilter("", tc.cfg)
+		f, err := newExprFilter("", tc.cfg, config.GetDefaultReplicaConfig().SQLMode)
 		require.Nil(t, err)
 		for _, c := range tc.cases {
 			rowDatums, err := utils.AdjustBinaryProtocolForDatum(sessCtx, c.row, tableInfo.Columns)
@@ -634,7 +634,7 @@ func TestShouldSkipDMLTableUpdated(t *testing.T) {
 
 	for _, tc := range testCases {
 		tableInfo := helper.execDDL(tc.ddl)
-		f, err := newExprFilter("", tc.cfg)
+		f, err := newExprFilter("", tc.cfg, config.GetDefaultReplicaConfig().SQLMode)
 		require.Nil(t, err)
 		for _, c := range tc.cases {
 			if c.updateDDl != "" {
@@ -754,7 +754,7 @@ func TestVerify(t *testing.T) {
 			ti := helper.execDDL(ddl)
 			tableInfos = append(tableInfos, ti)
 		}
-		f, err := newExprFilter("", tc.cfg)
+		f, err := newExprFilter("", tc.cfg, config.GetDefaultReplicaConfig().SQLMode)
 		require.Nil(t, err)
 		err = f.verify(tableInfos)
 		require.True(t, errors.ErrorEqual(tc.err, err), "case: %+v", tc, err)
diff --git a/pkg/filter/sql_event_filter_test.go b/pkg/filter/sql_event_filter_test.go
index 1fa03bedc..9e9707d07 100644
--- a/pkg/filter/sql_event_filter_test.go
+++ b/pkg/filter/sql_event_filter_test.go
@@ -171,7 +171,7 @@ func TestShouldSkipDDL(t *testing.T) {
 	}
 
 	for _, tc := range testCases {
-		f, err := newSQLEventFilter(tc.cfg)
+		f, err := newSQLEventFilter(tc.cfg, config.GetDefaultReplicaConfig().SQLMode)
 		require.True(t, errors.ErrorEqual(err, tc.err), "case: %+s", err)
 		for _, c := range tc.cases {
 			ddl := &model.DDLEvent{
@@ -298,7 +298,7 @@ func TestShouldSkipDML(t *testing.T) {
 		tc := tc
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
-			f, err := newSQLEventFilter(tc.cfg)
+			f, err := newSQLEventFilter(tc.cfg, config.GetDefaultReplicaConfig().SQLMode)
 			require.NoError(t, err)
 			for _, c := range tc.cases {
 				event := &model.RowChangedEvent{
diff --git a/pkg/filter/utils_test.go b/pkg/filter/utils_test.go
index 6207ce974..c3441e108 100644
--- a/pkg/filter/utils_test.go
+++ b/pkg/filter/utils_test.go
@@ -14,12 +14,14 @@
 package filter
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/pingcap/log"
 	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb/parser"
 	timodel "github.com/pingcap/tidb/parser/model"
+	"github.com/pingcap/tidb/parser/mysql"
 	tifilter "github.com/pingcap/tidb/util/filter"
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
@@ -174,3 +176,61 @@ func TestDDLToTypeSpecialDDL(t *testing.T) {
 		require.Equal(t, c.evenType, et, "case%v", c.ddl)
 	}
 }
+
+func TestToDDLEventWithSQLMode(t *testing.T) {
+	t.Parallel()
+	cases := []struct {
+		name    string
+		query   string
+		jobTp   timodel.ActionType
+		sqlMode string // sql mode
+		expect  bf.EventType
+		errMsg  string
+	}{
+		{
+			name:    "create table",
+			query:   "create table t1(id int primary key)",
+			jobTp:   timodel.ActionCreateTable,
+			sqlMode: config.GetDefaultReplicaConfig().SQLMode,
+			expect:  bf.CreateTable,
+		},
+		{
+			name:    "drop table",
+			query:   "drop table t1",
+			jobTp:   timodel.ActionDropTable,
+			sqlMode: config.GetDefaultReplicaConfig().SQLMode,
+			expect:  bf.DropTable,
+		},
+		{ // "" in table name or column name are not supported when sqlMode is set to ANSI_QUOTES
+			name:    "create table 2",
+			query:   `create table "t1" ("id" int primary key)`,
+			jobTp:   timodel.ActionCreateTable,
+			sqlMode: config.GetDefaultReplicaConfig().SQLMode,
+			expect:  bf.CreateTable,
+			errMsg:  "ErrConvertDDLToEventTypeFailed",
+		},
+		{ // "" in table name or column name are supported when sqlMode is set to ANSI_QUOTES
+			name:    "create table 3",
+			query:   `create table "t1" ("id" int primary key)`,
+			jobTp:   timodel.ActionCreateTable,
+			sqlMode: fmt.Sprint(config.GetDefaultReplicaConfig().SQLMode + ",ANSI_QUOTES"),
+			expect:  bf.CreateTable,
+		},
+	}
+	for _, c := range cases {
+		innerCase := c
+		t.Run(c.name, func(t *testing.T) {
+			t.Parallel()
+			p := parser.New()
+			mode, err := mysql.GetSQLMode(innerCase.sqlMode)
+			require.NoError(t, err)
+			p.SetSQLMode(mode)
+			tp, err := ddlToEventType(p, innerCase.query, innerCase.jobTp)
+			if innerCase.errMsg != "" {
+				require.Contains(t, err.Error(), innerCase.errMsg, innerCase.name)
+			} else {
+				require.Equal(t, innerCase.expect, tp)
+			}
+		})
+	}
+}
diff --git a/pkg/orchestrator/reactor_state_test.go b/pkg/orchestrator/reactor_state_test.go
index fd0b29f3a..61dc26f4b 100644
--- a/pkg/orchestrator/reactor_state_test.go
+++ b/pkg/orchestrator/reactor_state_test.go
@@ -123,6 +123,7 @@ func TestChangefeedStateUpdate(t *testing.T) {
 						Integrity: config.GetDefaultReplicaConfig().Integrity,
 						ChangefeedErrorStuckDuration: config.
 							GetDefaultReplicaConfig().ChangefeedErrorStuckDuration,
+						SQLMode: config.GetDefaultReplicaConfig().SQLMode,
 					},
 				},
 				Status: &model.ChangeFeedStatus{CheckpointTs: 421980719742451713},
@@ -177,6 +178,7 @@ func TestChangefeedStateUpdate(t *testing.T) {
 						Integrity: config.GetDefaultReplicaConfig().Integrity,
 						ChangefeedErrorStuckDuration: config.
 							GetDefaultReplicaConfig().ChangefeedErrorStuckDuration,
+						SQLMode: config.GetDefaultReplicaConfig().SQLMode,
 					},
 				},
 				Status: &model.ChangeFeedStatus{CheckpointTs: 421980719742451713},
@@ -236,6 +238,7 @@ func TestChangefeedStateUpdate(t *testing.T) {
 						Integrity: config.GetDefaultReplicaConfig().Integrity,
 						ChangefeedErrorStuckDuration: config.
 							GetDefaultReplicaConfig().ChangefeedErrorStuckDuration,
+						SQLMode: config.GetDefaultReplicaConfig().SQLMode,
 					},
 				},
 				Status: &model.ChangeFeedStatus{CheckpointTs: 421980719742451713},
@@ -330,6 +333,7 @@ func TestPatchInfo(t *testing.T) {
 			Scheduler:                    defaultConfig.Scheduler,
 			Integrity:                    defaultConfig.Integrity,
 			ChangefeedErrorStuckDuration: defaultConfig.ChangefeedErrorStuckDuration,
+			SQLMode:                      defaultConfig.SQLMode,
 		},
 	}
 	cfInfo.RmUnusedFields()
@@ -352,6 +356,7 @@ func TestPatchInfo(t *testing.T) {
 			Scheduler:                    defaultConfig.Scheduler,
 			Integrity:                    defaultConfig.Integrity,
 			ChangefeedErrorStuckDuration: defaultConfig.ChangefeedErrorStuckDuration,
+			SQLMode:                      defaultConfig.SQLMode,
 		},
 	}
 	cfInfo.RmUnusedFields()
diff --git a/pkg/sink/codec/simple/encoder_test.go b/pkg/sink/codec/simple/encoder_test.go
new file mode 100644
index 000000000..ea165e8ea
--- /dev/null
+++ b/pkg/sink/codec/simple/encoder_test.go
@@ -0,0 +1,44 @@
+// Copyright 2023 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package simple
+
+import (
+	"testing"
+
+	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/stretchr/testify/require"
+)
+
+func TestEncodeCheckpoint(t *testing.T) {
+	t.Parallel()
+
+	enc := NewBuilder().Build()
+
+	checkpoint := 23
+	m, err := enc.EncodeCheckpointEvent(uint64(checkpoint))
+	require.NoError(t, err)
+
+	dec := NewDecoder()
+	err = dec.AddKeyValue(m.Key, m.Value)
+	require.NoError(t, err)
+
+	messageType, hasNext, err := dec.HasNext()
+	require.NoError(t, err)
+	require.True(t, hasNext)
+	require.Equal(t, model.MessageTypeResolved, messageType)
+
+	ts, err := dec.NextResolvedEvent()
+	require.NoError(t, err)
+	require.Equal(t, uint64(checkpoint), ts)
+}
diff --git a/tests/integration_tests/api_v2/model.go b/tests/integration_tests/api_v2/model.go
index 7ef56dffc..a2d1fa72f 100644
--- a/tests/integration_tests/api_v2/model.go
+++ b/tests/integration_tests/api_v2/model.go
@@ -272,11 +272,12 @@ type ColumnSelector struct {
 // ConsistentConfig represents replication consistency config for a changefeed
 // This is a duplicate of config.ConsistentConfig
 type ConsistentConfig struct {
-	Level             string `json:"level"`
-	MaxLogSize        int64  `json:"max_log_size"`
-	FlushIntervalInMs int64  `json:"flush_interval"`
-	Storage           string `json:"storage"`
-	UseFileBackend    bool   `json:"use_file_backend"`
+	Level                 string `json:"level"`
+	MaxLogSize            int64  `json:"max_log_size"`
+	FlushIntervalInMs     int64  `json:"flush_interval"`
+	MetaFlushIntervalInMs int64  `json:"meta_flush_interval"`
+	Storage               string `json:"storage"`
+	UseFileBackend        bool   `json:"use_file_backend"`
 }
 
 // ChangefeedSchedulerConfig is per changefeed scheduler settings.
diff --git a/tests/integration_tests/changefeed_error/run.sh b/tests/integration_tests/changefeed_error/run.sh
index 5065d1478..f48b72fbf 100755
--- a/tests/integration_tests/changefeed_error/run.sh
+++ b/tests/integration_tests/changefeed_error/run.sh
@@ -108,7 +108,7 @@ 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"