Repository: pingcap/tiflow
Pull request #10585 (2024-02-04 09:22:19)

kv (ticdc): Improve the codebase of kvClient.
<!--
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: ref #10584 

### What is changed and how it works?
- Move stream related filed into `eventFeedStream` struct to make the code more clear. 
- Reduce a lock usage by make sure the `storeStreamsCache` can only be accessed in one goroutine. 

### Check List <!--REMOVE the items that are not applicable-->

#### Tests <!-- At least one of them must be included. -->

 - Manual test (add detailed scripts or steps below)


#### 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 [asddongmen](https://github.com/asddongmen). 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":["asddongmen"]} -->
ti-chi-bot[bot]: @asddongmen: 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
--- | --- | --- | --- | ---
jenkins-ticdc/verify | 6f0170bdf291d22e78a624f59703e7e3659a3187 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/686/display/redirect) | true | `/test verify`

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

<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 -->
ti-chi-bot[bot]: @asddongmen: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
jenkins-ticdc/verify | 6f0170bdf291d22e78a624f59703e7e3659a3187 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/686/display/redirect) | true | `/test verify`
pull-cdc-integration-pulsar-test | 6f0170bdf291d22e78a624f59703e7e3659a3187 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_pulsar_test/395/display/redirect) | true | `/test cdc-integration-pulsar-test`

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

<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 -->
ti-chi-bot[bot]: @asddongmen: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
jenkins-ticdc/verify | 6f0170bdf291d22e78a624f59703e7e3659a3187 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/686/display/redirect) | true | `/test verify`
pull-cdc-integration-pulsar-test | 6f0170bdf291d22e78a624f59703e7e3659a3187 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_pulsar_test/395/display/redirect) | true | `/test cdc-integration-pulsar-test`
pull-cdc-integration-kafka-test | 6f0170bdf291d22e78a624f59703e7e3659a3187 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/681/display/redirect) | true | `/test cdc-integration-kafka-test`

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

<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 -->
asddongmen: It is no need to merge this PR into master branch.
ti-chi-bot[bot]: 
[FORMAT CHECKER NOTIFICATION]

**Notice**: To remove the `do-not-merge/needs-linked-issue` label, please provide the linked issue number on one line in the PR body, for example: `Issue Number: close #123` or `Issue Number: ref #456`.

<sub>:open_book: For more info, you can check the ["Contribute Code"](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue) section in the development guide.</sub>


<!--Checker Notification Identifier-->

ti-chi-bot[bot]: @asddongmen: The following tests **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-cdc-integration-pulsar-test | 6f0170bdf291d22e78a624f59703e7e3659a3187 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_pulsar_test/395/display/redirect) | true | `/test cdc-integration-pulsar-test`
pull-cdc-integration-kafka-test | 6f0170bdf291d22e78a624f59703e7e3659a3187 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/681/display/redirect) | true | `/test cdc-integration-kafka-test`
jenkins-ticdc/verify | 6f0170bdf291d22e78a624f59703e7e3659a3187 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/721/display/redirect) | true | `/test verify`

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

<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/cdc/kv/client.go b/cdc/kv/client.go
index f15ac1f25..92f0a6ef1 100644
--- a/cdc/kv/client.go
+++ b/cdc/kv/client.go
@@ -19,7 +19,6 @@ import (
 	"io"
 	"math/rand"
 	"strconv"
-	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -140,6 +139,8 @@ func newRegionErrorInfo(info singleRegionInfo, err error) regionErrorInfo {
 type eventFeedStream struct {
 	client cdcpb.ChangeData_EventFeedClient
 	conn   *sharedConn
+	// regions is used to store the state of the regions that are being processed by the stream.
+	regions *syncRegionFeedStateMap
 	// addr is the address of the TiKV store
 	addr string
 	// storeID is the ID of the TiKV store
@@ -147,7 +148,15 @@ type eventFeedStream struct {
 	// id is the stream ID, which is used to identify the stream.
 	id uint64
 	// cancel is used to cancel the gRPC stream
-	cancel context.CancelFunc
+	cancel     context.CancelFunc
+	isCanceled atomic.Bool
+	createTime time.Time
+}
+
+func (s *eventFeedStream) close() {
+	if s.isCanceled.CompareAndSwap(false, true) {
+		s.cancel()
+	}
 }
 
 // CDCKVClient is an interface to receives kv changed logs from TiKV
@@ -246,59 +255,58 @@ func NewCDCClient(
 
 func (c *CDCClient) newStream(
 	ctx context.Context,
-	cancel context.CancelFunc,
 	addr string,
 	storeID uint64,
-) (stream *eventFeedStream, newStreamErr error) {
-	streamFunc := func() (err error) {
-		var conn *sharedConn
-		defer func() {
-			if err != nil && conn != nil {
-				c.grpcPool.ReleaseConn(conn, addr)
-			}
-		}()
-		conn, err = c.grpcPool.GetConn(addr)
-		if err != nil {
-			return errors.Trace(err)
-		}
-		err = version.CheckStoreVersion(ctx, c.pd, storeID)
-		if err != nil {
-			return errors.Trace(err)
-		}
-		client := cdcpb.NewChangeDataClient(conn.ClientConn)
-		var streamClient cdcpb.ChangeData_EventFeedClient
-		streamClient, err = client.EventFeed(ctx)
-		if err != nil {
-			return cerror.WrapError(cerror.ErrTiKVEventFeed, err)
-		}
-		stream = &eventFeedStream{
-			client:  streamClient,
-			conn:    conn,
-			addr:    addr,
-			storeID: storeID,
-			id:      allocateStreamID(),
-			cancel:  cancel,
+) (stream *eventFeedStream, err error) {
+	var conn *sharedConn
+	defer func() {
+		if err != nil && conn != nil {
+			c.grpcPool.ReleaseConn(conn, addr)
 		}
-		log.Info("created stream to store",
-			zap.String("namespace", c.changefeed.Namespace),
-			zap.String("changefeed", c.changefeed.ID),
-			zap.Int64("tableID", c.tableID),
-			zap.String("tableName", c.tableName),
-			zap.String("store", addr),
-			zap.Uint64("storeID", storeID),
-			zap.Uint64("streamID", stream.id))
-		return nil
+	}()
+
+	conn, err = c.grpcPool.GetConn(addr)
+	if err != nil {
+		return nil, errors.Trace(err)
 	}
-	if c.config.Debug.EnableKVConnectBackOff {
-		newStreamErr = retry.Do(ctx, streamFunc,
-			retry.WithBackoffBaseDelay(100),
-			retry.WithMaxTries(2),
-			retry.WithIsRetryableErr(cerror.IsRetryableError),
-		)
-		return
+
+	ctx, cancel := context.WithCancel(ctx)
+	err = version.CheckStoreVersion(ctx, c.pd, storeID)
+	if err != nil {
+		cancel()
+		return nil, errors.Trace(err)
+	}
+
+	client := cdcpb.NewChangeDataClient(conn.ClientConn)
+	var streamClient cdcpb.ChangeData_EventFeedClient
+
+	streamClient, err = client.EventFeed(ctx)
+	if err != nil {
+		cancel()
+		return nil, cerror.WrapError(cerror.ErrTiKVEventFeed, err)
+	}
+
+	stream = &eventFeedStream{
+		client:     streamClient,
+		conn:       conn,
+		regions:    newSyncRegionFeedStateMap(),
+		addr:       addr,
+		storeID:    storeID,
+		id:         allocateStreamID(),
+		cancel:     cancel,
+		isCanceled: atomic.Bool{},
+		createTime: time.Now(),
 	}
-	newStreamErr = streamFunc()
-	return
+
+	log.Info("created stream to store",
+		zap.String("namespace", c.changefeed.Namespace),
+		zap.String("changefeed", c.changefeed.ID),
+		zap.Int64("tableID", c.tableID),
+		zap.String("tableName", c.tableName),
+		zap.String("store", addr),
+		zap.Uint64("storeID", storeID),
+		zap.Uint64("streamID", stream.id))
+	return stream, nil
 }
 
 // EventFeed divides a EventFeed request on range boundaries and establishes
@@ -401,14 +409,9 @@ type eventFeedSession struct {
 	rangeChSizeGauge   prometheus.Gauge
 
 	// storeStreamsCache is used to cache the established gRPC streams to TiKV stores.
-	storeStreamsCache struct {
-		sync.RWMutex
-		m map[string]*eventFeedStream
-		// lastAlterTime is used to record last time a stream is created/deleted to/from the cache.
-		// It is used to avoid creation/deleting streams too frequently, which may cause
-		// huge overhead of incremental region scanning in TiKV.
-		lastAlterTime map[string]time.Time
-	}
+	// Note: The cache is not thread-safe, so it should be accessed in the same goroutine.
+	// For now, it is only accessed in the `requestRegionToStore` goroutine.
+	storeStreamsCache map[string]*eventFeedStream
 
 	// use sync.Pool to store resolved ts event only, because resolved ts event
 	// has the same size and generate cycle.
@@ -431,13 +434,13 @@ func newEventFeedSession(
 	rangeLock := regionlock.NewRegionRangeLock(
 		id, totalSpan.StartKey, totalSpan.EndKey, startTs,
 		client.changefeed.Namespace+"."+client.changefeed.ID)
-	res := &eventFeedSession{
-		client:     client,
-		startTs:    startTs,
-		changefeed: client.changefeed,
-		tableID:    client.tableID,
-		tableName:  client.tableName,
-
+	return &eventFeedSession{
+		client:             client,
+		startTs:            startTs,
+		changefeed:         client.changefeed,
+		tableID:            client.tableID,
+		tableName:          client.tableName,
+		storeStreamsCache:  make(map[string]*eventFeedStream),
 		totalSpan:          totalSpan,
 		eventCh:            eventCh,
 		rangeLock:          rangeLock,
@@ -457,9 +460,6 @@ func newEventFeedSession(
 			},
 		},
 	}
-	res.storeStreamsCache.m = make(map[string]*eventFeedStream)
-	res.storeStreamsCache.lastAlterTime = make(map[string]time.Time)
-	return res
 }
 
 func (s *eventFeedSession) eventFeed(ctx context.Context) error {
@@ -638,11 +638,6 @@ func (s *eventFeedSession) requestRegionToStore(
 	ctx context.Context,
 	g *errgroup.Group,
 ) error {
-	// Stores pending regions info for each stream. After sending a new request, the region info wil be put to the map,
-	// and it will be loaded by the receiver thread when it receives the first response from that region. We need this
-	// to pass the region info to the receiver since the region info cannot be inferred from the response from TiKV.
-	storePendingRegions := make(map[uint64]*syncRegionFeedStateMap)
-
 	header := &cdcpb.Header{
 		ClusterId:    s.client.clusterID,
 		TicdcVersion: version.ReleaseSemver(),
@@ -679,18 +674,18 @@ func (s *eventFeedSession) requestRegionToStore(
 		// each TiKV store has an independent pendingRegions.
 		storeAddr := rpcCtx.Addr
 		storeID := rpcCtx.Peer.GetStoreId()
+
 		var (
 			stream *eventFeedStream
 			err    error
 		)
-		stream, ok := s.getStream(storeAddr)
-		if !ok {
-			// when a new stream is established, always create a new pending
-			// regions map, the old map will be used in old `receiveFromStream`
-			// and won't be deleted until that goroutine exits.
-			pendingRegions := newSyncRegionFeedStateMap()
-			streamCtx, streamCancel := context.WithCancel(ctx)
-			stream, err = s.client.newStream(streamCtx, streamCancel, storeAddr, storeID)
+		stream, ok := s.storeStreamsCache[storeAddr]
+		if !ok || stream.isCanceled.Load() {
+			if ok {
+				// If the stream is canceled, we need to delete it from the cache and close it.
+				s.deleteStream(stream)
+			}
+			stream, err = s.client.newStream(ctx, storeAddr, storeID)
 			if err != nil {
 				// get stream failed, maybe the store is down permanently, we should try to relocate the active store
 				log.Warn("get grpc stream client failed",
@@ -716,8 +711,8 @@ func (s *eventFeedSession) requestRegionToStore(
 				continue
 			}
 
-			storePendingRegions[stream.id] = pendingRegions
 			s.addStream(stream)
+
 			log.Info("creating new stream to store to send request",
 				zap.String("namespace", s.changefeed.Namespace),
 				zap.String("changefeed", s.changefeed.ID),
@@ -728,29 +723,13 @@ func (s *eventFeedSession) requestRegionToStore(
 				zap.Uint64("storeID", storeID),
 				zap.String("store", storeAddr),
 				zap.Uint64("streamID", stream.id))
-
 			g.Go(func() error {
-				return s.receiveFromStream(ctx, stream, pendingRegions)
+				return s.receiveFromStream(ctx, stream)
 			})
 		}
 
-		pendingRegions, ok := storePendingRegions[stream.id]
-		if !ok {
-			// Should never happen
-			log.Error("pending regions is not found for store",
-				zap.String("namespace", s.changefeed.Namespace),
-				zap.String("changefeed", s.changefeed.ID),
-				zap.Int64("tableID", s.tableID),
-				zap.String("tableName", s.tableName),
-				zap.Uint64("storeID", storeID),
-				zap.String("store", storeAddr),
-				zap.Uint64("regionID", sri.verID.GetID()),
-				zap.Uint64("requestID", requestID))
-			return cerror.ErrUnexpected.FastGenByArgs("pending regions is not found for store")
-		}
-
 		state := newRegionFeedState(sri, requestID)
-		pendingRegions.setByRequestID(requestID, state)
+		stream.regions.setByRequestID(requestID, state)
 		s.client.logRegionDetails("start new request",
 			zap.String("namespace", s.changefeed.Namespace),
 			zap.String("changefeed", s.changefeed.ID),
@@ -763,8 +742,6 @@ func (s *eventFeedSession) requestRegionToStore(
 			zap.Stringer("span", &sri.span))
 
 		err = stream.client.Send(req)
-		// If Send returns an error, the stream.client.Recv (In s.receiveFromStream)
-		// would also receive an error.
 		if err != nil {
 			log.Warn("send request to stream failed",
 				zap.String("namespace", s.changefeed.Namespace),
@@ -788,18 +765,13 @@ func (s *eventFeedSession) requestRegionToStore(
 					zap.String("store", storeAddr),
 					zap.Error(err))
 			}
-			// Delete the stream from the cache so that when next time the store is accessed,
-			// the stream can be re-established.
-			s.deleteStream(stream)
-			// Delete `pendingRegions` from `storePendingRegions` so that the next time a region of this store is
-			// requested, it will create a new one. So if the `receiveFromStream` goroutine tries to stop all
-			// pending regions, the new pending regions that are requested after reconnecting won't be stopped
-			// incorrectly.
-			delete(storePendingRegions, stream.id)
 
+			// Delete the stream from the cache so that when next time a region of
+			// this store is requested, a new stream to this store will be created.
+			s.deleteStream(stream)
 			// Remove the region from pendingRegions. If it's already removed, it should be already retried by
 			// `receiveFromStream`, so no need to retry here.
-			_, ok := pendingRegions.takeByRequestID(requestID)
+			_, ok := stream.regions.takeByRequestID(requestID)
 			if !ok {
 				continue
 			}
@@ -1060,7 +1032,6 @@ func (s *eventFeedSession) getRPCContextForRegion(ctx context.Context, id tikv.R
 func (s *eventFeedSession) receiveFromStream(
 	parentCtx context.Context,
 	stream *eventFeedStream,
-	pendingRegions *syncRegionFeedStateMap,
 ) error {
 	var tsStat *tableStoreStat
 	s.client.tableStoreStats.Lock()
@@ -1086,7 +1057,7 @@ func (s *eventFeedSession) receiveFromStream(
 
 		failpoint.Inject("kvClientStreamCloseDelay", nil)
 
-		remainingRegions := pendingRegions.takeAll()
+		remainingRegions := stream.regions.takeAll()
 		for _, state := range remainingRegions {
 			s.client.logRegionDetails("region canceled",
 				zap.String("namespace", s.changefeed.Namespace),
@@ -1109,11 +1080,7 @@ func (s *eventFeedSession) receiveFromStream(
 
 	// always create a new region worker, because `receiveFromStream` is ensured
 	// to call exactly once from outer code logic
-	worker := newRegionWorker(
-		parentCtx,
-		stream,
-		s,
-		pendingRegions)
+	worker := newRegionWorker(parentCtx, stream, s)
 	defer worker.evictAllRegions()
 
 	ctx, cancel := context.WithCancel(parentCtx)
@@ -1247,7 +1214,7 @@ func (s *eventFeedSession) receiveFromStream(
 				maxCommitTs = commitTs
 			}
 
-			err = s.sendRegionChangeEvents(ctx, cevent.Events, worker, pendingRegions)
+			err = s.sendRegionChangeEvents(ctx, cevent.Events, worker, stream.regions)
 			if err != nil {
 				return err
 			}
@@ -1423,34 +1390,41 @@ func (s *eventFeedSession) sendResolvedTs(
 	return nil
 }
 
+// addStream adds a stream to the session.streams.
+// Note: It must be called with deleteStream in a same goroutine.
 func (s *eventFeedSession) addStream(stream *eventFeedStream) {
-	s.storeStreamsCache.Lock()
-	defer s.storeStreamsCache.Unlock()
-	if lastTime, ok := s.storeStreamsCache.lastAlterTime[stream.addr]; ok {
-		if time.Since(lastTime) < streamAlterInterval {
-			log.Warn(
-				"add a stream of a same store too frequently, wait a while and try again",
+	oldStream, ok := s.storeStreamsCache[stream.addr]
+	if ok {
+		failpoint.Inject("kvClientAddDuplicatedStream", func() {
+			log.Error("A stream to a same store already exists, it shouldn't happen, please report a bug",
 				zap.String("namespace", s.changefeed.Namespace),
 				zap.String("changefeed", s.changefeed.ID),
 				zap.Int64("tableID", s.tableID),
 				zap.String("tableName", s.tableName),
-				zap.Uint64("storeID", stream.storeID),
-				zap.Duration("sinceLastTime", time.Since(lastTime)))
-			time.Sleep(streamAlterInterval - time.Since(lastTime))
-		}
+				zap.Uint64("oldStreamID", oldStream.id),
+				zap.Uint64("newStreamID", stream.id))
+		})
+		// There is no need to return an error here because even if it happens,
+		// it does not harm the data correctness, but may only cause some lag spikes.
+		// Log it to help us improve the code.
+		log.Error("A stream to a same store already exists, it shouldn't happen, please report a bug",
+			zap.String("namespace", s.changefeed.Namespace),
+			zap.String("changefeed", s.changefeed.ID),
+			zap.Int64("tableID", s.tableID),
+			zap.String("tableName", s.tableName),
+			zap.Uint64("oldStreamID", oldStream.id),
+			zap.Uint64("newStreamID", stream.id))
 	}
-	s.storeStreamsCache.m[stream.addr] = stream
-	s.storeStreamsCache.lastAlterTime[stream.addr] = time.Now()
+	s.storeStreamsCache[stream.addr] = stream
 }
 
 // deleteStream deletes a stream from the session.streams.
-// If the stream is not found, it will be ignored.
+// If the stream is not found, it takes no effect.
 func (s *eventFeedSession) deleteStream(streamToDelete *eventFeedStream) {
-	s.storeStreamsCache.Lock()
-	defer s.storeStreamsCache.Unlock()
-
-	streamInMap, ok := s.storeStreamsCache.m[streamToDelete.addr]
+	streamInMap, ok := s.storeStreamsCache[streamToDelete.addr]
 	if !ok {
+		// This should not happen, but it will be no harm if it happens.
+		// Log a warning message to help us diagnose the problem.
 		log.Warn("delete stream failed, stream not found, ignore it",
 			zap.String("namespace", s.changefeed.Namespace),
 			zap.String("changefeed", s.changefeed.ID),
@@ -1460,8 +1434,9 @@ func (s *eventFeedSession) deleteStream(streamToDelete *eventFeedStream) {
 			zap.Uint64("streamIDInMap", streamInMap.id))
 		return
 	}
-
 	if streamInMap.id != streamToDelete.id {
+		// This should not happen, but it will be no harm if it happens.
+		// Log a warning message to help us diagnose the problem.
 		log.Warn("delete stream failed, stream id mismatch, ignore it",
 			zap.String("namespace", s.changefeed.Namespace),
 			zap.String("changefeed", s.changefeed.ID),
@@ -1472,25 +1447,20 @@ func (s *eventFeedSession) deleteStream(streamToDelete *eventFeedStream) {
 		return
 	}
 
-	if lastTime, ok := s.storeStreamsCache.lastAlterTime[streamToDelete.addr]; ok {
-		if time.Since(lastTime) < streamAlterInterval {
-			log.Warn(
-				"delete a stream of a same store too frequently, wait a while and try again",
-				zap.String("namespace", s.changefeed.Namespace),
-				zap.String("changefeed", s.changefeed.ID),
-				zap.Int64("tableID", s.tableID),
-				zap.String("tableName", s.tableName),
-				zap.Uint64("streamID", streamToDelete.id),
-				zap.Duration("sinceLastTime", time.Since(lastTime)))
-			time.Sleep(streamAlterInterval - time.Since(lastTime))
-		}
+	if time.Since(streamToDelete.createTime) < streamAlterInterval {
+		log.Warn("It's too soon to delete a stream, wait for a while",
+			zap.String("namespace", s.changefeed.Namespace),
+			zap.String("changefeed", s.changefeed.ID),
+			zap.Int64("tableID", s.tableID),
+			zap.String("tableName", s.tableName),
+			zap.Uint64("streamID", streamToDelete.id),
+			zap.Duration("sinceCreateDuration", time.Since(streamToDelete.createTime)))
+		time.Sleep(streamAlterInterval - time.Since(streamToDelete.createTime))
 	}
 
 	s.client.grpcPool.ReleaseConn(streamToDelete.conn, streamToDelete.addr)
-	delete(s.storeStreamsCache.m, streamToDelete.addr)
-	streamToDelete.cancel()
-	s.storeStreamsCache.lastAlterTime[streamToDelete.addr] = time.Now()
-
+	streamToDelete.close()
+	delete(s.storeStreamsCache, streamToDelete.addr)
 	log.Info("A stream to store has been removed",
 		zap.String("namespace", s.changefeed.Namespace),
 		zap.String("changefeed", s.changefeed.ID),
@@ -1498,14 +1468,8 @@ func (s *eventFeedSession) deleteStream(streamToDelete *eventFeedStream) {
 		zap.String("tableName", s.tableName),
 		zap.String("store", streamToDelete.addr),
 		zap.Uint64("storeID", streamToDelete.storeID),
-		zap.Uint64("streamID", streamToDelete.id))
-}
-
-func (s *eventFeedSession) getStream(storeAddr string) (stream *eventFeedStream, ok bool) {
-	s.storeStreamsCache.RLock()
-	defer s.storeStreamsCache.RUnlock()
-	stream, ok = s.storeStreamsCache.m[storeAddr]
-	return
+		zap.Uint64("streamID", streamToDelete.id),
+		zap.Duration("sinceCreateDuration", time.Since(streamToDelete.createTime)))
 }
 
 func (s *eventFeedSession) logSlowRegions(ctx context.Context) error {
@@ -1519,7 +1483,7 @@ func (s *eventFeedSession) logSlowRegions(ctx context.Context) error {
 		}
 
 		attr := s.rangeLock.CollectLockedRangeAttrs(nil)
-		ckptTime := oracle.GetTimeFromTS(attr.SlowestRegion.CheckpointTs)
+		ckptTime := oracle.GetTimeFromTS(attr.SlowestRegion.ResolvedTs)
 		currTime := s.client.pdClock.CurrentTime()
 		log.Info("event feed starts to check locked regions",
 			zap.String("namespace", s.changefeed.Namespace),
@@ -1555,16 +1519,12 @@ func (s *eventFeedSession) logSlowRegions(ctx context.Context) error {
 				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, ", ")))
+				zap.Any("holes", attr.Holes))
 		}
 	}
 }
diff --git a/cdc/kv/region_state.go b/cdc/kv/region_state.go
index 1263eaee9..321e22c5e 100644
--- a/cdc/kv/region_state.go
+++ b/cdc/kv/region_state.go
@@ -53,7 +53,7 @@ func newSingleRegionInfo(
 }
 
 func (s singleRegionInfo) resolvedTs() uint64 {
-	return s.lockedRange.CheckpointTs.Load()
+	return s.lockedRange.ResolvedTs.Load()
 }
 
 type regionFeedState struct {
@@ -134,18 +134,18 @@ func (s *regionFeedState) getRegionID() uint64 {
 }
 
 func (s *regionFeedState) getLastResolvedTs() uint64 {
-	return s.sri.lockedRange.CheckpointTs.Load()
+	return s.sri.lockedRange.ResolvedTs.Load()
 }
 
 // updateResolvedTs update the resolved ts of the current region feed
 func (s *regionFeedState) updateResolvedTs(resolvedTs uint64) {
 	state := s.sri.lockedRange
 	for {
-		last := state.CheckpointTs.Load()
+		last := state.ResolvedTs.Load()
 		if last > resolvedTs {
 			return
 		}
-		if state.CheckpointTs.CompareAndSwap(last, resolvedTs) {
+		if state.ResolvedTs.CompareAndSwap(last, resolvedTs) {
 			break
 		}
 	}
diff --git a/cdc/kv/region_worker.go b/cdc/kv/region_worker.go
index 5eee15899..8e3d7267b 100644
--- a/cdc/kv/region_worker.go
+++ b/cdc/kv/region_worker.go
@@ -115,9 +115,8 @@ type regionWorker struct {
 
 	metrics *regionWorkerMetrics
 
-	// how many pending input events
-	inputPending   int32
-	pendingRegions *syncRegionFeedStateMap
+	// how many pending input events from the input channel
+	inputPendingEvents int32
 }
 
 func newRegionWorkerMetrics(changefeedID model.ChangeFeedID, tableID string, storeAddr string) *regionWorkerMetrics {
@@ -158,22 +157,20 @@ func newRegionWorker(
 	ctx context.Context,
 	stream *eventFeedStream,
 	s *eventFeedSession,
-	pendingRegions *syncRegionFeedStateMap,
 ) *regionWorker {
 	return &regionWorker{
-		parentCtx:      ctx,
-		session:        s,
-		inputCh:        make(chan []*regionStatefulEvent, regionWorkerInputChanSize),
-		outputCh:       s.eventCh,
-		stream:         stream,
-		errorCh:        make(chan error, 1),
-		statesManager:  newRegionStateManager(-1),
-		rtsManager:     newRegionTsManager(),
-		rtsUpdateCh:    make(chan *rtsUpdateEvent, 1024),
-		concurrency:    int(s.client.config.KVClient.WorkerConcurrent),
-		metrics:        newRegionWorkerMetrics(s.changefeed, strconv.FormatInt(s.tableID, 10), stream.addr),
-		inputPending:   0,
-		pendingRegions: pendingRegions,
+		parentCtx:          ctx,
+		session:            s,
+		inputCh:            make(chan []*regionStatefulEvent, regionWorkerInputChanSize),
+		outputCh:           s.eventCh,
+		stream:             stream,
+		errorCh:            make(chan error, 1),
+		statesManager:      newRegionStateManager(-1),
+		rtsManager:         newRegionTsManager(),
+		rtsUpdateCh:        make(chan *rtsUpdateEvent, 1024),
+		concurrency:        int(s.client.config.KVClient.WorkerConcurrent),
+		metrics:            newRegionWorkerMetrics(s.changefeed, strconv.FormatInt(s.tableID, 10), stream.addr),
+		inputPendingEvents: 0,
 	}
 }
 
@@ -209,9 +206,9 @@ func (w *regionWorker) checkShouldExit() error {
 	empty := w.checkRegionStateEmpty()
 	// If there is no region maintained by this region worker, exit it and
 	// cancel the gRPC stream.
-	if empty && w.pendingRegions.len() == 0 {
+	if empty && w.stream.regions.len() == 0 {
 		log.Info("A single region error happens before, "+
-			"and there is no region maintained by this region worker, "+
+			"and there is no region maintained by the stream, "+
 			"exit it and cancel the gRPC stream",
 			zap.String("namespace", w.session.client.changefeed.Namespace),
 			zap.String("changefeed", w.session.client.changefeed.ID),
@@ -521,13 +518,13 @@ func (w *regionWorker) eventHandler(ctx context.Context, enableTableMonitor bool
 			regionEventsBatchSize.Observe(float64(len(events)))
 
 			start := time.Now()
-			inputPending := atomic.LoadInt32(&w.inputPending)
+			inputPending := atomic.LoadInt32(&w.inputPendingEvents)
 			if highWatermarkMet {
 				highWatermarkMet = int(inputPending) >= regionWorkerLowWatermark
 			} else {
 				highWatermarkMet = int(inputPending) >= regionWorkerHighWatermark
 			}
-			atomic.AddInt32(&w.inputPending, -int32(len(events)))
+			atomic.AddInt32(&w.inputPendingEvents, -int32(len(events)))
 
 			if highWatermarkMet {
 				// All events in one batch can be hashed into one handle slot.
@@ -631,7 +628,7 @@ func (w *regionWorker) cancelStream(delay time.Duration) {
 	// This will make the receiveFromStream goroutine exit and the stream can
 	// be re-established by the caller.
 	// Note: use context cancel is the only way to terminate a gRPC stream.
-	w.stream.cancel()
+	w.stream.close()
 	// Failover in stream.Recv has 0-100ms delay, the onRegionFail
 	// should be called after stream has been deleted. Add a delay here
 	// to avoid too frequent region rebuilt.
@@ -905,10 +902,10 @@ func (w *regionWorker) evictAllRegions() {
 // sendEvents puts events into inputCh and updates some internal states.
 // Callers must ensure that all items in events can be hashed into one handle slot.
 func (w *regionWorker) sendEvents(ctx context.Context, events []*regionStatefulEvent) error {
-	atomic.AddInt32(&w.inputPending, int32(len(events)))
+	atomic.AddInt32(&w.inputPendingEvents, int32(len(events)))
 	select {
 	case <-ctx.Done():
-		atomic.AddInt32(&w.inputPending, -int32(len(events)))
+		atomic.AddInt32(&w.inputPendingEvents, -int32(len(events)))
 		return ctx.Err()
 	case w.inputCh <- events:
 		return nil
diff --git a/cdc/kv/regionlock/region_range_lock.go b/cdc/kv/regionlock/region_range_lock.go
index 8d4a91991..9502e58f4 100644
--- a/cdc/kv/regionlock/region_range_lock.go
+++ b/cdc/kv/regionlock/region_range_lock.go
@@ -31,84 +31,213 @@ import (
 )
 
 type rangeTsEntry struct {
-	// Only startKey is necessary. End key can be inferred by the next item since the map always keeps a continuous
-	// range.
+	// Only startKey is necessary. End key can be inferred by the next item,
+	// since the map always keeps a continuous range.
 	startKey []byte
 	ts       uint64
+
+	// rangeTsEntry is used in rangeTsMap. rangeTsMap.set will associate a timestamp to a given key range,
+	// and rangeTsMap.unset will revoke the relationship. `isSet` indicates whether a key range is
+	// generated from a set or unset operation.
+	isSet bool
 }
 
-func rangeTsEntryWithKey(key []byte) *rangeTsEntry {
-	return &rangeTsEntry{
-		startKey: key,
-	}
+func rangeTsEntryWithKey(key []byte) rangeTsEntry {
+	return rangeTsEntry{startKey: key}
 }
 
-func rangeTsEntryLess(a, b *rangeTsEntry) bool {
+func rangeTsEntryLess(a, b rangeTsEntry) bool {
 	return bytes.Compare(a.startKey, b.startKey) < 0
 }
 
 // rangeTsMap represents a map from key range to a timestamp. It supports
 // range set and calculating min value among a specified range.
 type rangeTsMap struct {
-	m *btree.BTreeG[*rangeTsEntry]
+	m     *btree.BTreeG[rangeTsEntry]
+	start []byte
+	end   []byte
 }
 
 // newRangeTsMap creates a RangeTsMap.
 func newRangeTsMap(startKey, endKey []byte, startTs uint64) *rangeTsMap {
 	m := &rangeTsMap{
-		m: btree.NewG(16, rangeTsEntryLess),
+		m:     btree.NewG(16, rangeTsEntryLess),
+		start: startKey,
+		end:   endKey,
 	}
-	m.Set(startKey, endKey, startTs)
+	m.set(startKey, endKey, startTs)
 	return m
 }
 
-// Set sets the corresponding ts of the given range to the specified value.
-func (m *rangeTsMap) Set(startKey, endKey []byte, ts uint64) {
-	if _, ok := m.m.Get(rangeTsEntryWithKey(endKey)); !ok {
-		// To calculate the minimal ts, the default value is math.MaxUint64
-		tailTs := uint64(math.MaxUint64)
-		m.m.DescendLessOrEqual(rangeTsEntryWithKey(endKey), func(i *rangeTsEntry) bool {
-			tailTs = i.ts
+func (m *rangeTsMap) clone() (res *rangeTsMap) {
+	res = &rangeTsMap{
+		m:     btree.NewG(16, rangeTsEntryLess),
+		start: m.start,
+		end:   m.end,
+	}
+	m.m.Ascend(func(i rangeTsEntry) bool {
+		res.m.ReplaceOrInsert(i)
+		return true
+	})
+	return
+}
+
+// set the timestamp of range [startKey, endKey) to ts.
+// RegionRangeLock uses rangeTsMap to store unsubscribed regions.
+// So `set` should be called after a region gets unlocked from RegionRangeLock.
+// Note: It leaves the timestamp of range [endKey, +∞) intact.
+func (m *rangeTsMap) set(startKey, endKey []byte, ts uint64) {
+	startEntry := rangeTsEntryWithKey(startKey)
+	endEntry := rangeTsEntryWithKey(endKey)
+
+	// Check and update the range overlapped with endKey is it exists.
+	if !bytes.Equal(m.end, endKey) && !m.m.Has(endEntry) {
+		var found bool
+		var endKeyOverlapped rangeTsEntry
+		m.m.DescendLessOrEqual(endEntry, func(i rangeTsEntry) bool {
+			found = true
+			endKeyOverlapped.ts = i.ts
+			endKeyOverlapped.isSet = i.isSet
 			return false
 		})
-		m.m.ReplaceOrInsert(&rangeTsEntry{
-			startKey: endKey,
-			ts:       tailTs,
-		})
+		if found {
+			if endKeyOverlapped.isSet {
+				log.Panic("rangeTsMap double set")
+			}
+			endKeyOverlapped.startKey = endKey
+			m.m.ReplaceOrInsert(endKeyOverlapped)
+		}
 	}
 
-	entriesToDelete := make([]*rangeTsEntry, 0)
-	m.m.AscendRange(rangeTsEntryWithKey(startKey), rangeTsEntryWithKey(endKey),
-		func(i *rangeTsEntry) bool {
-			entriesToDelete = append(entriesToDelete, i)
-			return true
+	// Check and update the range overlapped with startKey if it exists.
+	if !m.m.Has(startEntry) {
+		var found bool
+		var startKeyOverlapped rangeTsEntry
+		m.m.DescendLessOrEqual(startEntry, func(i rangeTsEntry) bool {
+			found = true
+			startKeyOverlapped.isSet = i.isSet
+			return false
 		})
+		if found && startKeyOverlapped.isSet {
+			log.Panic("rangeTsMap double set")
+		}
+	}
 
+	// Check and delete all covered ranges.
+	entriesToDelete := make([]rangeTsEntry, 0)
+	m.m.AscendRange(startEntry, endEntry, func(i rangeTsEntry) bool {
+		if i.isSet {
+			log.Panic("rangeTsMap double set")
+		}
+		entriesToDelete = append(entriesToDelete, i)
+		return true
+	})
 	for _, e := range entriesToDelete {
 		m.m.Delete(e)
 	}
 
-	m.m.ReplaceOrInsert(&rangeTsEntry{
-		startKey: startKey,
-		ts:       ts,
+	m.m.ReplaceOrInsert(rangeTsEntry{startKey: startKey, ts: ts, isSet: true})
+}
+
+// RegionRangeLock uses rangeTsMap to store unsubscribed regions.
+// So `unset` should be called after a region gets locked in RegionRangeLock.
+func (m *rangeTsMap) unset(startKey, endKey []byte) {
+	var neighbor rangeTsEntry
+	var exist bool
+	startEntry := rangeTsEntryWithKey(startKey)
+	endEntry := rangeTsEntryWithKey(endKey)
+
+	// Check and update the range overlapped with endKey is it exists.
+	if !bytes.Equal(m.end, endKey) {
+		if neighbor, exist = m.m.Get(endEntry); !exist {
+			var found bool
+			var endKeyOverlapped rangeTsEntry
+			m.m.DescendLessOrEqual(endEntry, func(i rangeTsEntry) bool {
+				found = true
+				endKeyOverlapped.ts = i.ts
+				endKeyOverlapped.isSet = i.isSet
+				return false
+			})
+			if found {
+				if !endKeyOverlapped.isSet {
+					log.Panic("rangeTsMap double unset")
+				}
+				endKeyOverlapped.startKey = endKey
+				m.m.ReplaceOrInsert(endKeyOverlapped)
+			}
+		} else if !neighbor.isSet {
+			m.m.Delete(neighbor)
+		}
+	}
+
+	// Check and update the range overlapped with startKey if it exists.
+	exist = false
+	m.m.DescendLessOrEqual(startEntry, func(i rangeTsEntry) bool {
+		if bytes.Compare(i.startKey, startKey) < 0 {
+			neighbor = i
+			exist = true
+			return false
+		}
+		return true
+	})
+	shouldInsert := !exist || neighbor.isSet
+
+	// Check and delete all covered ranges.
+	entriesToDelete := make([]rangeTsEntry, 0)
+	m.m.AscendRange(startEntry, endEntry, func(i rangeTsEntry) bool {
+		if !i.isSet {
+			log.Panic("rangeTsMap double unset")
+		}
+		entriesToDelete = append(entriesToDelete, i)
+		return true
 	})
+	for _, e := range entriesToDelete {
+		m.m.Delete(e)
+	}
+
+	if shouldInsert {
+		m.m.ReplaceOrInsert(rangeTsEntry{startKey: startKey, isSet: false})
+	}
 }
 
-// GetMin gets the min ts value among the given range. endKey must be greater than startKey.
-func (m *rangeTsMap) GetMin(startKey, endKey []byte) uint64 {
+func (m *rangeTsMap) getMinTsInRange(startKey, endKey []byte) uint64 {
 	var ts uint64 = math.MaxUint64
-	m.m.DescendLessOrEqual(rangeTsEntryWithKey(startKey), func(i *rangeTsEntry) bool {
-		ts = i.ts
-		return false
-	})
-	m.m.AscendRange(rangeTsEntryWithKey(startKey), rangeTsEntryWithKey(endKey),
-		func(i *rangeTsEntry) bool {
-			thisTs := i.ts
-			if ts > thisTs {
-				ts = thisTs
+
+	startEntry := rangeTsEntryWithKey(startKey)
+	if _, ok := m.m.Get(startEntry); !ok {
+		m.m.DescendLessOrEqual(startEntry, func(i rangeTsEntry) bool {
+			if !i.isSet {
+				log.Panic("rangeTsMap get after unset")
 			}
-			return true
+			ts = i.ts
+			return false
 		})
+	}
+
+	endEntry := rangeTsEntryWithKey(endKey)
+	m.m.AscendRange(startEntry, endEntry, func(i rangeTsEntry) bool {
+		if !i.isSet {
+			log.Panic("rangeTsMap get after unset")
+		}
+		if ts > i.ts {
+			ts = i.ts
+		}
+		return true
+	})
+
+	return ts
+}
+
+func (m *rangeTsMap) getMinTs() uint64 {
+	var ts uint64 = math.MaxUint64
+
+	m.m.Ascend(func(i rangeTsEntry) bool {
+		if i.isSet && ts > i.ts {
+			ts = i.ts
+		}
+		return true
+	})
+
 	return ts
 }
 
@@ -150,12 +279,11 @@ type RegionRangeLock struct {
 	totalSpan         tablepb.Span
 	changefeedLogInfo string
 
-	mu                sync.Mutex
-	rangeCheckpointTs *rangeTsMap
-	rangeLock         *btree.BTreeG[*rangeLockEntry]
-	regionIDLock      map[uint64]*rangeLockEntry
-	stopped           bool
-	refCount          uint64
+	mu              sync.RWMutex
+	rangeResolvedTs *rangeTsMap
+	rangeLock       *btree.BTreeG[*rangeLockEntry]
+	regionIDLock    map[uint64]*rangeLockEntry
+	stopped         bool
 }
 
 // NewRegionRangeLock creates a new RegionRangeLock.
@@ -167,7 +295,7 @@ func NewRegionRangeLock(
 		id:                id,
 		totalSpan:         tablepb.Span{StartKey: startKey, EndKey: endKey},
 		changefeedLogInfo: changefeedLogInfo,
-		rangeCheckpointTs: newRangeTsMap(startKey, endKey, startTs),
+		rangeResolvedTs:   newRangeTsMap(startKey, endKey, startTs),
 		rangeLock:         btree.NewG(16, rangeLockEntryLess),
 		regionIDLock:      make(map[uint64]*rangeLockEntry),
 	}
@@ -218,31 +346,32 @@ func (l *RegionRangeLock) tryLockRange(startKey, endKey []byte, regionID, versio
 	overlappingEntries := l.getOverlappedEntries(startKey, endKey, regionID)
 
 	if len(overlappingEntries) == 0 {
-		checkpointTs := l.rangeCheckpointTs.GetMin(startKey, endKey)
+		resolvedTs := l.rangeResolvedTs.getMinTsInRange(startKey, endKey)
 		newEntry := &rangeLockEntry{
 			startKey: startKey,
 			endKey:   endKey,
 			regionID: regionID,
 			version:  version,
 		}
-		newEntry.state.CheckpointTs.Store(checkpointTs)
+		newEntry.state.ResolvedTs.Store(resolvedTs)
 		newEntry.state.Created = time.Now()
 		l.rangeLock.ReplaceOrInsert(newEntry)
 		l.regionIDLock[regionID] = newEntry
 
+		l.rangeResolvedTs.unset(startKey, endKey)
 		log.Debug("range locked",
 			zap.String("changefeed", l.changefeedLogInfo),
-			zap.Uint64("lockID", l.id), zap.Uint64("regionID", regionID),
+			zap.Uint64("lockID", l.id),
+			zap.Uint64("regionID", regionID),
 			zap.Uint64("version", version),
-			zap.Uint64("checkpointTs", checkpointTs),
+			zap.Uint64("resolvedTs", resolvedTs),
 			zap.String("startKey", hex.EncodeToString(startKey)),
 			zap.String("endKey", hex.EncodeToString(endKey)))
 
-		l.refCount += 1
 		return LockRangeResult{
-			Status:       LockRangeStatusSuccess,
-			CheckpointTs: checkpointTs,
-			LockedRange:  &newEntry.state,
+			Status:      LockRangeStatusSuccess,
+			ResolvedTs:  resolvedTs,
+			LockedRange: &newEntry.state,
 		}, nil
 	}
 
@@ -348,11 +477,11 @@ func (l *RegionRangeLock) LockRange(
 	return res
 }
 
-// UnlockRange unlocks a range and update checkpointTs of the range to specified value.
+// UnlockRange unlocks a range and update resolvedTs of the range to specified value.
 // If it returns true it means it is stopped and all ranges are unlocked correctly.
 func (l *RegionRangeLock) UnlockRange(
 	startKey, endKey []byte, regionID, version uint64,
-	checkpointTs ...uint64,
+	resolvedTs ...uint64,
 ) (drained bool) {
 	l.mu.Lock()
 	defer l.mu.Unlock()
@@ -382,8 +511,7 @@ func (l *RegionRangeLock) UnlockRange(
 			zap.String("regionIDLockEntry", l.regionIDLock[regionID].String()))
 	}
 	delete(l.regionIDLock, regionID)
-	l.refCount -= 1
-	drained = l.stopped && l.refCount == 0
+	drained = l.stopped && len(l.regionIDLock) == 0
 
 	if entry.version != version || !bytes.Equal(entry.endKey, endKey) {
 		log.Panic("unlocking region doesn't match the locked region",
@@ -403,18 +531,18 @@ func (l *RegionRangeLock) UnlockRange(
 		panic("unreachable")
 	}
 
-	var newCheckpointTs uint64
-	if len(checkpointTs) > 0 {
-		newCheckpointTs = checkpointTs[0]
+	var newResolvedTs uint64
+	if len(resolvedTs) > 0 {
+		newResolvedTs = resolvedTs[0]
 	} else {
-		newCheckpointTs = entry.state.CheckpointTs.Load()
+		newResolvedTs = entry.state.ResolvedTs.Load()
 	}
 
-	l.rangeCheckpointTs.Set(startKey, endKey, newCheckpointTs)
+	l.rangeResolvedTs.set(startKey, endKey, newResolvedTs)
 	log.Debug("unlocked range",
 		zap.String("changefeed", l.changefeedLogInfo),
 		zap.Uint64("lockID", l.id), zap.Uint64("regionID", entry.regionID),
-		zap.Uint64("checkpointTs", newCheckpointTs),
+		zap.Uint64("resolvedTs", newResolvedTs),
 		zap.String("startKey", hex.EncodeToString(startKey)),
 		zap.String("endKey", hex.EncodeToString(endKey)))
 	return
@@ -422,16 +550,16 @@ func (l *RegionRangeLock) UnlockRange(
 
 // LockedRanges returns count of locked ranges.
 func (l *RegionRangeLock) LockedRanges() int {
-	l.mu.Lock()
-	defer l.mu.Unlock()
+	l.mu.RLock()
+	defer l.mu.RUnlock()
 	return l.rangeLock.Len()
 }
 
 // RefCount returns how many ranges are locked.
 func (l *RegionRangeLock) RefCount() uint64 {
-	l.mu.Lock()
-	defer l.mu.Unlock()
-	return l.refCount
+	l.mu.RLock()
+	defer l.mu.RUnlock()
+	return uint64(len(l.regionIDLock))
 }
 
 // Stop stops the instance.
@@ -439,7 +567,7 @@ func (l *RegionRangeLock) Stop() (drained bool) {
 	l.mu.Lock()
 	defer l.mu.Unlock()
 	l.stopped = true
-	return l.stopped && l.refCount == 0
+	return l.stopped && len(l.regionIDLock) == 0
 }
 
 const (
@@ -455,7 +583,7 @@ const (
 
 // LockRangeResult represents the result of LockRange method of RegionRangeLock.
 // If Status is LockRangeStatusSuccess:
-//   - CheckpointTs will be the minimal checkpoint ts among the locked range;
+//   - ResolvedTs will be the minimal checkpoint ts among the locked range;
 //   - LockedRange is for recording real-time state changes;
 //
 // If Status is LockRangeStatusWait, it means the lock cannot be acquired immediately. WaitFn must be invoked to
@@ -464,31 +592,31 @@ const (
 // If Status is LockRangeStatusStale, it means the LockRange request is stale because there's already a overlapping
 // locked range, whose version is greater or equals to the requested one.
 type LockRangeResult struct {
-	Status       int
-	CheckpointTs uint64
-	LockedRange  *LockedRange
-	WaitFn       func() LockRangeResult
-	RetryRanges  []tablepb.Span
+	Status      int
+	ResolvedTs  uint64
+	LockedRange *LockedRange
+	WaitFn      func() LockRangeResult
+	RetryRanges []tablepb.Span
 }
 
 // LockedRange is returned by `RegionRangeLock.LockRange`, which can be used to
 // collect informations for the range. And collected informations can be accessed
 // by iterating `RegionRangeLock`.
 type LockedRange struct {
-	CheckpointTs atomic.Uint64
-	Initialzied  atomic.Bool
-	Created      time.Time
+	ResolvedTs  atomic.Uint64
+	Initialzied atomic.Bool
+	Created     time.Time
 }
 
 // CollectLockedRangeAttrs collects locked range attributes.
 func (l *RegionRangeLock) CollectLockedRangeAttrs(
 	action func(regionID, version uint64, state *LockedRange, span tablepb.Span),
 ) (r CollectedLockedRangeAttrs) {
-	l.mu.Lock()
-	defer l.mu.Unlock()
+	l.mu.RLock()
+	defer l.mu.RUnlock()
 	r.LockedRegionCount = l.rangeLock.Len()
-	r.FastestRegion.CheckpointTs = 0
-	r.SlowestRegion.CheckpointTs = math.MaxUint64
+	r.FastestRegion.ResolvedTs = 0
+	r.SlowestRegion.ResolvedTs = math.MaxUint64
 
 	lastEnd := l.totalSpan.StartKey
 	l.rangeLock.Ascend(func(item *rangeLockEntry) bool {
@@ -497,18 +625,20 @@ func (l *RegionRangeLock) CollectLockedRangeAttrs(
 			action(item.regionID, item.version, &item.state, span)
 		}
 		if spanz.EndCompare(lastEnd, item.startKey) < 0 {
-			r.Holes = append(r.Holes, tablepb.Span{StartKey: lastEnd, EndKey: item.startKey})
+			span := tablepb.Span{StartKey: lastEnd, EndKey: item.startKey}
+			ts := l.rangeResolvedTs.getMinTsInRange(lastEnd, item.startKey)
+			r.Holes = append(r.Holes, HoleAttrs{Span: span, ResolvedTs: ts})
 		}
-		ckpt := item.state.CheckpointTs.Load()
-		if ckpt > r.FastestRegion.CheckpointTs {
+		ckpt := item.state.ResolvedTs.Load()
+		if ckpt > r.FastestRegion.ResolvedTs {
 			r.FastestRegion.RegionID = item.regionID
-			r.FastestRegion.CheckpointTs = ckpt
+			r.FastestRegion.ResolvedTs = ckpt
 			r.FastestRegion.Initialized = item.state.Initialzied.Load()
 			r.FastestRegion.Created = item.state.Created
 		}
-		if ckpt < r.SlowestRegion.CheckpointTs {
+		if ckpt < r.SlowestRegion.ResolvedTs {
 			r.SlowestRegion.RegionID = item.regionID
-			r.SlowestRegion.CheckpointTs = ckpt
+			r.SlowestRegion.ResolvedTs = ckpt
 			r.SlowestRegion.Initialized = item.state.Initialzied.Load()
 			r.SlowestRegion.Created = item.state.Created
 		}
@@ -516,7 +646,9 @@ func (l *RegionRangeLock) CollectLockedRangeAttrs(
 		return true
 	})
 	if spanz.EndCompare(lastEnd, l.totalSpan.EndKey) < 0 {
-		r.Holes = append(r.Holes, tablepb.Span{StartKey: lastEnd, EndKey: l.totalSpan.EndKey})
+		span := tablepb.Span{StartKey: lastEnd, EndKey: l.totalSpan.EndKey}
+		ts := l.rangeResolvedTs.getMinTsInRange(lastEnd, l.totalSpan.EndKey)
+		r.Holes = append(r.Holes, HoleAttrs{Span: span, ResolvedTs: ts})
 	}
 	return
 }
@@ -524,15 +656,47 @@ func (l *RegionRangeLock) CollectLockedRangeAttrs(
 // CollectedLockedRangeAttrs returns by `RegionRangeLock.CollectedLockedRangeAttrs`.
 type CollectedLockedRangeAttrs struct {
 	LockedRegionCount int
-	Holes             []tablepb.Span
-	FastestRegion     LockedRangeAttrs
-	SlowestRegion     LockedRangeAttrs
+
+	// If Holes isn't empty it means some regions are not captured at the time.
+	// Those regions can just split or merge, or be transferred, or temporarily be unavailable.
+	Holes []HoleAttrs
+
+	FastestRegion LockedRangeAttrs
+	SlowestRegion LockedRangeAttrs
 }
 
 // LockedRangeAttrs is like `LockedRange`, but only contains some read-only attributes.
 type LockedRangeAttrs struct {
-	RegionID     uint64
-	CheckpointTs uint64
-	Initialized  bool
-	Created      time.Time
+	RegionID    uint64
+	ResolvedTs  uint64
+	Initialized bool
+	Created     time.Time
+}
+
+// HoleAttrs is used for `CollectedLockedRangeAttrs`.
+type HoleAttrs struct {
+	Span       tablepb.Span
+	ResolvedTs uint64
+}
+
+// CalculateMinResolvedTs gets the minimum checkpoint timestamp from range lock.
+func (l *RegionRangeLock) CalculateMinResolvedTs() uint64 {
+	l.mu.RLock()
+	defer l.mu.RUnlock()
+
+	var minTs uint64 = math.MaxUint64
+	l.rangeLock.Ascend(func(item *rangeLockEntry) bool {
+		ts := item.state.ResolvedTs.Load()
+		if ts < minTs {
+			minTs = ts
+		}
+		return true
+	})
+
+	unlockedMinTs := l.rangeResolvedTs.getMinTs()
+	if unlockedMinTs < minTs {
+		minTs = unlockedMinTs
+	}
+
+	return minTs
 }
diff --git a/cdc/kv/shared_client.go b/cdc/kv/shared_client.go
index c9cfbfc50..69588d0f3 100644
--- a/cdc/kv/shared_client.go
+++ b/cdc/kv/shared_client.go
@@ -16,8 +16,6 @@ package kv
 import (
 	"context"
 	"encoding/binary"
-	"fmt"
-	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -124,6 +122,8 @@ type requestedTable struct {
 	rangeLock *regionlock.RegionRangeLock
 	eventCh   chan<- MultiplexingEvent
 
+	lastAdvanceTime atomic.Int64
+
 	// To handle table removing.
 	stopped atomic.Bool
 
@@ -671,7 +671,7 @@ func (s *SharedClient) resolveLock(ctx context.Context) error {
 	}
 
 	doResolve := func(regionID uint64, state *regionlock.LockedRange, maxVersion uint64) {
-		if state.CheckpointTs.Load() > maxVersion || !state.Initialzied.Load() {
+		if state.ResolvedTs.Load() > maxVersion || !state.Initialzied.Load() {
 			return
 		}
 		if lastRun, ok := resolveLastRun[regionID]; ok {
@@ -725,7 +725,7 @@ func (s *SharedClient) logSlowRegions(ctx context.Context) error {
 		s.totalSpans.RLock()
 		for subscriptionID, rt := range s.totalSpans.v {
 			attr := rt.rangeLock.CollectLockedRangeAttrs(nil)
-			ckptTime := oracle.GetTimeFromTS(attr.SlowestRegion.CheckpointTs)
+			ckptTime := oracle.GetTimeFromTS(attr.SlowestRegion.ResolvedTs)
 			if attr.SlowestRegion.Initialized {
 				if currTime.Sub(ckptTime) > 2*resolveLockMinInterval {
 					log.Info("event feed finds a initialized slow region",
@@ -748,15 +748,11 @@ func (s *SharedClient) logSlowRegions(ctx context.Context) error {
 					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.Any("subscriptionID", subscriptionID),
-					zap.String("holes", strings.Join(holes, ", ")))
+					zap.Any("holes", attr.Holes))
 			}
 		}
 		s.totalSpans.RUnlock()
@@ -780,7 +776,7 @@ func (s *SharedClient) newRequestedTable(
 
 	rt.postUpdateRegionResolvedTs = func(regionID, _ uint64, state *regionlock.LockedRange, _ tablepb.Span) {
 		maxVersion := rt.staleLocksVersion.Load()
-		if state.CheckpointTs.Load() <= maxVersion && state.Initialzied.Load() {
+		if state.ResolvedTs.Load() <= maxVersion && state.Initialzied.Load() {
 			enter := time.Now()
 			s.resolveLockCh.In() <- resolveLockTask{regionID, maxVersion, state, enter}
 		}
diff --git a/cdc/kv/shared_region_worker.go b/cdc/kv/shared_region_worker.go
index de1448ce4..b2cca89be 100644
--- a/cdc/kv/shared_region_worker.go
+++ b/cdc/kv/shared_region_worker.go
@@ -39,6 +39,7 @@ type eventItem struct {
 	state *regionFeedState
 }
 
+// NOTE: all regions must come from the same requestedTable, and regions will never be empty.
 type resolvedTsBatch struct {
 	ts      uint64
 	regions []*regionFeedState
@@ -174,6 +175,14 @@ func (w *sharedRegionWorker) handleEventEntry(ctx context.Context, x *cdcpb.Even
 }
 
 func (w *sharedRegionWorker) handleResolvedTs(ctx context.Context, batch resolvedTsBatch) {
+	if w.client.config.KVClient.AdvanceIntervalInMs > 0 {
+		w.advanceTableSpan(ctx, batch)
+	} else {
+		w.forwardResolvedTsToPullerFrontier(ctx, batch)
+	}
+}
+
+func (w *sharedRegionWorker) forwardResolvedTsToPullerFrontier(ctx context.Context, batch resolvedTsBatch) {
 	resolvedSpans := make(map[SubscriptionID]*struct {
 		spans          []model.RegionComparableSpan
 		requestedTable *requestedTable
@@ -231,3 +240,45 @@ func (w *sharedRegionWorker) handleResolvedTs(ctx context.Context, batch resolve
 		}
 	}
 }
+
+func (w *sharedRegionWorker) advanceTableSpan(ctx context.Context, batch resolvedTsBatch) {
+	for _, state := range batch.regions {
+		if state.isStale() || !state.isInitialized() {
+			continue
+		}
+
+		regionID := state.getRegionID()
+		lastResolvedTs := state.getLastResolvedTs()
+		if batch.ts < lastResolvedTs {
+			log.Debug("The resolvedTs is fallen back in kvclient",
+				zap.String("namespace", w.changefeed.Namespace),
+				zap.String("changefeed", w.changefeed.ID),
+				zap.Uint64("regionID", regionID),
+				zap.Uint64("resolvedTs", batch.ts),
+				zap.Uint64("lastResolvedTs", lastResolvedTs))
+			continue
+		}
+		state.updateResolvedTs(batch.ts)
+	}
+
+	rt := batch.regions[0].sri.requestedTable
+	now := time.Now().UnixMilli()
+	lastAdvance := rt.lastAdvanceTime.Load()
+	if now-lastAdvance > int64(w.client.config.KVClient.AdvanceIntervalInMs) && rt.lastAdvanceTime.CompareAndSwap(lastAdvance, now) {
+		ts := rt.rangeLock.CalculateMinResolvedTs()
+		if ts > rt.startTs {
+			revent := model.RegionFeedEvent{
+				Resolved: &model.ResolvedSpans{
+					Spans:      []model.RegionComparableSpan{{Span: rt.span, Region: 0}},
+					ResolvedTs: ts,
+				},
+			}
+			x := rt.associateSubscriptionID(revent)
+			select {
+			case rt.eventCh <- x:
+				w.metrics.metricSendEventResolvedCounter.Add(float64(len(batch.regions)))
+			case <-ctx.Done():
+			}
+		}
+	}
+}
diff --git a/cdc/syncpointstore/mysql_syncpoint_store.go b/cdc/syncpointstore/mysql_syncpoint_store.go
index ce4e88df2..53aaa4f0c 100644
--- a/cdc/syncpointstore/mysql_syncpoint_store.go
+++ b/cdc/syncpointstore/mysql_syncpoint_store.go
@@ -26,17 +26,11 @@ import (
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/errorutil"
+	"github.com/pingcap/tiflow/pkg/filter"
 	"github.com/pingcap/tiflow/pkg/sink/mysql"
 	"go.uber.org/zap"
 )
 
-const (
-	// syncPointTableName is the name of table where all syncpoint maps sit
-	syncPointTableName string = "syncpoint_v1"
-	// schemaName is the name of database where syncPoint maps sit
-	schemaName = "tidb_cdc"
-)
-
 type mysqlSyncPointStore struct {
 	db                     *sql.DB
 	clusterID              string
@@ -88,7 +82,7 @@ func newMySQLSyncPointStore(
 }
 
 func (s *mysqlSyncPointStore) CreateSyncTable(ctx context.Context) error {
-	database := schemaName
+	database := filter.TiCDCSystemSchema
 	tx, err := s.db.BeginTx(ctx, nil)
 	if err != nil {
 		log.Error("create sync table: begin Tx fail", zap.Error(err))
@@ -120,7 +114,7 @@ func (s *mysqlSyncPointStore) CreateSyncTable(ctx context.Context) error {
 		INDEX (created_at),
 		PRIMARY KEY (changefeed, primary_ts)
 	);`
-	query = fmt.Sprintf(query, syncPointTableName)
+	query = fmt.Sprintf(query, filter.SyncPointTable)
 	_, err = tx.Exec(query)
 	if err != nil {
 		err2 := tx.Rollback()
@@ -154,7 +148,7 @@ func (s *mysqlSyncPointStore) SinkSyncPoint(ctx context.Context,
 		return cerror.WrapError(cerror.ErrMySQLTxnError, err)
 	}
 	// insert ts map
-	query := "insert ignore into " + schemaName + "." + syncPointTableName +
+	query := "insert ignore into " + filter.TiCDCSystemSchema + "." + filter.SyncPointTable +
 		"(ticdc_cluster_id, changefeed, primary_ts, secondary_ts) VALUES (?,?,?,?)"
 	_, err = tx.Exec(query, s.clusterID, id.ID, checkpointTs, secondaryTs)
 	if err != nil {
@@ -186,8 +180,8 @@ func (s *mysqlSyncPointStore) SinkSyncPoint(ctx context.Context,
 	if time.Since(s.lastCleanSyncPointTime) >= s.syncPointRetention {
 		query = fmt.Sprintf(
 			"DELETE IGNORE FROM "+
-				schemaName+"."+
-				syncPointTableName+
+				filter.TiCDCSystemSchema+"."+
+				filter.SyncPointTable+
 				" WHERE ticdc_cluster_id = '%s' and changefeed = '%s' and created_at < (NOW() - INTERVAL %.2f SECOND)",
 			s.clusterID,
 			id.ID,
diff --git a/dm/syncer/ddl.go b/dm/syncer/ddl.go
index 47035e75f..fdca8906d 100644
--- a/dm/syncer/ddl.go
+++ b/dm/syncer/ddl.go
@@ -1173,7 +1173,7 @@ func (ddl *DDLWorker) AstToDDLEvent(qec *queryEventContext, info *ddlInfo) (et b
 					}
 				}
 			case ast.AlterTableReorganizePartition:
-				return bf.ReorganizePartion
+				return bf.ReorganizePartition
 			case ast.AlterTableRebuildPartition:
 				return bf.RebuildPartition
 			case ast.AlterTableCoalescePartitions:
diff --git a/go.mod b/go.mod
index 1d280d822..c6982dfbd 100644
--- a/go.mod
+++ b/go.mod
@@ -69,7 +69,7 @@ require (
 	github.com/pingcap/kvproto v0.0.0-20231222062942-c0c73f41d0b2
 	github.com/pingcap/log v1.1.1-0.20230317032135-a0d097d16e22
 	github.com/pingcap/tidb v1.1.0-beta.0.20240105042433-54d8a1416ab0
-	github.com/pingcap/tidb-tools v0.0.0-20231228035519-c4bdf178b3d6
+	github.com/pingcap/tidb-tools v0.0.0-20240131075425-5bb51786a1dc
 	github.com/pingcap/tidb/pkg/parser v0.0.0-20231229060758-e19e06e1bc19
 	github.com/prometheus/client_golang v1.18.0
 	github.com/prometheus/client_model v0.5.0
diff --git a/go.sum b/go.sum
index 72a4e2408..d9f3afc89 100644
--- a/go.sum
+++ b/go.sum
@@ -867,8 +867,8 @@ github.com/pingcap/sysutil v1.0.1-0.20230407040306-fb007c5aff21 h1:QV6jqlfOkh8hq
 github.com/pingcap/sysutil v1.0.1-0.20230407040306-fb007c5aff21/go.mod h1:QYnjfA95ZaMefyl1NO8oPtKeb8pYUdnDVhQgf+qdpjM=
 github.com/pingcap/tidb v1.1.0-beta.0.20240105042433-54d8a1416ab0 h1:EAeL37UVvIMvAG5JP0qDSBWZvV4tKn0gUbSP05vjtqU=
 github.com/pingcap/tidb v1.1.0-beta.0.20240105042433-54d8a1416ab0/go.mod h1:1jrwjaNIwD9JmFR3NghQ/xu/PCuPQVLhK8sshyFadN0=
-github.com/pingcap/tidb-tools v0.0.0-20231228035519-c4bdf178b3d6 h1:NJQd14hU0pLj8vS+vYtgFw4bXrFIXa/HfMdxALHtgVU=
-github.com/pingcap/tidb-tools v0.0.0-20231228035519-c4bdf178b3d6/go.mod h1:7m72uoBoTrr8JNNA7iFBH824j3clIJx34zlUWJ45Q6Y=
+github.com/pingcap/tidb-tools v0.0.0-20240131075425-5bb51786a1dc h1:z9gMlNH+AMmp62EniIRtyVVgOsrFcuxiUu6Ty60uEMU=
+github.com/pingcap/tidb-tools v0.0.0-20240131075425-5bb51786a1dc/go.mod h1:7m72uoBoTrr8JNNA7iFBH824j3clIJx34zlUWJ45Q6Y=
 github.com/pingcap/tidb/pkg/parser v0.0.0-20231229060758-e19e06e1bc19 h1:nmFYQ4q3m0v7AleVVyiijMtEzfxCh8zqMeuD1npsy+w=
 github.com/pingcap/tidb/pkg/parser v0.0.0-20231229060758-e19e06e1bc19/go.mod h1:yRkiqLFwIqibYg2P7h4bclHjHcJiIFRLKhGRyBcKYus=
 github.com/pingcap/tipb v0.0.0-20230919054518-dfd7d194838f h1:NCiI4Wyu4GkViLGTu6cYcxt79LZ1SenBBQX1OwEV6Jg=
diff --git a/pkg/config/debug.go b/pkg/config/debug.go
index 73821bd0e..d44cb1e71 100644
--- a/pkg/config/debug.go
+++ b/pkg/config/debug.go
@@ -26,9 +26,6 @@ type DebugConfig struct {
 	// Scheduler is the configuration of the two-phase scheduler.
 	Scheduler *SchedulerConfig `toml:"scheduler" json:"scheduler"`
 
-	// EnableKVConnectBackOff enables the backoff for kv connect.
-	EnableKVConnectBackOff bool `toml:"enable-kv-connect-backoff" json:"enable-kv-connect-backoff"`
-
 	// CDCV2 enables ticdc version 2 implementation with new metastore
 	CDCV2 *CDCV2 `toml:"cdc-v2" json:"cdc-v2"`
 
diff --git a/pkg/config/kvclient.go b/pkg/config/kvclient.go
index 9bdd8060e..f6a9c49af 100644
--- a/pkg/config/kvclient.go
+++ b/pkg/config/kvclient.go
@@ -22,6 +22,8 @@ type KVClientConfig struct {
 	WorkerConcurrent uint `toml:"worker-concurrent" json:"worker-concurrent"`
 	// how many grpc streams will be established to every TiKV node
 	GrpcStreamConcurrent uint `toml:"grpc-stream-concurrent" json:"grpc-stream-concurrent"`
+	// Advance table ResolvedTs interval.
+	AdvanceIntervalInMs uint `toml:"advance-interval-in-ms" json:"advance-interval-in-ms"`
 	// how many goroutines to maintain frontiers.
 	FrontierConcurrent uint `toml:"frontier-concurrent" json:"frontier-concurrent"`
 	// background workerpool size, the workrpool is shared by all goroutines in cdc server
diff --git a/pkg/config/server_config.go b/pkg/config/server_config.go
index 91f40e3f6..fafc4565f 100644
--- a/pkg/config/server_config.go
+++ b/pkg/config/server_config.go
@@ -111,6 +111,7 @@ var defaultServerConfig = &ServerConfig{
 		EnableMultiplexing:   true,
 		WorkerConcurrent:     8,
 		GrpcStreamConcurrent: 1,
+		AdvanceIntervalInMs:  300,
 		FrontierConcurrent:   8,
 		WorkerPoolSize:       0, // 0 will use NumCPU() * 2
 		RegionScanLimit:      40,
@@ -132,9 +133,8 @@ var defaultServerConfig = &ServerConfig{
 		},
 		Messages: defaultMessageConfig.Clone(),
 
-		Scheduler:              NewDefaultSchedulerConfig(),
-		EnableKVConnectBackOff: false,
-		CDCV2:                  &CDCV2{Enable: false},
+		Scheduler: NewDefaultSchedulerConfig(),
+		CDCV2:     &CDCV2{Enable: false},
 		Puller: &PullerConfig{
 			EnableResolvedTsStuckDetection: false,
 			ResolvedTsStuckInterval:        TomlDuration(5 * time.Minute),
diff --git a/pkg/filter/expr_filter.go b/pkg/filter/expr_filter.go
index 61835c6d6..25b04cf73 100644
--- a/pkg/filter/expr_filter.go
+++ b/pkg/filter/expr_filter.go
@@ -419,6 +419,9 @@ func (f *dmlExprFilter) shouldSkipDML(
 	rawRow model.RowChangedDatums,
 	ti *model.TableInfo,
 ) (bool, error) {
+	if len(f.rules) == 0 {
+		return false, nil
+	}
 	// for defense purpose, normally the row and ti should not be nil.
 	if ti == nil || row == nil || rawRow.IsEmpty() {
 		return false, nil
diff --git a/pkg/filter/filter.go b/pkg/filter/filter.go
index f1e464f7b..a3c017c34 100644
--- a/pkg/filter/filter.go
+++ b/pkg/filter/filter.go
@@ -14,51 +14,77 @@
 package filter
 
 import (
+	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	timodel "github.com/pingcap/tidb/pkg/parser/model"
 	tfilter "github.com/pingcap/tidb/pkg/util/table-filter"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/pkg/config"
 )
 
-// allowDDLList is a list of DDL types that can be applied to cdc's schema storage.
-// It's a white list.
-var allowDDLList = []timodel.ActionType{
-	timodel.ActionCreateSchema,
-	timodel.ActionDropSchema,
-	timodel.ActionCreateTable,
-	timodel.ActionDropTable,
-	timodel.ActionAddColumn,
-	timodel.ActionDropColumn,
-	timodel.ActionAddIndex,
-	timodel.ActionDropIndex,
-	timodel.ActionTruncateTable,
-	timodel.ActionModifyColumn,
-	timodel.ActionRenameTable,
-	timodel.ActionRenameTables,
-	timodel.ActionSetDefaultValue,
-	timodel.ActionModifyTableComment,
-	timodel.ActionRenameIndex,
-	timodel.ActionAddTablePartition,
-	timodel.ActionDropTablePartition,
-	timodel.ActionCreateView,
-	timodel.ActionModifyTableCharsetAndCollate,
-	timodel.ActionTruncateTablePartition,
-	timodel.ActionDropView,
-	timodel.ActionRecoverTable,
-	timodel.ActionModifySchemaCharsetAndCollate,
-	timodel.ActionAddPrimaryKey,
-	timodel.ActionDropPrimaryKey,
-	timodel.ActionAddColumns,  // Removed in TiDB v6.2.0, see https://github.com/pingcap/tidb/pull/35862.
-	timodel.ActionDropColumns, // Removed in TiDB v6.2.0
-	timodel.ActionRebaseAutoID,
-	timodel.ActionAlterIndexVisibility,
-	timodel.ActionMultiSchemaChange,
-	timodel.ActionExchangeTablePartition,
-	timodel.ActionReorganizePartition,
-	timodel.ActionAlterTTLInfo,
-	timodel.ActionAlterTTLRemove,
-	timodel.ActionAlterTablePartitioning,
-	timodel.ActionRemovePartitioning,
+const (
+	// SyncPointTable is the tale name use to write ts-map when sync-point is enable.
+	SyncPointTable = "syncpoint_v1"
+	// TiCDCSystemSchema is the schema only use by TiCDC.
+	TiCDCSystemSchema = "tidb_cdc"
+)
+
+// ddlWhiteListMap is a map of all DDL types that can be applied to cdc's schema storage.
+var ddlWhiteListMap = map[timodel.ActionType]bf.EventType{
+	// schema related DDLs
+	timodel.ActionCreateSchema:                  bf.CreateDatabase,
+	timodel.ActionDropSchema:                    bf.DropDatabase,
+	timodel.ActionModifySchemaCharsetAndCollate: bf.ModifySchemaCharsetAndCollate,
+
+	// table related DDLs
+	timodel.ActionCreateTable:                  bf.CreateTable,
+	timodel.ActionDropTable:                    bf.DropTable,
+	timodel.ActionTruncateTable:                bf.TruncateTable,
+	timodel.ActionRenameTable:                  bf.RenameTable,
+	timodel.ActionRenameTables:                 bf.RenameTable,
+	timodel.ActionRecoverTable:                 bf.RecoverTable,
+	timodel.ActionModifyTableComment:           bf.ModifyTableComment,
+	timodel.ActionModifyTableCharsetAndCollate: bf.ModifyTableCharsetAndCollate,
+
+	// view related DDLs
+	timodel.ActionCreateView: bf.CreateView,
+	timodel.ActionDropView:   bf.DropView,
+
+	// partition related DDLs
+	timodel.ActionAddTablePartition:      bf.AddTablePartition,
+	timodel.ActionDropTablePartition:     bf.DropTablePartition,
+	timodel.ActionTruncateTablePartition: bf.TruncateTablePartition,
+	timodel.ActionExchangeTablePartition: bf.ExchangePartition,
+	timodel.ActionReorganizePartition:    bf.ReorganizePartition,
+	timodel.ActionAlterTablePartitioning: bf.AlterTablePartitioning,
+	timodel.ActionRemovePartitioning:     bf.RemovePartitioning,
+
+	// column related DDLs
+	timodel.ActionAddColumn:       bf.AddColumn,
+	timodel.ActionDropColumn:      bf.DropColumn,
+	timodel.ActionModifyColumn:    bf.ModifyColumn,
+	timodel.ActionSetDefaultValue: bf.SetDefaultValue,
+
+	// index related DDLs
+	timodel.ActionRebaseAutoID:         bf.RebaseAutoID,
+	timodel.ActionAddPrimaryKey:        bf.AddPrimaryKey,
+	timodel.ActionDropPrimaryKey:       bf.DropPrimaryKey,
+	timodel.ActionAddIndex:             bf.CreateIndex,
+	timodel.ActionDropIndex:            bf.DropIndex,
+	timodel.ActionRenameIndex:          bf.RenameIndex,
+	timodel.ActionAlterIndexVisibility: bf.AlterIndexVisibility,
+
+	// TTL related DDLs
+	timodel.ActionAlterTTLInfo:   bf.AlterTTLInfo,
+	timodel.ActionAlterTTLRemove: bf.AlterTTLRemove,
+
+	// difficult to classify DDLs
+	timodel.ActionMultiSchemaChange: bf.MultiSchemaChange,
+
+	// deprecated DDLs,see https://github.com/pingcap/tidb/pull/35862.
+	// DDL types below are deprecated in TiDB v6.2.0, but we still keep them here
+	// In case that some users will use TiCDC to replicate data from TiDB v6.1.x.
+	timodel.ActionAddColumns:  bf.AddColumn,
+	timodel.ActionDropColumns: bf.DropColumn,
 }
 
 // Filter are safe for concurrent use.
@@ -110,7 +136,7 @@ func NewFilter(cfg *config.ReplicaConfig, tz string) (Filter, error) {
 	if err != nil {
 		return nil, err
 	}
-	sqlEventFilter, err := newSQLEventFilter(cfg.Filter, cfg.SQLMode)
+	sqlEventFilter, err := newSQLEventFilter(cfg.Filter)
 	if err != nil {
 		return nil, err
 	}
@@ -215,12 +241,8 @@ func (f *filter) shouldIgnoreStartTs(ts uint64) bool {
 }
 
 func isAllowedDDL(actionType timodel.ActionType) bool {
-	for _, action := range allowDDLList {
-		if actionType == action {
-			return true
-		}
-	}
-	return false
+	_, ok := ddlWhiteListMap[actionType]
+	return ok
 }
 
 // IsSchemaDDL returns true if the action type is a schema DDL.
diff --git a/pkg/filter/sql_event_filter.go b/pkg/filter/sql_event_filter.go
index b18618eb7..d9c1bf794 100644
--- a/pkg/filter/sql_event_filter.go
+++ b/pkg/filter/sql_event_filter.go
@@ -14,14 +14,9 @@
 package filter
 
 import (
-	"fmt"
-	"sync"
-
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
-	"github.com/pingcap/tidb/pkg/parser"
-	"github.com/pingcap/tidb/pkg/parser/mysql"
 	tfilter "github.com/pingcap/tidb/pkg/util/table-filter"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/pkg/config"
@@ -84,8 +79,8 @@ func newSQLEventFilterRule(cfg *config.EventFilterRule) (*sqlEventRule, error) {
 }
 
 func verifyIgnoreEvents(types []bf.EventType) error {
-	typesMap := make(map[bf.EventType]struct{}, len(supportedEventTypes))
-	for _, et := range supportedEventTypes {
+	typesMap := make(map[bf.EventType]struct{}, len(SupportedEventTypes()))
+	for _, et := range SupportedEventTypes() {
 		typesMap[et] = struct{}{}
 	}
 	for _, et := range types {
@@ -98,28 +93,11 @@ func verifyIgnoreEvents(types []bf.EventType) error {
 
 // sqlEventFilter is a filter that filters DDL/DML event by its type or query.
 type sqlEventFilter struct {
-	// Please be careful, parser.Parser is not thread safe.
-	pLock sync.Mutex
-	// Currently, parser is only used to parse ddl query.
-	// So we can use a lock to protect it.
-	// If we want to use it to parse dml query in the future,
-	// we should create a parser for each goroutine.
-	ddlParser *parser.Parser
-	rules     []*sqlEventRule
+	rules []*sqlEventRule
 }
 
-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: p,
-	}
+func newSQLEventFilter(cfg *config.FilterConfig) (*sqlEventFilter, error) {
+	res := &sqlEventFilter{}
 	for _, rule := range cfg.EventFilters {
 		if err := res.addRule(rule); err != nil {
 			return nil, errors.Trace(err)
@@ -155,17 +133,12 @@ func (f *sqlEventFilter) getRules(schema, table string) []*sqlEventRule {
 
 // skipDDLEvent skips ddl event by its type and query.
 func (f *sqlEventFilter) shouldSkipDDL(ddl *model.DDLEvent) (bool, error) {
+	if len(f.rules) == 0 {
+		return false, nil
+	}
 	schema := ddl.TableInfo.TableName.Schema
 	table := ddl.TableInfo.TableName.Table
-	log.Info("sql event filter handle ddl event",
-		zap.Any("ddlType", ddl.Type), zap.String("schema", schema),
-		zap.String("table", table), zap.String("query", ddl.Query))
-	f.pLock.Lock()
-	evenType, err := ddlToEventType(f.ddlParser, ddl.Query, ddl.Type)
-	f.pLock.Unlock()
-	if err != nil {
-		return false, err
-	}
+	evenType := ddlToEventType(ddl.Type)
 	if evenType == bf.NullEvent {
 		log.Warn("sql event filter unsupported ddl type, do nothing",
 			zap.String("type", ddl.Type.String()),
@@ -185,12 +158,31 @@ func (f *sqlEventFilter) shouldSkipDDL(ddl *model.DDLEvent) (bool, error) {
 		if action == bf.Ignore {
 			return true, nil
 		}
+
+		// If the ddl is alter table's subtype,
+		// we need try to filter it by bf.AlterTable.
+		if isAlterTable(ddl.Type) {
+			action, err = rule.bf.Filter(
+				binlogFilterSchemaPlaceholder,
+				binlogFilterTablePlaceholder,
+				bf.AlterTable, ddl.Query)
+			if err != nil {
+				return false, errors.Trace(err)
+			}
+			if action == bf.Ignore {
+				return true, nil
+			}
+		}
 	}
 	return false, nil
 }
 
 // shouldSkipDML skips dml event by its type.
 func (f *sqlEventFilter) shouldSkipDML(event *model.RowChangedEvent) (bool, error) {
+	if len(f.rules) == 0 {
+		return false, nil
+	}
+
 	var et bf.EventType
 	switch {
 	case event.IsInsert():
@@ -216,29 +208,3 @@ func (f *sqlEventFilter) shouldSkipDML(event *model.RowChangedEvent) (bool, erro
 	}
 	return false, nil
 }
-
-var supportedEventTypes = []bf.EventType{
-	bf.AllDML,
-	bf.AllDDL,
-
-	// dml events
-	bf.InsertEvent,
-	bf.UpdateEvent,
-	bf.DeleteEvent,
-
-	// ddl events
-	bf.CreateSchema,
-	bf.CreateDatabase,
-	bf.DropSchema,
-	bf.DropDatabase,
-	bf.CreateTable,
-	bf.DropTable,
-	bf.RenameTable,
-	bf.TruncateTable,
-	bf.AlterTable,
-	bf.CreateView,
-	bf.DropView,
-	bf.AddTablePartition,
-	bf.DropTablePartition,
-	bf.TruncateTablePartition,
-}
diff --git a/pkg/filter/utils.go b/pkg/filter/utils.go
index 1a45d29cd..7a431ee58 100644
--- a/pkg/filter/utils.go
+++ b/pkg/filter/utils.go
@@ -17,7 +17,6 @@ import (
 	"fmt"
 
 	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
-	"github.com/pingcap/tidb/pkg/parser"
 	timodel "github.com/pingcap/tidb/pkg/parser/model"
 	tifilter "github.com/pingcap/tidb/pkg/util/filter"
 	tfilter "github.com/pingcap/tidb/pkg/util/table-filter"
@@ -27,7 +26,14 @@ import (
 
 // isSysSchema returns true if the given schema is a system schema
 func isSysSchema(db string) bool {
-	return tifilter.IsSystemSchema(db)
+	switch db {
+	// TiCDCSystemSchema is used by TiCDC only.
+	// Tables in TiCDCSystemSchema should not be replicated by cdc.
+	case TiCDCSystemSchema:
+		return true
+	default:
+		return tifilter.IsSystemSchema(db)
+	}
 }
 
 // VerifyTableRules checks the table filter rules in the configuration
@@ -51,37 +57,89 @@ func VerifyTableRules(cfg *config.FilterConfig) (tfilter.Filter, error) {
 }
 
 // ddlToEventType get event type from ddl query.
-func ddlToEventType(p *parser.Parser, query string, jobType timodel.ActionType) (bf.EventType, error) {
-	// Since `Parser` will return a AlterTable type `ast.StmtNode` for table partition related DDL,
-	// we need to check the ActionType of a ddl at first.
-	switch jobType {
-	case timodel.ActionAddTablePartition:
-		return bf.AddTablePartition, nil
-	case timodel.ActionDropTablePartition:
-		return bf.DropTablePartition, nil
-	case timodel.ActionTruncateTablePartition:
-		return bf.TruncateTablePartition, nil
-	}
-	stmt, err := p.ParseOneStmt(query, "", "")
-	if err != nil {
-		return bf.NullEvent, cerror.WrapError(cerror.ErrConvertDDLToEventTypeFailed, err, query)
+func ddlToEventType(jobType timodel.ActionType) bf.EventType {
+	evenType, ok := ddlWhiteListMap[jobType]
+	if ok {
+		return evenType
 	}
-	et := bf.AstToDDLEvent(stmt)
-	// `Parser` will return a `AlterTable` type `ast.StmtNode` for a query like:
-	// `alter table t1 add index (xxx)` and will return a `CreateIndex` type
-	// `ast.StmtNode` for a query like: `create index i on t1 (xxx)`.
-	// So we cast index related DDL to `AlterTable` event type for the sake of simplicity.
-	switch et {
-	case bf.DropIndex:
-		return bf.AlterTable, nil
-	case bf.CreateIndex:
-		return bf.AlterTable, nil
+	return bf.NullEvent
+}
+
+var alterTableSubType = []timodel.ActionType{
+	// table related DDLs
+	timodel.ActionRenameTable,
+	timodel.ActionRenameTables,
+	timodel.ActionModifyTableComment,
+	timodel.ActionModifyTableCharsetAndCollate,
+
+	// partition related DDLs
+	timodel.ActionAddTablePartition,
+	timodel.ActionDropTablePartition,
+	timodel.ActionTruncateTablePartition,
+	timodel.ActionExchangeTablePartition,
+	timodel.ActionReorganizePartition,
+	timodel.ActionAlterTablePartitioning,
+	timodel.ActionRemovePartitioning,
+
+	// column related DDLs
+	timodel.ActionAddColumn,
+	timodel.ActionDropColumn,
+	timodel.ActionModifyColumn,
+	timodel.ActionSetDefaultValue,
+
+	// index related DDLs
+	timodel.ActionRebaseAutoID,
+	timodel.ActionAddPrimaryKey,
+	timodel.ActionDropPrimaryKey,
+	timodel.ActionAddIndex,
+	timodel.ActionDropIndex,
+	timodel.ActionRenameIndex,
+	timodel.ActionAlterIndexVisibility,
+
+	// TTL related DDLs
+	timodel.ActionAlterTTLInfo,
+	timodel.ActionAlterTTLRemove,
+
+	// difficult to classify DDLs
+	timodel.ActionMultiSchemaChange,
+
+	// deprecated DDLs,see https://github.com/pingcap/tidb/pull/35862.
+	// DDL types below are deprecated in TiDB v6.2.0, but we still keep them here
+	// In case that some users will use TiCDC to replicate data from TiDB v6.1.x.
+	timodel.ActionAddColumns,
+	timodel.ActionDropColumns,
+}
+
+// isAlterTable returns true if the given job type is alter table's subtype.
+func isAlterTable(jobType timodel.ActionType) bool {
+	for _, t := range alterTableSubType {
+		if t == jobType {
+			return true
+		}
 	}
-	return et, nil
+	return false
 }
 
 // SupportedEventTypes returns the supported event types.
 func SupportedEventTypes() []bf.EventType {
+	supportedEventTypes := []bf.EventType{
+		bf.AllDML,
+		bf.AllDDL,
+
+		// dml events
+		bf.InsertEvent,
+		bf.UpdateEvent,
+		bf.DeleteEvent,
+
+		// ddl events
+		bf.AlterTable,
+		bf.CreateSchema,
+		bf.DropSchema,
+	}
+
+	for _, ddlType := range ddlWhiteListMap {
+		supportedEventTypes = append(supportedEventTypes, ddlType)
+	}
 	return supportedEventTypes
 }
 


Test patch:
diff --git a/cdc/kv/client_test.go b/cdc/kv/client_test.go
index 5f048e262..399438984 100644
--- a/cdc/kv/client_test.go
+++ b/cdc/kv/client_test.go
@@ -1382,6 +1382,12 @@ func testStreamRecvWithError(t *testing.T, failpointStr string) {
 	cluster.AddStore(1, addr1)
 	cluster.Bootstrap(regionID, []uint64{1}, []uint64{4}, 4)
 
+	err = failpoint.Enable("github.com/pingcap/tiflow/cdc/kv/kvClientAddDuplicatedStream", failpointStr)
+	require.Nil(t, err)
+	defer func() {
+		_ = failpoint.Disable("github.com/pingcap/tiflow/cdc/kv/kvClientAddDuplicatedStream")
+	}()
+
 	err = failpoint.Enable("github.com/pingcap/tiflow/cdc/kv/kvClientStreamRecvError", failpointStr)
 	require.Nil(t, err)
 	defer func() {
diff --git a/cdc/kv/region_worker_test.go b/cdc/kv/region_worker_test.go
index 370b28f53..2bc0f3517 100644
--- a/cdc/kv/region_worker_test.go
+++ b/cdc/kv/region_worker_test.go
@@ -162,8 +162,9 @@ func TestRegionWokerHandleEventEntryEventOutOfOrder(t *testing.T) {
 	stream := &eventFeedStream{
 		storeID: 1,
 		id:      2,
+		regions: newSyncRegionFeedStateMap(),
 	}
-	worker := newRegionWorker(ctx, stream, s, newSyncRegionFeedStateMap())
+	worker := newRegionWorker(ctx, stream, s)
 	require.Equal(t, 2, cap(worker.outputCh))
 
 	// Receive prewrite2 with empty value.
@@ -324,13 +325,14 @@ func TestRegionWorkerHandleEventsBeforeStartTs(t *testing.T) {
 		&tikv.RPCContext{}),
 		0)
 	s1.sri.lockedRange = &regionlock.LockedRange{}
-	s1.sri.lockedRange.CheckpointTs.Store(9)
+	s1.sri.lockedRange.ResolvedTs.Store(9)
 	s1.start()
 	stream := &eventFeedStream{
 		storeID: 1,
 		id:      2,
+		regions: newSyncRegionFeedStateMap(),
 	}
-	w := newRegionWorker(ctx, stream, s, newSyncRegionFeedStateMap())
+	w := newRegionWorker(ctx, stream, s)
 
 	err := w.handleResolvedTs(ctx, &resolvedTsEvent{
 		resolvedTs: 5,
diff --git a/cdc/kv/regionlock/region_range_lock_test.go b/cdc/kv/regionlock/region_range_lock_test.go
index fc85bf687..78395639f 100644
--- a/cdc/kv/regionlock/region_range_lock_test.go
+++ b/cdc/kv/regionlock/region_range_lock_test.go
@@ -15,17 +15,20 @@ package regionlock
 
 import (
 	"context"
+	"encoding/binary"
+	"fmt"
 	"math"
 	"testing"
 	"time"
 
 	"github.com/pingcap/tiflow/cdc/processor/tablepb"
+	"github.com/pingcap/tiflow/pkg/spanz"
 	"github.com/stretchr/testify/require"
 )
 
-func mustSuccess(t *testing.T, res LockRangeResult, expectedCheckpointTs uint64) {
+func mustSuccess(t *testing.T, res LockRangeResult, expectedResolvedTs uint64) {
 	require.Equal(t, LockRangeStatusSuccess, res.Status)
-	require.Equal(t, expectedCheckpointTs, res.CheckpointTs)
+	require.Equal(t, expectedResolvedTs, res.ResolvedTs)
 }
 
 func mustStale(t *testing.T, res LockRangeResult, expectedRetryRanges ...tablepb.Span) {
@@ -43,10 +46,10 @@ func mustLockRangeSuccess(
 	t *testing.T,
 	l *RegionRangeLock,
 	startKey, endKey string,
-	regionID, version, expectedCheckpointTs uint64,
+	regionID, version, expectedResolvedTs uint64,
 ) {
 	res := l.LockRange(ctx, []byte(startKey), []byte(endKey), regionID, version)
-	mustSuccess(t, res, expectedCheckpointTs)
+	mustSuccess(t, res, expectedResolvedTs)
 }
 
 // nolint:unparam
@@ -82,8 +85,8 @@ func mustLockRangeWait(
 	return mustWaitFn(t, res)
 }
 
-func unlockRange(l *RegionRangeLock, startKey, endKey string, regionID, version uint64, checkpointTs uint64) {
-	l.UnlockRange([]byte(startKey), []byte(endKey), regionID, version, checkpointTs)
+func unlockRange(l *RegionRangeLock, startKey, endKey string, regionID, version uint64, resolvedTs uint64) {
+	l.UnlockRange([]byte(startKey), []byte(endKey), regionID, version, resolvedTs)
 }
 
 func TestRegionRangeLock(t *testing.T) {
@@ -157,7 +160,7 @@ func TestRegionRangeLockLockingRegionID(t *testing.T) {
 	unlockRange(l, "e", "f", 1, 11, 11)
 	res := <-ch
 	// CheckpointTS calculation should still be based on range and do not consider the regionID. So
-	// the result's checkpointTs should be 20 from of range ["g", "h"), instead of 11 from min(11, 20).
+	// the result's resolvedTs should be 20 from of range ["g", "h"), instead of 11 from min(11, 20).
 	mustSuccess(t, res, 20)
 	unlockRange(l, "g", "h", 1, 12, 30)
 }
@@ -174,20 +177,52 @@ func TestRegionRangeLockCanBeCancelled(t *testing.T) {
 	require.Equal(t, LockRangeStatusCancel, lockResult.Status)
 }
 
+func TestRangeTsMapSetUnset(t *testing.T) {
+	t.Parallel()
+
+	m := newRangeTsMap([]byte("a"), []byte("z"), 100)
+	require.Equal(t, 1, m.m.Len())
+	require.Equal(t, uint64(100), m.getMinTsInRange([]byte("a"), []byte("z")))
+
+	// Double set, should panic.
+	require.Panics(t, func() { m.clone().set([]byte("a"), []byte("m"), 101) })
+
+	// Unset and then get other ranges.
+	m.unset([]byte("m"), []byte("x"))
+	require.Equal(t, 3, m.m.Len())
+	require.Equal(t, uint64(100), m.getMinTsInRange([]byte("a"), []byte("m")))
+	require.Equal(t, uint64(100), m.getMinTsInRange([]byte("x"), []byte("z")))
+
+	// Unset and then get, should panic.
+	require.Panics(t, func() { m.clone().getMinTsInRange([]byte("m"), []byte("x")) })
+	require.Panics(t, func() { m.clone().getMinTsInRange([]byte("n"), []byte("w")) })
+	require.Panics(t, func() { m.clone().getMinTsInRange([]byte("a"), []byte("z")) })
+
+	// Unset all ranges.
+	m.unset([]byte("a"), []byte("m"))
+	m.unset([]byte("x"), []byte("z"))
+	require.Equal(t, 1, m.m.Len())
+	require.Panics(t, func() { m.clone().getMinTsInRange([]byte("a"), []byte("z")) })
+}
+
 func TestRangeTsMap(t *testing.T) {
 	t.Parallel()
 
 	m := newRangeTsMap([]byte("a"), []byte("z"), math.MaxUint64)
 
 	mustGetMin := func(startKey, endKey string, expectedTs uint64) {
-		ts := m.GetMin([]byte(startKey), []byte(endKey))
+		ts := m.getMinTsInRange([]byte(startKey), []byte(endKey))
 		require.Equal(t, expectedTs, ts)
 	}
 	set := func(startKey, endKey string, ts uint64) {
-		m.Set([]byte(startKey), []byte(endKey), ts)
+		m.set([]byte(startKey), []byte(endKey), ts)
+	}
+	unset := func(startKey, endKey string) {
+		m.unset([]byte(startKey), []byte(endKey))
 	}
 
 	mustGetMin("a", "z", math.MaxUint64)
+	unset("b", "e")
 	set("b", "e", 100)
 	mustGetMin("a", "z", 100)
 	mustGetMin("b", "e", 100)
@@ -197,12 +232,14 @@ func TestRangeTsMap(t *testing.T) {
 	mustGetMin("e", "f", math.MaxUint64)
 	mustGetMin("a", "b\x00", 100)
 
+	unset("d", "g")
 	set("d", "g", 80)
 	mustGetMin("d", "g", 80)
 	mustGetMin("a", "z", 80)
 	mustGetMin("d", "e", 80)
 	mustGetMin("a", "d", 100)
 
+	unset("c", "f")
 	set("c", "f", 120)
 	mustGetMin("c", "f", 120)
 	mustGetMin("c", "d", 120)
@@ -211,6 +248,7 @@ func TestRangeTsMap(t *testing.T) {
 	mustGetMin("b", "e", 100)
 	mustGetMin("a", "z", 80)
 
+	unset("c", "f")
 	set("c", "f", 130)
 	mustGetMin("c", "f", 130)
 	mustGetMin("c", "d", 130)
@@ -228,25 +266,75 @@ func TestRegionRangeLockCollect(t *testing.T) {
 	attrs := l.CollectLockedRangeAttrs(nil)
 	require.Equal(t, 1, len(attrs.Holes))
 
-	l.LockRange(ctx, []byte("a"), []byte("m"), 1, 1).LockedRange.CheckpointTs.Add(1)
+	l.LockRange(ctx, []byte("a"), []byte("m"), 1, 1).LockedRange.ResolvedTs.Add(1)
 	attrs = l.CollectLockedRangeAttrs(nil)
 	require.Equal(t, 1, len(attrs.Holes))
-	require.Equal(t, uint64(101), attrs.SlowestRegion.CheckpointTs)
-	require.Equal(t, uint64(101), attrs.FastestRegion.CheckpointTs)
+	require.Equal(t, uint64(101), attrs.SlowestRegion.ResolvedTs)
+	require.Equal(t, uint64(101), attrs.FastestRegion.ResolvedTs)
 
-	l.LockRange(ctx, []byte("m"), []byte("z"), 2, 1).LockedRange.CheckpointTs.Add(2)
+	l.LockRange(ctx, []byte("m"), []byte("z"), 2, 1).LockedRange.ResolvedTs.Add(2)
 	attrs = l.CollectLockedRangeAttrs(nil)
 	require.Equal(t, 0, len(attrs.Holes))
-	require.Equal(t, uint64(101), attrs.SlowestRegion.CheckpointTs)
-	require.Equal(t, uint64(102), attrs.FastestRegion.CheckpointTs)
+	require.Equal(t, uint64(101), attrs.SlowestRegion.ResolvedTs)
+	require.Equal(t, uint64(102), attrs.FastestRegion.ResolvedTs)
 
 	l.UnlockRange([]byte("a"), []byte("m"), 1, 1)
 	attrs = l.CollectLockedRangeAttrs(nil)
 	require.Equal(t, 1, len(attrs.Holes))
-	require.Equal(t, uint64(102), attrs.SlowestRegion.CheckpointTs)
-	require.Equal(t, uint64(102), attrs.FastestRegion.CheckpointTs)
+	require.Equal(t, uint64(102), attrs.SlowestRegion.ResolvedTs)
+	require.Equal(t, uint64(102), attrs.FastestRegion.ResolvedTs)
 
 	l.UnlockRange([]byte("m"), []byte("z"), 2, 1)
 	attrs = l.CollectLockedRangeAttrs(nil)
 	require.Equal(t, 1, len(attrs.Holes))
 }
+
+func TestCalculateMinResolvedTs(t *testing.T) {
+	l := NewRegionRangeLock(1, []byte("a"), []byte("z"), 100, "")
+
+	res := l.LockRange(context.Background(), []byte("m"), []byte("x"), 1, 1)
+	res.LockedRange.ResolvedTs.Store(101)
+	require.Equal(t, LockRangeStatusSuccess, res.Status)
+	require.Equal(t, uint64(100), l.CalculateMinResolvedTs())
+
+	res = l.LockRange(context.Background(), []byte("a"), []byte("m"), 2, 1)
+	require.Equal(t, LockRangeStatusSuccess, res.Status)
+	res.LockedRange.ResolvedTs.Store(102)
+	res = l.LockRange(context.Background(), []byte("x"), []byte("z"), 3, 1)
+	require.Equal(t, LockRangeStatusSuccess, res.Status)
+	res.LockedRange.ResolvedTs.Store(103)
+	require.Equal(t, uint64(101), l.CalculateMinResolvedTs())
+}
+
+func BenchmarkOneMillionRegions(b *testing.B) {
+	ctx := context.Background()
+	startKey, endKey := spanz.GetTableRange(1)
+	l := NewRegionRangeLock(1, startKey, endKey, 100, "")
+
+	for i := 1; i <= 1000*1000; i++ {
+		var rangeStart, rangeEnd []byte
+
+		rangeStart = make([]byte, len(startKey)+8)
+		copy(rangeStart, startKey)
+		binary.BigEndian.PutUint64(rangeStart[len(startKey):], uint64(i))
+
+		if i < 1000*1000 {
+			rangeEnd = make([]byte, len(startKey)+8)
+			copy(rangeEnd, startKey)
+			binary.BigEndian.PutUint64(rangeEnd[len(startKey):], uint64(i+1))
+		} else {
+			rangeEnd = make([]byte, len(endKey))
+			copy(rangeEnd, endKey)
+		}
+		lockRes := l.LockRange(ctx, rangeStart, rangeEnd, uint64(i), 1)
+		if lockRes.Status != LockRangeStatusSuccess {
+			panic(fmt.Sprintf("bad lock range, i: %d\n", i))
+		}
+		lockRes.LockedRange.ResolvedTs.Store(uint64(100 + i))
+	}
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		l.CalculateMinResolvedTs()
+	}
+}
diff --git a/cdc/kv/shared_client_test.go b/cdc/kv/shared_client_test.go
index cff7d3d32..1c6bfcc5b 100644
--- a/cdc/kv/shared_client_test.go
+++ b/cdc/kv/shared_client_test.go
@@ -132,8 +132,12 @@ func TestConnectToOfflineOrFailedTiKV(t *testing.T) {
 	client := NewSharedClient(
 		model.ChangeFeedID{ID: "test"},
 		&config.ServerConfig{
-			KVClient: &config.KVClientConfig{WorkerConcurrent: 1, GrpcStreamConcurrent: 1},
-			Debug:    &config.DebugConfig{Puller: &config.PullerConfig{LogRegionDetails: false}},
+			KVClient: &config.KVClientConfig{
+				WorkerConcurrent:     1,
+				GrpcStreamConcurrent: 1,
+				AdvanceIntervalInMs:  10,
+			},
+			Debug: &config.DebugConfig{Puller: &config.PullerConfig{LogRegionDetails: false}},
 		},
 		false, pdClient, grpcPool, regionCache, pdClock, lockResolver,
 	)
diff --git a/cdc/kv/shared_region_worker_test.go b/cdc/kv/shared_region_worker_test.go
index 0b11de7ab..fa9059e7f 100644
--- a/cdc/kv/shared_region_worker_test.go
+++ b/cdc/kv/shared_region_worker_test.go
@@ -30,7 +30,17 @@ import (
 
 func newSharedClientForTestSharedRegionWorker() *SharedClient {
 	// sharedRegionWorker only requires `SharedClient.onRegionFail`.
-	cfg := &config.ServerConfig{Debug: &config.DebugConfig{Puller: &config.PullerConfig{LogRegionDetails: false}}}
+	cfg := &config.ServerConfig{
+		KVClient: &config.KVClientConfig{
+			EnableMultiplexing:   true,
+			WorkerConcurrent:     8,
+			GrpcStreamConcurrent: 2,
+			AdvanceIntervalInMs:  10,
+		},
+		Debug: &config.DebugConfig{
+			Puller: &config.PullerConfig{LogRegionDetails: false},
+		},
+	}
 	return NewSharedClient(model.ChangeFeedID{}, cfg, false, nil, nil, nil, nil, nil)
 }
 
@@ -183,14 +193,4 @@ func TestSharedRegionWorkerHandleResolvedTs(t *testing.T) {
 	require.Equal(t, uint64(10), s1.getLastResolvedTs())
 	require.Equal(t, uint64(11), s2.getLastResolvedTs())
 	require.Equal(t, uint64(8), s3.getLastResolvedTs())
-
-	select {
-	case event := <-eventCh:
-		require.Equal(t, uint64(0), event.RegionID)
-		require.Equal(t, uint64(10), event.Resolved.ResolvedTs)
-		require.Equal(t, 1, len(event.Resolved.Spans))
-		require.Equal(t, uint64(1), event.Resolved.Spans[0].Region)
-	case <-time.NewTimer(100 * time.Millisecond).C:
-		require.True(t, false, "must get an event")
-	}
 }
diff --git a/cdc/puller/frontier/frontier_test.go b/cdc/puller/frontier/frontier_test.go
index e19d24f31..445761c3b 100644
--- a/cdc/puller/frontier/frontier_test.go
+++ b/cdc/puller/frontier/frontier_test.go
@@ -522,7 +522,7 @@ func TestRandomMergeAndSplit(t *testing.T) {
 			selected := rand.Intn(totalLockedRanges)
 			count := 0
 			rangelock.CollectLockedRangeAttrs(func(regionID, version uint64, state *regionlock.LockedRange, span tablepb.Span) {
-				ts := state.CheckpointTs.Load()
+				ts := state.ResolvedTs.Load()
 				startKey := span.StartKey
 				endKey := span.EndKey
 				if count == selected {
@@ -547,7 +547,7 @@ func TestRandomMergeAndSplit(t *testing.T) {
 			selected := rand.Intn(totalLockedRanges - 1)
 			count := 0
 			rangelock.CollectLockedRangeAttrs(func(regionID, version uint64, state *regionlock.LockedRange, span tablepb.Span) {
-				ts := state.CheckpointTs.Load()
+				ts := state.ResolvedTs.Load()
 				startKey := span.StartKey
 				endKey := span.EndKey
 				if count == selected {
diff --git a/dm/tests/incompatible_ddl_changes/conf/dm-task1.yaml b/dm/tests/incompatible_ddl_changes/conf/dm-task1.yaml
index f2de75176..f35febfcb 100644
--- a/dm/tests/incompatible_ddl_changes/conf/dm-task1.yaml
+++ b/dm/tests/incompatible_ddl_changes/conf/dm-task1.yaml
@@ -40,7 +40,7 @@ filters:
         "drop primary key",
         "drop unique key",
         "modify default value",
-        "modify constaints",
+        "modify constraint",
         "modify columns order",
         "modify charset",
         "modify collation",
diff --git a/dm/tests/incompatible_ddl_changes/run.sh b/dm/tests/incompatible_ddl_changes/run.sh
index 90710ff44..882455a43 100755
--- a/dm/tests/incompatible_ddl_changes/run.sh
+++ b/dm/tests/incompatible_ddl_changes/run.sh
@@ -123,11 +123,11 @@ function incompatible_ddl() {
 		"binlog skip test" \
 		"\"result\": true" 2
 
-	# modify constaints
+	# modify constraint
 	run_sql_source1 "alter table incompatible_ddl_changes.t1 add constraint c_int_unique unique(c_int)"
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"query-status test" \
-		"event modify constaint" 1
+		"event modify constraint" 1
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"binlog skip test" \
 		"\"result\": true" 2
@@ -238,7 +238,7 @@ function incompatible_ddl() {
 	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
+		"event modify constraint" 1
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"binlog skip test" \
 		"\"result\": true" 2
diff --git a/pkg/cmd/server/server_test.go b/pkg/cmd/server/server_test.go
index 723920efb..0311007c5 100644
--- a/pkg/cmd/server/server_test.go
+++ b/pkg/cmd/server/server_test.go
@@ -164,6 +164,7 @@ func TestParseCfg(t *testing.T) {
 			EnableMultiplexing:   true,
 			WorkerConcurrent:     8,
 			GrpcStreamConcurrent: 1,
+			AdvanceIntervalInMs:  300,
 			FrontierConcurrent:   8,
 			WorkerPoolSize:       0,
 			RegionScanLimit:      40,
@@ -312,6 +313,7 @@ check-balance-interval = "10s"
 			EnableMultiplexing:   true,
 			WorkerConcurrent:     8,
 			GrpcStreamConcurrent: 1,
+			AdvanceIntervalInMs:  300,
 			FrontierConcurrent:   8,
 			WorkerPoolSize:       0,
 			RegionScanLimit:      40,
@@ -450,6 +452,7 @@ cert-allowed-cn = ["dd","ee"]
 			EnableMultiplexing:   true,
 			WorkerConcurrent:     8,
 			GrpcStreamConcurrent: 1,
+			AdvanceIntervalInMs:  300,
 			FrontierConcurrent:   8,
 			WorkerPoolSize:       0,
 			RegionScanLimit:      40,
diff --git a/pkg/config/config_test_data.go b/pkg/config/config_test_data.go
index 3e26479f1..253c75146 100644
--- a/pkg/config/config_test_data.go
+++ b/pkg/config/config_test_data.go
@@ -142,6 +142,7 @@ const (
     "enable-multiplexing": true,
     "worker-concurrent": 8,
     "grpc-stream-concurrent": 1,
+    "advance-interval-in-ms": 300,
     "frontier-concurrent": 8,
     "worker-pool-size": 0,
     "region-scan-limit": 40,
@@ -176,7 +177,6 @@ const (
       "check-balance-interval": 60000000000,
       "add-table-batch-size": 50
     },
-    "enable-kv-connect-backoff": false,
     "cdc-v2": {
       "enable": false,
       "meta-store": {
diff --git a/pkg/filter/filter_test.go b/pkg/filter/filter_test.go
index 97789821b..43f1b2335 100644
--- a/pkg/filter/filter_test.go
+++ b/pkg/filter/filter_test.go
@@ -33,7 +33,7 @@ func TestShouldUseDefaultRules(t *testing.T) {
 	require.True(t, filter.ShouldIgnoreTable("performance_schema", ""))
 	require.False(t, filter.ShouldIgnoreTable("metric_schema", "query_duration"))
 	require.False(t, filter.ShouldIgnoreTable("sns", "user"))
-	require.False(t, filter.ShouldIgnoreTable("tidb_cdc", "repl_mark_a_a"))
+	require.True(t, filter.ShouldIgnoreTable("tidb_cdc", "repl_mark_a_a"))
 }
 
 func TestShouldUseCustomRules(t *testing.T) {
@@ -371,13 +371,14 @@ func TestShouldDiscardDDL(t *testing.T) {
 }
 
 func TestIsAllowedDDL(t *testing.T) {
+	require.Len(t, ddlWhiteListMap, 36)
 	type testCase struct {
 		timodel.ActionType
 		allowed bool
 	}
-	testCases := make([]testCase, 0, len(allowDDLList))
-	for _, action := range allowDDLList {
-		testCases = append(testCases, testCase{action, true})
+	testCases := make([]testCase, 0, len(ddlWhiteListMap))
+	for ddlType := range ddlWhiteListMap {
+		testCases = append(testCases, testCase{ddlType, true})
 	}
 	testCases = append(testCases, testCase{timodel.ActionAddForeignKey, false})
 	testCases = append(testCases, testCase{timodel.ActionDropForeignKey, false})
diff --git a/pkg/filter/sql_event_filter_test.go b/pkg/filter/sql_event_filter_test.go
index 27cba219e..87571d0aa 100644
--- a/pkg/filter/sql_event_filter_test.go
+++ b/pkg/filter/sql_event_filter_test.go
@@ -18,6 +18,7 @@ import (
 
 	"github.com/pingcap/errors"
 	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
+	timodel "github.com/pingcap/tidb/pkg/parser/model"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
@@ -27,10 +28,11 @@ import (
 func TestShouldSkipDDL(t *testing.T) {
 	t.Parallel()
 	type innerCase struct {
-		schema string
-		table  string
-		query  string
-		skip   bool
+		schema  string
+		table   string
+		query   string
+		ddlType timodel.ActionType
+		skip    bool
 	}
 
 	type testCase struct {
@@ -51,28 +53,32 @@ func TestShouldSkipDDL(t *testing.T) {
 			},
 			cases: []innerCase{
 				{
-					schema: "test",
-					table:  "t1",
-					query:  "alter table t1 modify column age int",
-					skip:   true,
+					schema:  "test",
+					table:   "t1",
+					query:   "alter table t1 modify column age int",
+					ddlType: timodel.ActionModifyColumn,
+					skip:    true,
 				},
 				{
-					schema: "test",
-					table:  "t1",
-					query:  "create table t1(id int primary key)",
-					skip:   true,
+					schema:  "test",
+					table:   "t1",
+					query:   "create table t1(id int primary key)",
+					ddlType: timodel.ActionCreateTable,
+					skip:    true,
 				},
 				{
-					schema: "test",
-					table:  "t2", // table name not match
-					query:  "alter table t2 modify column age int",
-					skip:   false,
+					schema:  "test",
+					table:   "t2", // table name not match
+					query:   "alter table t2 modify column age int",
+					ddlType: timodel.ActionModifyColumn,
+					skip:    false,
 				},
 				{
-					schema: "test2", // schema name not match
-					table:  "t1",
-					query:  "alter table t1 modify column age int",
-					skip:   false,
+					schema:  "test2", // schema name not match
+					table:   "t1",
+					query:   "alter table t1 modify column age int",
+					ddlType: timodel.ActionModifyColumn,
+					skip:    false,
 				},
 			},
 		},
@@ -87,28 +93,32 @@ func TestShouldSkipDDL(t *testing.T) {
 			},
 			cases: []innerCase{
 				{
-					schema: "test",
-					table:  "t1",
-					query:  "alter table t1 modify column age int",
-					skip:   false,
+					schema:  "test",
+					table:   "t1",
+					query:   "alter table t1 modify column age int",
+					ddlType: timodel.ActionModifyColumn,
+					skip:    false,
 				},
 				{
-					schema: "test",
-					table:  "t1",
-					query:  "alter table t1 drop column age",
-					skip:   false,
+					schema:  "test",
+					table:   "t1",
+					query:   "alter table t1 drop column age",
+					ddlType: timodel.ActionDropColumn,
+					skip:    false,
 				},
 				{
-					schema: "test2",
-					table:  "t1",
-					query:  "drop database test2",
-					skip:   true,
+					schema:  "test2",
+					table:   "t1",
+					query:   "drop database test2",
+					ddlType: timodel.ActionDropSchema,
+					skip:    true,
 				},
 				{
-					schema: "test3",
-					table:  "t1",
-					query:  "drop index i3 on t1",
-					skip:   false,
+					schema:  "test3",
+					table:   "t1",
+					query:   "drop index i3 on t1",
+					ddlType: timodel.ActionDropIndex,
+					skip:    false,
 				},
 			},
 		},
@@ -123,27 +133,31 @@ func TestShouldSkipDDL(t *testing.T) {
 			},
 			cases: []innerCase{
 				{
-					schema: "test",
-					table:  "t1",
-					query:  "ALTER TABLE t1 MODIFY COLUMN age int(11) NOT NULL",
-					skip:   true,
+					schema:  "test",
+					table:   "t1",
+					query:   "ALTER TABLE t1 MODIFY COLUMN age int(11) NOT NULL",
+					ddlType: timodel.ActionModifyColumn,
+					skip:    true,
 				},
 				{
-					schema: "test",
-					table:  "t1",
-					query:  "ALTER TABLE t1 DROP COLUMN age",
-					skip:   true,
+					schema:  "test",
+					table:   "t1",
+					query:   "ALTER TABLE t1 DROP COLUMN age",
+					ddlType: timodel.ActionDropColumn,
+					skip:    true,
 				},
 				{ // no table name
-					schema: "test2",
-					query:  "DROP DATABASE test",
-					skip:   true,
+					schema:  "test2",
+					query:   "DROP DATABASE test",
+					ddlType: timodel.ActionDropSchema,
+					skip:    true,
 				},
 				{
-					schema: "test3",
-					table:  "t1",
-					query:  "Drop Index i1 on test3.t1",
-					skip:   false,
+					schema:  "test3",
+					table:   "t1",
+					query:   "Drop Index i1 on test3.t1",
+					ddlType: timodel.ActionDropIndex,
+					skip:    false,
 				},
 			},
 		},
@@ -171,7 +185,7 @@ func TestShouldSkipDDL(t *testing.T) {
 	}
 
 	for _, tc := range testCases {
-		f, err := newSQLEventFilter(tc.cfg, config.GetDefaultReplicaConfig().SQLMode)
+		f, err := newSQLEventFilter(tc.cfg)
 		require.True(t, errors.ErrorEqual(err, tc.err), "case: %+s", err)
 		for _, c := range tc.cases {
 			ddl := &model.DDLEvent{
@@ -182,6 +196,7 @@ func TestShouldSkipDDL(t *testing.T) {
 					},
 				},
 				Query: c.query,
+				Type:  c.ddlType,
 			}
 			skip, err := f.shouldSkipDDL(ddl)
 			require.NoError(t, err)
@@ -297,7 +312,7 @@ func TestShouldSkipDML(t *testing.T) {
 		tc := tc
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
-			f, err := newSQLEventFilter(tc.cfg, config.GetDefaultReplicaConfig().SQLMode)
+			f, err := newSQLEventFilter(tc.cfg)
 			require.NoError(t, err)
 			for _, c := range tc.cases {
 				event := &model.RowChangedEvent{
@@ -329,8 +344,8 @@ func TestVerifyIgnoreEvents(t *testing.T) {
 		err         error
 	}
 
-	cases := make([]testCase, len(supportedEventTypes))
-	for i, eventType := range supportedEventTypes {
+	cases := make([]testCase, len(SupportedEventTypes()))
+	for i, eventType := range SupportedEventTypes() {
 		cases[i] = testCase{
 			ignoreEvent: []bf.EventType{eventType},
 			err:         nil,
diff --git a/pkg/filter/utils_test.go b/pkg/filter/utils_test.go
index 9d671a5c8..b3f926d52 100644
--- a/pkg/filter/utils_test.go
+++ b/pkg/filter/utils_test.go
@@ -14,17 +14,10 @@
 package filter
 
 import (
-	"fmt"
 	"testing"
 
-	"github.com/pingcap/log"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
-	"github.com/pingcap/tidb/pkg/parser"
-	timodel "github.com/pingcap/tidb/pkg/parser/model"
-	"github.com/pingcap/tidb/pkg/parser/mysql"
 	tifilter "github.com/pingcap/tidb/pkg/util/filter"
 	"github.com/pingcap/tiflow/pkg/config"
-	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/stretchr/testify/require"
 )
 
@@ -42,17 +35,13 @@ func TestIsSchema(t *testing.T) {
 		{tifilter.InspectionSchemaName, true},
 		{tifilter.PerformanceSchemaName, true},
 		{tifilter.MetricSchemaName, true},
+		{TiCDCSystemSchema, true},
 	}
 	for _, c := range cases {
 		require.Equal(t, c.result, isSysSchema(c.schema))
 	}
 }
 
-func TestSupportedEventTypeString(t *testing.T) {
-	t.Parallel()
-	require.Equal(t, supportedEventTypes, SupportedEventTypes())
-}
-
 func TestVerifyTableRules(t *testing.T) {
 	t.Parallel()
 	cases := []struct {
@@ -71,166 +60,3 @@ func TestVerifyTableRules(t *testing.T) {
 		require.Equal(t, c.hasError, err != nil, "case: %s", c.cfg.Rules)
 	}
 }
-
-func TestDDLToEventType(t *testing.T) {
-	t.Parallel()
-	cases := []struct {
-		ddl       string
-		jobType   timodel.ActionType
-		eventType bf.EventType
-		err       error
-	}{
-		{"CREATE DATABASE test", timodel.ActionCreateSchema, bf.CreateDatabase, nil},
-		{"DROP DATABASE test", timodel.ActionDropSchema, bf.DropDatabase, nil},
-		{"CREATE TABLE test.t1(id int primary key)", timodel.ActionCreateTable, bf.CreateTable, nil},
-		{"DROP TABLE test.t1", timodel.ActionDropTable, bf.DropTable, nil},
-		{"TRUNCATE TABLE test.t1", timodel.ActionTruncateTable, bf.TruncateTable, nil},
-		{"rename table s1.t1 to s2.t2", timodel.ActionRenameTable, bf.RenameTable, nil},
-		{"rename table s1.t1 to s2.t2, test.t1 to test.t2", timodel.ActionRenameTables, bf.RenameTable, nil},
-		{"create index i1 on test.t1 (age)", timodel.ActionAddIndex, bf.AlterTable, nil},
-		{"drop index i1 on test.t1", timodel.ActionDropIndex, bf.AlterTable, nil},
-		{"CREATE VIEW test.v AS SELECT * FROM t", timodel.ActionCreateView, bf.CreateView, nil},
-		{"DROP view if exists test.v", timodel.ActionDropView, bf.DropView, nil},
-
-		{"alter table test.t1 add column name varchar(50)", timodel.ActionAddColumn, bf.AlterTable, nil},
-		{"alter table test.t1 drop column name", timodel.ActionDropColumn, bf.AlterTable, nil},
-		{"alter table test.t1 modify column name varchar(100)", timodel.ActionModifyColumn, bf.AlterTable, nil},
-		{"ALTER TABLE test.t1 CONVERT TO CHARACTER SET gbk", timodel.ActionModifyTableCharsetAndCollate, bf.AlterTable, nil},
-		{"alter table test add primary key(b)", timodel.ActionAddIndex, bf.AlterTable, nil},
-		{"ALTER DATABASE dbname CHARACTER SET utf8 COLLATE utf8_general_ci;", timodel.ActionModifySchemaCharsetAndCollate, bf.AlterDatabase, nil},
-		{"Alter table test.t1 drop partition t11", timodel.ActionDropTablePartition, bf.DropTablePartition, nil},
-		{"Alter table test.t1 add partition (partition p3 values less than (2002))", timodel.ActionDropTablePartition, bf.DropTablePartition, nil},
-		{"Alter table test.t1 truncate partition t11", timodel.ActionDropTablePartition, bf.DropTablePartition, nil},
-		{"Alter table test.t1 reorganize partition p11 into (partition p1 values less than (10), partition p2 values less than (20))", timodel.ActionReorganizePartition, bf.AlterTable, nil},
-		{"alter table add i", timodel.ActionAddIndex, bf.NullEvent, cerror.ErrConvertDDLToEventTypeFailed},
-	}
-	p := parser.New()
-	for _, c := range cases {
-		et, err := ddlToEventType(p, c.ddl, c.jobType)
-		if c.err != nil {
-			errRFC, ok := cerror.RFCCode(err)
-			require.True(t, ok)
-			caseErrRFC, ok := cerror.RFCCode(c.err)
-			require.True(t, ok)
-			require.Equal(t, caseErrRFC, errRFC)
-		} else {
-			require.NoError(t, err)
-		}
-		require.Equal(t, c.eventType, et, "case%v", c.ddl)
-	}
-}
-
-func TestDDLToTypeSpecialDDL(t *testing.T) {
-	type c struct {
-		ddl      string
-		jobType  timodel.ActionType
-		evenType bf.EventType
-		err      error
-	}
-
-	ddlWithTab := `CREATE TABLE if not exists sbtest25 
-	(
-		id bigint NOT NULL,
-		k bigint NOT NULL DEFAULT '0',
-		c char(30) NOT NULL DEFAULT '',
-		pad char(20) NOT NULL DEFAULT '',
-		PRIMARY KEY (id),
-	    KEY k_1 (k)
-	) 	ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin`
-	ddlWithTwoTab := `		CREATE TABLE if not exists sbtest25 
-	(
-		id bigint NOT NULL,
-		k bigint NOT NULL DEFAULT '0',
-		c char(30) NOT NULL DEFAULT '',
-		pad char(20) NOT NULL DEFAULT '',
-		PRIMARY KEY (id),
-		KEY k_1 (k)
-		)
-		ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin`
-	ddlWithNewLine := `CREATE TABLE finish_mark 
-	(
-		
-		id INT AUTO_INCREMENT PRIMARY KEY,
-		val INT DEFAULT 0,                     
-		col0 INT NOT NULL)`
-
-	cases := []c{
-		{"CREATE DATABASE test", timodel.ActionCreateSchema, bf.CreateDatabase, nil},
-		{ddlWithTwoTab, timodel.ActionCreateTable, bf.CreateTable, nil},
-		{ddlWithTab, timodel.ActionCreateTable, bf.CreateTable, nil},
-		{ddlWithNewLine, timodel.ActionCreateTable, bf.CreateTable, nil},
-	}
-	p := parser.New()
-	for _, c := range cases {
-		log.Info(c.ddl)
-		et, err := ddlToEventType(p, c.ddl, c.jobType)
-		if c.err != nil {
-			errRFC, ok := cerror.RFCCode(err)
-			require.True(t, ok)
-			caseErrRFC, ok := cerror.RFCCode(c.err)
-			require.True(t, ok)
-			require.Equal(t, caseErrRFC, errRFC)
-		} else {
-			require.NoError(t, err)
-		}
-		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/scripts/download-integration-test-binaries.sh b/scripts/download-integration-test-binaries.sh
index f19aad8d6..f90f182d5 100755
--- a/scripts/download-integration-test-binaries.sh
+++ b/scripts/download-integration-test-binaries.sh
@@ -120,8 +120,7 @@ function download_binaries() {
 	# tidb_sha1=$(curl "${file_server_url}/download/refs/pingcap/tidb/${branch}/sha1")
 	tidb_sha1=ff8b3d356d604db05d8ca18e93d2c6b81db9c449
 	tikv_sha1=$(curl "${file_server_url}/download/refs/pingcap/tikv/${branch}/sha1")
-	# pd_sha1=$(curl "${file_server_url}/download/refs/pingcap/pd/v7.5.0/sha1")
-	pd_sha1=fb6606aa3f41cc5dc4b804d4715e1613d30835bf
+	pd_sha1=$(curl "${file_server_url}/download/refs/pingcap/pd/${branch}/sha1")
 	tiflash_sha1=$(curl "${file_server_url}/download/refs/pingcap/tiflash/${branch}/sha1")
 
 	# All download links.
diff --git a/tests/integration_tests/bdr_mode/conf/cf.toml b/tests/integration_tests/bdr_mode/conf/cf.toml
deleted file mode 100644
index b6929758e..000000000
--- a/tests/integration_tests/bdr_mode/conf/cf.toml
+++ /dev/null
@@ -1,1 +0,0 @@
-bdr-mode = true
diff --git a/tests/integration_tests/bdr_mode/conf/down.toml b/tests/integration_tests/bdr_mode/conf/down.toml
new file mode 100644
index 000000000..6b23a8694
--- /dev/null
+++ b/tests/integration_tests/bdr_mode/conf/down.toml
@@ -0,0 +1,1 @@
+bdr-mode = true
\ No newline at end of file
diff --git a/tests/integration_tests/bdr_mode/conf/up.toml b/tests/integration_tests/bdr_mode/conf/up.toml
new file mode 100644
index 000000000..ac2e6c7cf
--- /dev/null
+++ b/tests/integration_tests/bdr_mode/conf/up.toml
@@ -0,0 +1,3 @@
+bdr-mode = true
+enable-sync-point = true
+sync-point-interval = "30s"
\ No newline at end of file
diff --git a/tests/integration_tests/bdr_mode/run.sh b/tests/integration_tests/bdr_mode/run.sh
index 2b30e8c64..dd1167f61 100644
--- a/tests/integration_tests/bdr_mode/run.sh
+++ b/tests/integration_tests/bdr_mode/run.sh
@@ -30,15 +30,19 @@ function run() {
 	SINK_URI_2="mysql://root@127.0.0.1:4000"
 
 	# up -> down
-	run_cdc_cli changefeed create --sink-uri="$SINK_URI_1" -c "test-1" --config="$CUR/conf/cf.toml"
+	run_cdc_cli changefeed create --sink-uri="$SINK_URI_1" -c "test-1" --config="$CUR/conf/up.toml"
 	# down -> up
-	run_cdc_cli changefeed create --sink-uri="$SINK_URI_2" -c "test-2" --server "http://127.0.0.1:8400" --config="$CUR/conf/cf.toml"
+	run_cdc_cli changefeed create --sink-uri="$SINK_URI_2" -c "test-2" --server "http://127.0.0.1:8400" --config="$CUR/conf/down.toml"
 
 	run_sql_file $CUR/data/up.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 	run_sql_file $CUR/data/down.sql ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 
 	run_sql_file $CUR/data/finished.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 
+	# syncpoint table should exists in secondary tidb, but does not exists in primary cluster
+	check_table_exists "tidb_cdc.syncpoint_v1" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 60
+	check_table_not_exists "tidb_cdc.syncpoint_v1" ${UP_TIDB_HOST} ${UP_TIDB_PORT} 60
+
 	check_table_exists "bdr_mode.finish_mark" ${UP_TIDB_HOST} ${UP_TIDB_PORT} 60
 
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml