Repository: pingcap/tiflow
Pull request #10368 (2023-12-27 05:24:22)

conflict detector(cdc): fix potential dead lock when a transaction generated multiple same hash (#10335)
This is an automated cherry-pick of #10335

<!--
Thank you for contributing to TiFlow! 
Please read MD's [CONTRIBUTING](https://github.com/pingcap/tiflow/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.
-->

### What problem does this PR solve?
<!--
Please create an issue first to describe the problem.

There MUST be one line starting with "Issue Number:  " and 
linking the relevant issues via the "close" or "ref".

For more info, check https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue.
 -->

Issue Number: close https://github.com/pingcap/tiflow/issues/10334

### What is changed and how it works?

- Deduplicate the generated hash in a single transaction to prevent self cyclic dependency in the conflict detector.
- Refine the conflict detector, make it more readable


### Check List <!--REMOVE the items that are not applicable-->

#### Tests <!-- At least one of them must be included. -->

 - Unit test

#### Questions <!-- Authors should answer these questions and reviewers should consider these questions. -->

##### Will it cause performance regression or break compatibility?

##### Do you need to update user documentation, design documentation or monitoring documentation?

### Release note <!-- bugfixes or new features need a release note -->

```release-note
Fix sink-to-mysql checkpoint-ts stuck issue in some rare cases.
```


ti-chi-bot[bot]: This cherry pick PR is for a release branch and has not yet been approved by triage owners. 
Adding the `do-not-merge/cherry-pick-not-approved` label.

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


<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>
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 assign [amyangfei](https://github.com/amyangfei) for approval. For more information see [the Kubernetes Code Review Process](https://book.prow.tidb.net/#/workflows/pr).

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

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

- **[OWNERS](https://github.com/pingcap/tiflow/blob/release-7.1/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":["amyangfei"]} -->
ti-chi-bot[bot]: @ti-chi-bot: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

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

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

<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]: @ti-chi-bot: 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-dm-integration-test | d94090292afed4af97b635c926bc280ad876e04e | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/release-7.1/job/pull_dm_integration_test/1/display/redirect) | true | `/test dm-integration-test`
pull-cdc-integration-mysql-test | d94090292afed4af97b635c926bc280ad876e04e | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/release-7.1/job/pull_cdc_integration_test/1/display/redirect) | true | `/test cdc-integration-mysql-test`

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

<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/api/v2/api.go b/cdc/api/v2/api.go
index d00a0a577..40779b58d 100644
--- a/cdc/api/v2/api.go
+++ b/cdc/api/v2/api.go
@@ -59,6 +59,7 @@ func RegisterOpenAPIV2Routes(router *gin.Engine, api OpenAPIV2) {
 	changefeedGroup.POST("/:changefeed_id/resume", api.resumeChangefeed)
 	changefeedGroup.POST("/:changefeed_id/pause", api.pauseChangefeed)
 	changefeedGroup.GET("/:changefeed_id/status", api.status)
+	changefeedGroup.GET("/:changefeed_id/synced", api.synced)
 
 	// capture apis
 	captureGroup := v2.Group("/captures")
diff --git a/cdc/api/v2/api_helpers.go b/cdc/api/v2/api_helpers.go
index 1e74c97e3..86dd77703 100644
--- a/cdc/api/v2/api_helpers.go
+++ b/cdc/api/v2/api_helpers.go
@@ -88,7 +88,7 @@ type APIV2Helpers interface {
 		pdClient pd.Client,
 		gcServiceID string,
 		changefeedID model.ChangeFeedID,
-		checkpointTs uint64,
+		overrideCheckpointTs uint64,
 	) error
 
 	// getPDClient returns a PDClient given the PD cluster addresses and a credential
@@ -387,24 +387,24 @@ func (APIV2HelpersImpl) verifyUpdateChangefeedConfig(
 	return newInfo, newUpInfo, nil
 }
 
+// verifyResumeChangefeedConfig verifies the changefeed config before resuming a changefeed
+// overrideCheckpointTs is the checkpointTs of the changefeed that specified by the user.
+// or it is the checkpointTs of the changefeed before it is paused.
+// we need to check weather the resuming changefeed is gc safe or not.
 func (APIV2HelpersImpl) verifyResumeChangefeedConfig(ctx context.Context,
 	pdClient pd.Client,
 	gcServiceID string,
 	changefeedID model.ChangeFeedID,
-	checkpointTs uint64,
+	overrideCheckpointTs uint64,
 ) error {
-	if checkpointTs == 0 {
-		return nil
-	}
-
 	// 1h is enough for resuming a changefeed.
 	gcTTL := int64(60 * 60)
 	err := gc.EnsureChangefeedStartTsSafety(
 		ctx,
 		pdClient,
 		gcServiceID,
-		model.DefaultChangeFeedID(changefeedID.ID),
-		gcTTL, checkpointTs)
+		changefeedID,
+		gcTTL, overrideCheckpointTs)
 	if err != nil {
 		if !cerror.ErrStartTsBeforeGC.Equal(err) {
 			return cerror.ErrPDEtcdAPIError.Wrap(err)
diff --git a/cdc/api/v2/api_helpers_mock.go b/cdc/api/v2/api_helpers_mock.go
index 98dbd26ad..612a09d39 100644
--- a/cdc/api/v2/api_helpers_mock.go
+++ b/cdc/api/v2/api_helpers_mock.go
@@ -119,17 +119,17 @@ func (mr *MockAPIV2HelpersMockRecorder) verifyCreateChangefeedConfig(ctx, cfg, p
 }
 
 // verifyResumeChangefeedConfig mocks base method.
-func (m *MockAPIV2Helpers) verifyResumeChangefeedConfig(ctx context.Context, pdClient client.Client, gcServiceID string, changefeedID model.ChangeFeedID, checkpointTs uint64) error {
+func (m *MockAPIV2Helpers) verifyResumeChangefeedConfig(ctx context.Context, pdClient client.Client, gcServiceID string, changefeedID model.ChangeFeedID, overrideCheckpointTs uint64) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "verifyResumeChangefeedConfig", ctx, pdClient, gcServiceID, changefeedID, checkpointTs)
+	ret := m.ctrl.Call(m, "verifyResumeChangefeedConfig", ctx, pdClient, gcServiceID, changefeedID, overrideCheckpointTs)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
 // verifyResumeChangefeedConfig indicates an expected call of verifyResumeChangefeedConfig.
-func (mr *MockAPIV2HelpersMockRecorder) verifyResumeChangefeedConfig(ctx, pdClient, gcServiceID, changefeedID, checkpointTs interface{}) *gomock.Call {
+func (mr *MockAPIV2HelpersMockRecorder) verifyResumeChangefeedConfig(ctx, pdClient, gcServiceID, changefeedID, overrideCheckpointTs interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "verifyResumeChangefeedConfig", reflect.TypeOf((*MockAPIV2Helpers)(nil).verifyResumeChangefeedConfig), ctx, pdClient, gcServiceID, changefeedID, checkpointTs)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "verifyResumeChangefeedConfig", reflect.TypeOf((*MockAPIV2Helpers)(nil).verifyResumeChangefeedConfig), ctx, pdClient, gcServiceID, changefeedID, overrideCheckpointTs)
 }
 
 // verifyUpdateChangefeedConfig mocks base method.
diff --git a/cdc/api/v2/changefeed.go b/cdc/api/v2/changefeed.go
index 1f7f2ec66..989bf178b 100644
--- a/cdc/api/v2/changefeed.go
+++ b/cdc/api/v2/changefeed.go
@@ -44,6 +44,8 @@ const (
 	apiOpVarChangefeedState = "state"
 	// apiOpVarChangefeedID is the key of changefeed ID in HTTP API
 	apiOpVarChangefeedID = "changefeed_id"
+	// timeout for pd client
+	timeout = 30 * time.Second
 )
 
 // createChangefeed handles create changefeed request,
@@ -671,6 +673,11 @@ func (h *OpenAPIV2) resumeChangefeed(c *gin.Context) {
 		_ = c.Error(cerror.WrapError(cerror.ErrAPIInvalidParam, err))
 		return
 	}
+	status, err := h.capture.StatusProvider().GetChangeFeedStatus(ctx, changefeedID)
+	if err != nil {
+		_ = c.Error(err)
+		return
+	}
 
 	var pdClient pd.Client
 	// if PDAddrs is empty, use the default pdClient
@@ -692,13 +699,17 @@ func (h *OpenAPIV2) resumeChangefeed(c *gin.Context) {
 		}
 		defer pdClient.Close()
 	}
-
+	// If there is no overrideCheckpointTs, then check whether the currentCheckpointTs is smaller than gc safepoint or not.
+	newCheckpointTs := status.CheckpointTs
+	if cfg.OverwriteCheckpointTs != 0 {
+		newCheckpointTs = cfg.OverwriteCheckpointTs
+	}
 	if err := h.helpers.verifyResumeChangefeedConfig(
 		ctx,
 		pdClient,
 		h.capture.GetEtcdClient().GetEnsureGCServiceID(gc.EnsureGCServiceResuming),
 		changefeedID,
-		cfg.OverwriteCheckpointTs); err != nil {
+		newCheckpointTs); err != nil {
 		_ = c.Error(err)
 		return
 	}
@@ -726,9 +737,7 @@ func (h *OpenAPIV2) resumeChangefeed(c *gin.Context) {
 	}
 
 	if err := api.HandleOwnerJob(ctx, h.capture, job); err != nil {
-		if cfg.OverwriteCheckpointTs > 0 {
-			needRemoveGCSafePoint = true
-		}
+		needRemoveGCSafePoint = true
 		_ = c.Error(err)
 		return
 	}
@@ -826,6 +835,141 @@ func (h *OpenAPIV2) status(c *gin.Context) {
 	})
 }
 
+// synced get the synced status of a changefeed
+// @Summary Get synced status
+// @Description get the synced status of a changefeed
+// @Tags changefeed,v2
+// @Accept json
+// @Produce json
+// @Param changefeed_id  path  string  true  "changefeed_id"
+// @Param namespace query string false "default"
+// @Success 200 {object} SyncedStatus
+// @Failure 500,400 {object} model.HTTPError
+// @Router /api/v2/changefeeds/{changefeed_id}/synced [get]
+func (h *OpenAPIV2) synced(c *gin.Context) {
+	ctx := c.Request.Context()
+
+	changefeedID := model.DefaultChangeFeedID(c.Param(apiOpVarChangefeedID))
+	if err := model.ValidateChangefeedID(changefeedID.ID); err != nil {
+		_ = c.Error(cerror.ErrAPIInvalidParam.GenWithStack("invalid changefeed_id: %s",
+			changefeedID.ID))
+		return
+	}
+
+	status, err := h.capture.StatusProvider().GetChangeFeedSyncedStatus(ctx, changefeedID)
+	if err != nil {
+		_ = c.Error(err)
+		return
+	}
+
+	log.Info("Get changefeed synced status:", zap.Any("status", status), zap.Any("changefeedID", changefeedID))
+
+	cfg := &ChangefeedConfig{ReplicaConfig: GetDefaultReplicaConfig()}
+	if (status.SyncedCheckInterval != 0) && (status.CheckpointInterval != 0) {
+		cfg.ReplicaConfig.SyncedStatus.CheckpointInterval = status.CheckpointInterval
+		cfg.ReplicaConfig.SyncedStatus.SyncedCheckInterval = status.SyncedCheckInterval
+	}
+
+	// try to get pd client to get pd time, and determine synced status based on the pd time
+	if len(cfg.PDAddrs) == 0 {
+		up, err := getCaptureDefaultUpstream(h.capture)
+		if err != nil {
+			_ = c.Error(err)
+			return
+		}
+		cfg.PDConfig = getUpstreamPDConfig(up)
+	}
+	credential := cfg.PDConfig.toCredential()
+
+	timeoutCtx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+
+	pdClient, err := h.helpers.getPDClient(timeoutCtx, cfg.PDAddrs, credential)
+	if err != nil {
+		// case 1. we can't get pd client, pd may be unavailable.
+		//         if pullerResolvedTs - checkpointTs > checkpointInterval, data is not synced
+		//         otherwise, if pd is unavailable, we decide data whether is synced based on
+		//         the time difference between current time and lastSyncedTs.
+		var message string
+		if (oracle.ExtractPhysical(status.PullerResolvedTs) - oracle.ExtractPhysical(status.CheckpointTs)) >
+			cfg.ReplicaConfig.SyncedStatus.CheckpointInterval*1000 {
+			message = fmt.Sprintf("%s. Besides the data is not finish syncing", err.Error())
+		} else {
+			message = fmt.Sprintf("%s. You should check the pd status first. If pd status is normal, means we don't finish sync data. "+
+				"If pd is offline, please check whether we satisfy the condition that "+
+				"the time difference from lastSyncedTs to the current time from the time zone of pd is greater than %v secs. "+
+				"If it's satisfied, means the data syncing is totally finished", err, cfg.ReplicaConfig.SyncedStatus.SyncedCheckInterval)
+		}
+		c.JSON(http.StatusOK, SyncedStatus{
+			Synced:           false,
+			SinkCheckpointTs: model.JSONTime(oracle.GetTimeFromTS(status.CheckpointTs)),
+			PullerResolvedTs: model.JSONTime(oracle.GetTimeFromTS(status.PullerResolvedTs)),
+			LastSyncedTs:     model.JSONTime(oracle.GetTimeFromTS(status.LastSyncedTs)),
+			NowTs:            model.JSONTime(time.Unix(0, 0)),
+			Info:             message,
+		})
+		return
+	}
+	defer pdClient.Close()
+	// get time from pd
+	physicalNow, _, _ := pdClient.GetTS(ctx)
+
+	// We can normally get pd time. Thus we determine synced status based on physicalNow, lastSyncedTs, checkpointTs and pullerResolvedTs
+	if (physicalNow-oracle.ExtractPhysical(status.LastSyncedTs) > cfg.ReplicaConfig.SyncedStatus.SyncedCheckInterval*1000) &&
+		(physicalNow-oracle.ExtractPhysical(status.CheckpointTs) < cfg.ReplicaConfig.SyncedStatus.CheckpointInterval*1000) {
+		// case 2: If physcialNow - lastSyncedTs > SyncedCheckInterval && physcialNow - CheckpointTs < CheckpointInterval
+		//         --> reach strict synced status
+		c.JSON(http.StatusOK, SyncedStatus{
+			Synced:           true,
+			SinkCheckpointTs: model.JSONTime(oracle.GetTimeFromTS(status.CheckpointTs)),
+			PullerResolvedTs: model.JSONTime(oracle.GetTimeFromTS(status.PullerResolvedTs)),
+			LastSyncedTs:     model.JSONTime(oracle.GetTimeFromTS(status.LastSyncedTs)),
+			NowTs:            model.JSONTime(time.Unix(physicalNow/1e3, 0)),
+			Info:             "Data syncing is finished",
+		})
+		return
+	}
+
+	if physicalNow-oracle.ExtractPhysical(status.LastSyncedTs) > cfg.ReplicaConfig.SyncedStatus.SyncedCheckInterval*1000 {
+		// case 3: If physcialNow - lastSyncedTs > SyncedCheckInterval && physcialNow - CheckpointTs > CheckpointInterval
+		//         we should consider the situation that pd or tikv region is not healthy to block the advancing resolveTs.
+		//         if pullerResolvedTs - checkpointTs > CheckpointInterval-->  data is not synced
+		//         otherwise, if pd & tikv is healthy --> data is not synced
+		//                    if not healthy --> data is synced
+		var message string
+		if (oracle.ExtractPhysical(status.PullerResolvedTs) - oracle.ExtractPhysical(status.CheckpointTs)) <
+			cfg.ReplicaConfig.SyncedStatus.CheckpointInterval*1000 {
+			message = fmt.Sprintf("Please check whether PD is online and TiKV Regions are all available. " +
+				"If PD is offline or some TiKV regions are not available, it means that the data syncing process is complete. " +
+				"To check whether TiKV regions are all available, " +
+				"you can view 'TiKV-Details' > 'Resolved-Ts' > 'Max Leader Resolved TS gap' on Grafana. " +
+				"If the gap is large, such as a few minutes, it means that some regions in TiKV are unavailable. " +
+				"Otherwise, if the gap is small and PD is online, it means the data syncing is incomplete, so please wait")
+		} else {
+			message = "The data syncing is not finished, please wait"
+		}
+		c.JSON(http.StatusOK, SyncedStatus{
+			Synced:           false,
+			SinkCheckpointTs: model.JSONTime(oracle.GetTimeFromTS(status.CheckpointTs)),
+			PullerResolvedTs: model.JSONTime(oracle.GetTimeFromTS(status.PullerResolvedTs)),
+			LastSyncedTs:     model.JSONTime(oracle.GetTimeFromTS(status.LastSyncedTs)),
+			NowTs:            model.JSONTime(time.Unix(physicalNow/1e3, 0)),
+			Info:             message,
+		})
+		return
+	}
+
+	// case	4: If physcialNow - lastSyncedTs < SyncedCheckInterval --> data is not synced
+	c.JSON(http.StatusOK, SyncedStatus{
+		Synced:           false,
+		SinkCheckpointTs: model.JSONTime(oracle.GetTimeFromTS(status.CheckpointTs)),
+		PullerResolvedTs: model.JSONTime(oracle.GetTimeFromTS(status.PullerResolvedTs)),
+		LastSyncedTs:     model.JSONTime(oracle.GetTimeFromTS(status.LastSyncedTs)),
+		NowTs:            model.JSONTime(time.Unix(physicalNow/1e3, 0)),
+		Info:             "The data syncing is not finished, please wait",
+	})
+}
+
 func toAPIModel(
 	info *model.ChangeFeedInfo,
 	resolvedTs uint64,
diff --git a/cdc/api/v2/model.go b/cdc/api/v2/model.go
index bc5157da9..b8472ff01 100644
--- a/cdc/api/v2/model.go
+++ b/cdc/api/v2/model.go
@@ -102,6 +102,15 @@ type ChangefeedCommonInfo struct {
 	RunningError   *model.RunningError `json:"error"`
 }
 
+// SyncedStatusConfig represents synced check interval config for a changefeed
+type SyncedStatusConfig struct {
+	// The minimum interval between the latest synced ts and now required to reach synced state
+	SyncedCheckInterval int64 `json:"synced_check_interval"`
+	// The maximum interval between latest checkpoint ts and now or
+	// between latest sink's checkpoint ts and puller's checkpoint ts required to reach synced state
+	CheckpointInterval int64 `json:"checkpoint_interval"`
+}
+
 // MarshalJSON marshal changefeed common info to json
 // we need to set feed state to normal if it is uninitialized and pending to warning
 // to hide the detail of uninitialized and pending state from user
@@ -195,6 +204,7 @@ type ReplicaConfig struct {
 	Integrity                    *IntegrityConfig           `json:"integrity"`
 	ChangefeedErrorStuckDuration *JSONDuration              `json:"changefeed_error_stuck_duration,omitempty"`
 	SQLMode                      string                     `json:"sql_mode,omitempty"`
+	SyncedStatus                 *SyncedStatusConfig        `json:"synced_status,omitempty"`
 }
 
 // ToInternalReplicaConfig coverts *v2.ReplicaConfig into *config.ReplicaConfig
@@ -445,6 +455,12 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 	if c.ChangefeedErrorStuckDuration != nil {
 		res.ChangefeedErrorStuckDuration = &c.ChangefeedErrorStuckDuration.duration
 	}
+	if c.SyncedStatus != nil {
+		res.SyncedStatus = &config.SyncedStatusConfig{
+			SyncedCheckInterval: c.SyncedStatus.SyncedCheckInterval,
+			CheckpointInterval:  c.SyncedStatus.CheckpointInterval,
+		}
+	}
 	return res
 }
 
@@ -692,6 +708,12 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 	if cloned.ChangefeedErrorStuckDuration != nil {
 		res.ChangefeedErrorStuckDuration = &JSONDuration{*cloned.ChangefeedErrorStuckDuration}
 	}
+	if cloned.SyncedStatus != nil {
+		res.SyncedStatus = &SyncedStatusConfig{
+			SyncedCheckInterval: cloned.SyncedStatus.SyncedCheckInterval,
+			CheckpointInterval:  cloned.SyncedStatus.CheckpointInterval,
+		}
+	}
 	return res
 }
 
@@ -920,6 +942,16 @@ type ChangeFeedInfo struct {
 	TaskStatus     []model.CaptureTaskStatus `json:"task_status,omitempty"`
 }
 
+// SyncedStatus describes the detail of a changefeed's synced status
+type SyncedStatus struct {
+	Synced           bool           `json:"synced"`
+	SinkCheckpointTs model.JSONTime `json:"sink_checkpoint_ts"`
+	PullerResolvedTs model.JSONTime `json:"puller_resolved_ts"`
+	LastSyncedTs     model.JSONTime `json:"last_synced_ts"`
+	NowTs            model.JSONTime `json:"now_ts"`
+	Info             string         `json:"info"`
+}
+
 // RunningError represents some running error from cdc components,
 // such as processor.
 type RunningError struct {
diff --git a/cdc/entry/mounter.go b/cdc/entry/mounter.go
index 225540bdc..0b37ebc9e 100644
--- a/cdc/entry/mounter.go
+++ b/cdc/entry/mounter.go
@@ -16,6 +16,7 @@ package entry
 import (
 	"bytes"
 	"context"
+	"encoding/hex"
 	"encoding/json"
 	"fmt"
 	"math"
@@ -172,10 +173,18 @@ func (m *mounter) unmarshalAndMountRowChanged(ctx context.Context, raw *model.Ra
 		}
 		tableInfo, exist := snap.PhysicalTableByID(physicalTableID)
 		if !exist {
+			// for truncate table and truncate table partition DDL, the table ID is changed, but DML can be inserted to TiKV with old table ID.
+			// normally, cdc will close the old table pipeline and create a new one, and these invalid DMLs keys will not be pulled by CDC,
+			// but if redo is enabled or push based table pipeline is enabled, puller and mounter are not blocked by barrier ts.
+			// So some invalid DML keys will be decoded before processor removing the table pipeline
 			if snap.IsTruncateTableID(physicalTableID) {
 				log.Debug("skip the DML of truncated table", zap.Uint64("ts", raw.CRTs), zap.Int64("tableID", physicalTableID))
 				return nil, nil
 			}
+			log.Error("can not found table schema",
+				zap.Uint64("ts", raw.CRTs),
+				zap.String("key", hex.EncodeToString(raw.Key)),
+				zap.Int64("tableID", physicalTableID))
 			return nil, cerror.ErrSnapshotTableNotFound.GenWithStackByArgs(physicalTableID)
 		}
 		if bytes.HasPrefix(key, recordPrefix) {
diff --git a/cdc/entry/schema/snapshot.go b/cdc/entry/schema/snapshot.go
index cfb0a9d5c..1757c2672 100644
--- a/cdc/entry/schema/snapshot.go
+++ b/cdc/entry/schema/snapshot.go
@@ -467,11 +467,16 @@ func (s *Snapshot) DoHandleDDL(job *timodel.Job) error {
 		if err != nil {
 			return errors.Trace(err)
 		}
-	case timodel.ActionTruncateTablePartition,
+	case timodel.ActionTruncateTablePartition:
+		err := s.inner.updatePartition(getWrapTableInfo(job), true, job.BinlogInfo.FinishedTS)
+		if err != nil {
+			return errors.Trace(err)
+		}
+	case
 		timodel.ActionAddTablePartition,
 		timodel.ActionDropTablePartition,
 		timodel.ActionReorganizePartition:
-		err := s.inner.updatePartition(getWrapTableInfo(job), job.BinlogInfo.FinishedTS)
+		err := s.inner.updatePartition(getWrapTableInfo(job), false, job.BinlogInfo.FinishedTS)
 		if err != nil {
 			return errors.Trace(err)
 		}
@@ -857,7 +862,7 @@ func (s *snapshot) doCreateTable(tbInfo *model.TableInfo, currentTs uint64) {
 }
 
 // updatePartition updates partition info for `tbInfo`.
-func (s *snapshot) updatePartition(tbInfo *model.TableInfo, currentTs uint64) error {
+func (s *snapshot) updatePartition(tbInfo *model.TableInfo, isTruncate bool, currentTs uint64) error {
 	oldTbInfo, ok := s.physicalTableByID(tbInfo.ID)
 	if !ok {
 		return cerror.ErrSnapshotTableNotFound.GenWithStackByArgs(tbInfo.ID)
@@ -882,6 +887,7 @@ func (s *snapshot) updatePartition(tbInfo *model.TableInfo, currentTs uint64) er
 	for _, partition := range oldPi.Definitions {
 		s.partitions.ReplaceOrInsert(newVersionedID(partition.ID, tag))
 	}
+	newPartitionIDMap := make(map[int64]struct{}, len(newPi.Definitions))
 	for _, partition := range newPi.Definitions {
 		vid := newVersionedID(partition.ID, tag)
 		vid.target = tbInfo
@@ -889,6 +895,14 @@ func (s *snapshot) updatePartition(tbInfo *model.TableInfo, currentTs uint64) er
 		if ineligible {
 			s.ineligibleTables.ReplaceOrInsert(newVersionedID(partition.ID, tag))
 		}
+		newPartitionIDMap[partition.ID] = struct{}{}
+	}
+	if isTruncate {
+		for _, partition := range oldPi.Definitions {
+			if _, ok := newPartitionIDMap[partition.ID]; !ok {
+				s.truncatedTables.ReplaceOrInsert(newVersionedID(partition.ID, tag))
+			}
+		}
 	}
 	s.currentTs = currentTs
 
@@ -978,7 +992,7 @@ func (s *snapshot) exchangePartition(targetTable *model.TableInfo, currentTS uin
 	// ref: https://github.com/pingcap/tidb/issues/43819
 	targetTable.SchemaID = oldTable.SchemaID
 	targetTable.TableName = oldTable.TableName
-	err = s.updatePartition(targetTable, currentTS)
+	err = s.updatePartition(targetTable, false, currentTS)
 	if err != nil {
 		return errors.Trace(err)
 	}
diff --git a/cdc/model/changefeed.go b/cdc/model/changefeed.go
index 6868add16..02401ea7d 100644
--- a/cdc/model/changefeed.go
+++ b/cdc/model/changefeed.go
@@ -339,6 +339,9 @@ func (info *ChangeFeedInfo) VerifyAndComplete() {
 	if info.Config.SQLMode == "" {
 		info.Config.SQLMode = defaultConfig.SQLMode
 	}
+	if info.Config.SyncedStatus == nil {
+		info.Config.SyncedStatus = defaultConfig.SyncedStatus
+	}
 }
 
 // FixIncompatible fixes incompatible changefeed meta info.
@@ -582,3 +585,12 @@ type ChangeFeedStatusForAPI struct {
 	// initializing the changefeed.
 	MinTableBarrierTs uint64 `json:"min-table-barrier-ts"`
 }
+
+// ChangeFeedSyncedStatusForAPI uses to transfer the synced status of changefeed for API.
+type ChangeFeedSyncedStatusForAPI struct {
+	CheckpointTs        uint64 `json:"checkpoint-ts"`
+	LastSyncedTs        uint64 `json:"last-sync-time"`
+	PullerResolvedTs    uint64 `json:"puller-resolved-ts"`
+	SyncedCheckInterval int64  `json:"synced-check-interval"`
+	CheckpointInterval  int64  `json:"checkpoint-interval"`
+}
diff --git a/cdc/owner/changefeed.go b/cdc/owner/changefeed.go
old mode 100644
new mode 100755
index e79b39421..82524260e
--- a/cdc/owner/changefeed.go
+++ b/cdc/owner/changefeed.go
@@ -16,6 +16,7 @@ package owner
 import (
 	"context"
 	"fmt"
+	"math"
 	"strings"
 	"sync"
 	"time"
@@ -34,7 +35,7 @@ import (
 	"github.com/pingcap/tiflow/pkg/config"
 	cdcContext "github.com/pingcap/tiflow/pkg/context"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
-	"github.com/pingcap/tiflow/pkg/filter"
+	pfilter "github.com/pingcap/tiflow/pkg/filter"
 	"github.com/pingcap/tiflow/pkg/orchestrator"
 	"github.com/pingcap/tiflow/pkg/pdutil"
 	redoCfg "github.com/pingcap/tiflow/pkg/redo"
@@ -77,6 +78,13 @@ type changefeed struct {
 	feedStateManager *feedStateManager
 	resolvedTs       model.Ts
 
+	// lastSyncedTs is the lastest resolvedTs that has been synced to downstream.
+	// pullerResolvedTs is the minimum resolvedTs of all pullers.
+	// we don't need to initialize lastSyncedTs and pullerResolvedTs specially
+	// because it will be updated in tick.
+	lastSyncedTs     model.Ts
+	pullerResolvedTs model.Ts
+
 	// ddl related fields
 	ddlManager  *ddlManager
 	redoDDLMgr  redo.DDLManager
@@ -115,6 +123,9 @@ type changefeed struct {
 	metricsChangefeedBarrierTsGauge prometheus.Gauge
 	metricsChangefeedTickDuration   prometheus.Observer
 
+	metricsChangefeedCreateTimeGuage  prometheus.Gauge
+	metricsChangefeedRestartTimeGauge prometheus.Gauge
+
 	downstreamObserver observer.Observer
 	observerLastTick   *atomic.Time
 
@@ -124,7 +135,7 @@ type changefeed struct {
 		startTs uint64,
 		changefeed model.ChangeFeedID,
 		schemaStorage entry.SchemaStorage,
-		filter filter.Filter,
+		filter pfilter.Filter,
 	) (puller.DDLPuller, error)
 
 	newSink func(
@@ -180,7 +191,7 @@ func newChangefeed4Test(
 		startTs uint64,
 		changefeed model.ChangeFeedID,
 		schemaStorage entry.SchemaStorage,
-		filter filter.Filter,
+		filter pfilter.Filter,
 	) (puller.DDLPuller, error),
 	newSink func(
 		changefeedID model.ChangeFeedID, info *model.ChangeFeedInfo,
@@ -369,18 +380,35 @@ func (c *changefeed) tick(ctx cdcContext.Context, captures map[model.CaptureID]*
 		return nil
 	}
 
-	newCheckpointTs, newResolvedTs, err := c.scheduler.Tick(
+	watermark, err := c.scheduler.Tick(
 		ctx, preCheckpointTs, allPhysicalTables, captures, barrier)
 	if err != nil {
 		return errors.Trace(err)
 	}
+	if c.lastSyncedTs < watermark.LastSyncedTs {
+		c.lastSyncedTs = watermark.LastSyncedTs
+	} else if c.lastSyncedTs > watermark.LastSyncedTs {
+		log.Warn("LastSyncedTs should not be greater than newLastSyncedTs",
+			zap.Uint64("c.LastSyncedTs", c.lastSyncedTs),
+			zap.Uint64("newLastSyncedTs", watermark.LastSyncedTs))
+	}
+
+	if watermark.PullerResolvedTs != scheduler.CheckpointCannotProceed && watermark.PullerResolvedTs != math.MaxUint64 {
+		if watermark.PullerResolvedTs > c.pullerResolvedTs {
+			c.pullerResolvedTs = watermark.PullerResolvedTs
+		} else if watermark.PullerResolvedTs < c.pullerResolvedTs {
+			log.Warn("the newPullerResolvedTs should not be smaller than c.pullerResolvedTs",
+				zap.Uint64("c.pullerResolvedTs", c.pullerResolvedTs),
+				zap.Uint64("newPullerResolvedTs", watermark.PullerResolvedTs))
+		}
+	}
 
 	pdTime, _ := c.upstream.PDClock.CurrentTime()
 	currentTs := oracle.GetPhysical(pdTime)
 
 	// CheckpointCannotProceed implies that not all tables are being replicated normally,
 	// so in that case there is no need to advance the global watermarks.
-	if newCheckpointTs == scheduler.CheckpointCannotProceed {
+	if watermark.CheckpointTs == scheduler.CheckpointCannotProceed {
 		if c.state.Status != nil {
 			// We should keep the metrics updated even if the scheduler cannot
 			// advance the watermarks for now.
@@ -391,13 +419,13 @@ func (c *changefeed) tick(ctx cdcContext.Context, captures map[model.CaptureID]*
 
 	log.Debug("owner prepares to update status",
 		zap.Uint64("prevResolvedTs", c.resolvedTs),
-		zap.Uint64("newResolvedTs", newResolvedTs),
-		zap.Uint64("newCheckpointTs", newCheckpointTs),
+		zap.Uint64("newResolvedTs", watermark.ResolvedTs),
+		zap.Uint64("newCheckpointTs", watermark.CheckpointTs),
 		zap.String("namespace", c.id.Namespace),
 		zap.String("changefeed", c.id.ID))
 	// resolvedTs should never regress.
-	if newResolvedTs > c.resolvedTs {
-		c.resolvedTs = newResolvedTs
+	if watermark.ResolvedTs > c.resolvedTs {
+		c.resolvedTs = watermark.ResolvedTs
 	}
 
 	// MinTableBarrierTs should never regress
@@ -411,13 +439,17 @@ func (c *changefeed) tick(ctx cdcContext.Context, captures map[model.CaptureID]*
 				zap.String("namespace", c.id.Namespace),
 				zap.String("changefeed", c.id.ID),
 				zap.Uint64("keepCheckpoint", c.state.Status.CheckpointTs),
-				zap.Uint64("skipCheckpoint", newCheckpointTs))
-			newCheckpointTs = c.state.Status.CheckpointTs
+				zap.Uint64("skipCheckpoint", watermark.CheckpointTs))
+			watermark.CheckpointTs = c.state.Status.CheckpointTs
 		}
 	})
 
-	c.updateStatus(newCheckpointTs, barrier.MinTableBarrierTs)
-	c.updateMetrics(currentTs, newCheckpointTs, c.resolvedTs)
+	failpoint.Inject("ChangefeedOwnerNotUpdateCheckpoint", func() {
+		watermark.CheckpointTs = c.state.Status.CheckpointTs
+	})
+
+	c.updateStatus(watermark.CheckpointTs, barrier.MinTableBarrierTs)
+	c.updateMetrics(currentTs, watermark.CheckpointTs, c.resolvedTs)
 	c.tickDownstreamObserver(ctx)
 
 	return nil
@@ -526,7 +558,7 @@ LOOP2:
 	}
 	c.barriers.Update(finishBarrier, c.state.Info.GetTargetTs())
 
-	filter, err := filter.NewFilter(c.state.Info.Config, "")
+	filter, err := pfilter.NewFilter(c.state.Info.Config, "")
 	if err != nil {
 		return errors.Trace(err)
 	}
@@ -614,6 +646,7 @@ LOOP2:
 		ddlStartTs,
 		c.state.Status.CheckpointTs,
 		c.ddlSink,
+		filter,
 		c.ddlPuller,
 		c.schema,
 		c.redoDDLMgr,
@@ -633,6 +666,8 @@ LOOP2:
 	c.initMetrics()
 
 	c.initialized = true
+	c.metricsChangefeedCreateTimeGuage.Set(float64(oracle.GetPhysical(c.state.Info.CreateTime)))
+	c.metricsChangefeedRestartTimeGauge.Set(float64(oracle.GetPhysical(time.Now())))
 	log.Info("changefeed initialized",
 		zap.String("namespace", c.state.ID.Namespace),
 		zap.String("changefeed", c.state.ID.ID),
@@ -664,6 +699,11 @@ func (c *changefeed) initMetrics() {
 		WithLabelValues(c.id.Namespace, c.id.ID)
 	c.metricsChangefeedTickDuration = changefeedTickDuration.
 		WithLabelValues(c.id.Namespace, c.id.ID)
+
+	c.metricsChangefeedCreateTimeGuage = changefeedStartTimeGauge.
+		WithLabelValues(c.id.Namespace, c.id.ID, "create")
+	c.metricsChangefeedRestartTimeGauge = changefeedStartTimeGauge.
+		WithLabelValues(c.id.Namespace, c.id.ID, "restart")
 }
 
 // releaseResources is idempotent.
@@ -744,6 +784,8 @@ func (c *changefeed) cleanupMetrics() {
 
 	if c.isRemoved {
 		changefeedStatusGauge.DeleteLabelValues(c.id.Namespace, c.id.ID)
+		changefeedCheckpointTsGauge.DeleteLabelValues(c.id.Namespace, c.id.ID, "create")
+		changefeedCheckpointTsLagGauge.DeleteLabelValues(c.id.Namespace, c.id.ID, "restart")
 	}
 }
 
diff --git a/cdc/owner/ddl_manager.go b/cdc/owner/ddl_manager.go
index 1943ffb60..d3da4031c 100644
--- a/cdc/owner/ddl_manager.go
+++ b/cdc/owner/ddl_manager.go
@@ -108,6 +108,8 @@ type ddlManager struct {
 	// tableCheckpoint store the tableCheckpoint of each table. We need to wait
 	// for the tableCheckpoint to reach the next ddl commitTs before executing the ddl
 	tableCheckpoint map[model.TableName]model.Ts
+	filter          filter.Filter
+
 	// pendingDDLs store the pending DDL events of all tables
 	// the DDL events in the same table are ordered by commitTs.
 	pendingDDLs map[model.TableName][]*model.DDLEvent
@@ -133,6 +135,7 @@ func newDDLManager(
 	startTs model.Ts,
 	checkpointTs model.Ts,
 	ddlSink DDLSink,
+	filter filter.Filter,
 	ddlPuller puller.DDLPuller,
 	schema *schemaWrap4Owner,
 	redoManager redo.DDLManager,
@@ -151,6 +154,7 @@ func newDDLManager(
 	return &ddlManager{
 		changfeedID:     changefeedID,
 		ddlSink:         ddlSink,
+		filter:          filter,
 		ddlPuller:       ddlPuller,
 		schema:          schema,
 		redoDDLManager:  redoManager,
@@ -244,7 +248,14 @@ func (m *ddlManager) tick(
 			// Send DDL events to redo log.
 			if m.redoDDLManager.Enabled() {
 				for _, event := range events {
-					err := m.redoDDLManager.EmitDDLEvent(ctx, event)
+					skip, _, err := m.shouldSkipDDL(event)
+					if err != nil {
+						return nil, nil, errors.Trace(err)
+					}
+					if skip {
+						continue
+					}
+					err = m.redoDDLManager.EmitDDLEvent(ctx, event)
 					if err != nil {
 						return nil, nil, err
 					}
@@ -334,26 +345,34 @@ func (m *ddlManager) shouldExecDDL(nextDDL *model.DDLEvent) bool {
 	return checkpointReachBarrier && redoCheckpointReachBarrier && redoDDLResolvedTsExceedBarrier
 }
 
+func (m *ddlManager) shouldSkipDDL(ddl *model.DDLEvent) (bool, string, error) {
+	ignored, err := m.filter.ShouldIgnoreDDLEvent(ddl)
+	if err != nil {
+		return false, "", errors.Trace(err)
+	}
+	if ignored {
+		return true, "ddl is ignored by event filter rule, skip it", nil
+	}
+
+	// In a BDR mode cluster, TiCDC can receive DDLs from all roles of TiDB.
+	// However, CDC only executes the DDLs from the TiDB that has BDRRolePrimary role.
+	if m.BDRMode {
+		return true, "changefeed is in BDRMode, skip all ddl in release 6.5", nil
+	}
+	return false, "", nil
+}
+
 // executeDDL executes ddlManager.executingDDL.
 func (m *ddlManager) executeDDL(ctx context.Context) error {
 	if m.executingDDL == nil {
 		return nil
 	}
-
-	// If changefeed is in BDRMode, skip ddl.
-	if m.BDRMode {
-		log.Info("changefeed is in BDRMode, skip a ddl event",
-			zap.String("namespace", m.changfeedID.Namespace),
-			zap.String("ID", m.changfeedID.ID),
-			zap.Any("ddlEvent", m.executingDDL))
-		tableName := m.executingDDL.TableInfo.TableName
-		// Set it to nil first to accelerate GC.
-		m.pendingDDLs[tableName][0] = nil
-		m.pendingDDLs[tableName] = m.pendingDDLs[tableName][1:]
-		m.schema.DoGC(m.executingDDL.CommitTs - 1)
-		m.justSentDDL = m.executingDDL
-		m.executingDDL = nil
-		m.cleanCache()
+	skip, cleanMsg, err := m.shouldSkipDDL(m.executingDDL)
+	if err != nil {
+		return errors.Trace(err)
+	}
+	if skip {
+		m.cleanCache(cleanMsg)
 		return nil
 	}
 
@@ -375,22 +394,10 @@ func (m *ddlManager) executeDDL(ctx context.Context) error {
 
 	done, err := m.ddlSink.emitDDLEvent(ctx, m.executingDDL)
 	if err != nil {
-		return err
+		return errors.Trace(err)
 	}
 	if done {
-		tableName := m.executingDDL.TableInfo.TableName
-		log.Info("execute a ddl event successfully",
-			zap.String("ddl", m.executingDDL.Query),
-			zap.Uint64("commitTs", m.executingDDL.CommitTs),
-			zap.Stringer("table", tableName),
-		)
-		// Set it to nil first to accelerate GC.
-		m.pendingDDLs[tableName][0] = nil
-		m.pendingDDLs[tableName] = m.pendingDDLs[tableName][1:]
-		m.schema.DoGC(m.executingDDL.CommitTs - 1)
-		m.justSentDDL = m.executingDDL
-		m.executingDDL = nil
-		m.cleanCache()
+		m.cleanCache("execute a ddl event successfully")
 	}
 	return nil
 }
@@ -588,9 +595,21 @@ func (m *ddlManager) getSnapshotTs() (ts uint64) {
 }
 
 // cleanCache cleans the tableInfoCache and physicalTablesCache.
-// It should be called after a DDL is applied to schema or a DDL
-// is sent to downstream successfully.
-func (m *ddlManager) cleanCache() {
+// It should be called after a DDL is skipped or sent to downstream successfully.
+func (m *ddlManager) cleanCache(msg string) {
+	tableName := m.executingDDL.TableInfo.TableName
+	log.Info(msg, zap.String("ddl", m.executingDDL.Query),
+		zap.String("namespace", m.changfeedID.Namespace),
+		zap.String("changefeed", m.changfeedID.ID),
+		zap.Any("ddlEvent", m.executingDDL))
+
+	// Set it to nil first to accelerate GC.
+	m.pendingDDLs[tableName][0] = nil
+	m.pendingDDLs[tableName] = m.pendingDDLs[tableName][1:]
+	m.schema.DoGC(m.executingDDL.CommitTs - 1)
+	m.justSentDDL = m.executingDDL
+	m.executingDDL = nil
+
 	m.tableInfoCache = nil
 	m.physicalTablesCache = nil
 }
diff --git a/cdc/owner/metrics.go b/cdc/owner/metrics.go
index 97b7c43a4..38dc0f73c 100644
--- a/cdc/owner/metrics.go
+++ b/cdc/owner/metrics.go
@@ -120,6 +120,13 @@ var (
 			Name:      "ignored_ddl_event_count",
 			Help:      "The total count of ddl events that are ignored in changefeed.",
 		}, []string{"namespace", "changefeed"})
+	changefeedStartTimeGauge = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: "ticdc",
+			Subsystem: "owner",
+			Name:      "changefeed_start_time",
+			Help:      "The start time of changefeeds",
+		}, []string{"namespace", "changefeed", "type"})
 )
 
 const (
@@ -150,6 +157,7 @@ func InitMetrics(registry *prometheus.Registry) {
 	registry.MustRegister(changefeedTickDuration)
 	registry.MustRegister(changefeedCloseDuration)
 	registry.MustRegister(changefeedIgnoredDDLEventCounter)
+	registry.MustRegister(changefeedStartTimeGauge)
 }
 
 // lagBucket returns the lag buckets for prometheus metric
diff --git a/cdc/owner/mock/status_provider_mock.go b/cdc/owner/mock/status_provider_mock.go
index 5e1230f62..f86021b29 100644
--- a/cdc/owner/mock/status_provider_mock.go
+++ b/cdc/owner/mock/status_provider_mock.go
@@ -125,6 +125,21 @@ func (mr *MockStatusProviderMockRecorder) GetChangeFeedStatus(ctx, changefeedID
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetChangeFeedStatus", reflect.TypeOf((*MockStatusProvider)(nil).GetChangeFeedStatus), ctx, changefeedID)
 }
 
+// GetChangeFeedSyncedStatus mocks base method.
+func (m *MockStatusProvider) GetChangeFeedSyncedStatus(ctx context.Context, changefeedID model.ChangeFeedID) (*model.ChangeFeedSyncedStatusForAPI, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetChangeFeedSyncedStatus", ctx, changefeedID)
+	ret0, _ := ret[0].(*model.ChangeFeedSyncedStatusForAPI)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetChangeFeedSyncedStatus indicates an expected call of GetChangeFeedSyncedStatus.
+func (mr *MockStatusProviderMockRecorder) GetChangeFeedSyncedStatus(ctx, changefeedID interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetChangeFeedSyncedStatus", reflect.TypeOf((*MockStatusProvider)(nil).GetChangeFeedSyncedStatus), ctx, changefeedID)
+}
+
 // GetProcessors mocks base method.
 func (m *MockStatusProvider) GetProcessors(ctx context.Context) ([]*model.ProcInfoSnap, error) {
 	m.ctrl.T.Helper()
diff --git a/cdc/owner/owner.go b/cdc/owner/owner.go
index 12275337e..70ca2c567 100644
--- a/cdc/owner/owner.go
+++ b/cdc/owner/owner.go
@@ -522,6 +522,25 @@ func (o *ownerImpl) handleQueries(query *Query) error {
 			}
 		}
 		query.Data = ret
+	case QueryChangeFeedSyncedStatus:
+		cfReactor, ok := o.changefeeds[query.ChangeFeedID]
+		if !ok {
+			query.Data = nil
+			return nil
+		}
+		ret := &model.ChangeFeedSyncedStatusForAPI{}
+		ret.LastSyncedTs = cfReactor.lastSyncedTs
+		ret.CheckpointTs = cfReactor.state.Status.CheckpointTs
+		ret.PullerResolvedTs = cfReactor.pullerResolvedTs
+
+		if cfReactor.state == nil || cfReactor.state.Info == nil || cfReactor.state.Info.Config == nil {
+			ret.CheckpointInterval = 0
+			ret.SyncedCheckInterval = 0
+		} else {
+			ret.CheckpointInterval = cfReactor.state.Info.Config.SyncedStatus.CheckpointInterval
+			ret.SyncedCheckInterval = cfReactor.state.Info.Config.SyncedStatus.SyncedCheckInterval
+		}
+		query.Data = ret
 	case QueryAllTaskStatuses:
 		cfReactor, ok := o.changefeeds[query.ChangeFeedID]
 		if !ok {
diff --git a/cdc/owner/schema.go b/cdc/owner/schema.go
index b2f0ea39a..da8b208da 100644
--- a/cdc/owner/schema.go
+++ b/cdc/owner/schema.go
@@ -265,22 +265,15 @@ func (s *schemaWrap4Owner) filterDDLEvents(ddlEvents []*model.DDLEvent) ([]*mode
 			err     error
 		)
 		if event.Type == timodel.ActionRenameTable {
-			ignored, err = s.filter.ShouldDiscardDDL(
-				event.StartTs,
+			ignored = s.filter.ShouldDiscardDDL(
 				event.Type,
 				event.PreTableInfo.TableName.Schema,
-				event.PreTableInfo.TableName.Table,
-				event.Query)
-			if err != nil {
-				return nil, errors.Trace(err)
-			}
+				event.PreTableInfo.TableName.Table)
 		} else {
-			ignored, err = s.filter.ShouldDiscardDDL(
-				event.StartTs,
+			ignored = s.filter.ShouldDiscardDDL(
 				event.Type,
 				event.TableInfo.TableName.Schema,
-				event.TableInfo.TableName.Table,
-				event.Query)
+				event.TableInfo.TableName.Table)
 			if err != nil {
 				return nil, errors.Trace(err)
 			}
diff --git a/cdc/owner/status_provider.go b/cdc/owner/status_provider.go
index 9644f34e7..bf7efed87 100644
--- a/cdc/owner/status_provider.go
+++ b/cdc/owner/status_provider.go
@@ -33,6 +33,9 @@ type StatusProvider interface {
 	// GetAllChangeFeedInfo returns all changefeeds' info.
 	GetAllChangeFeedInfo(ctx context.Context) (map[model.ChangeFeedID]*model.ChangeFeedInfo, error)
 
+	// GetChangeFeedSyncedStatus returns a changefeeds' synced status.
+	GetChangeFeedSyncedStatus(ctx context.Context, changefeedID model.ChangeFeedID) (*model.ChangeFeedSyncedStatusForAPI, error)
+
 	// GetChangeFeedInfo returns a changefeeds' info.
 	GetChangeFeedInfo(ctx context.Context, changefeedID model.ChangeFeedID) (*model.ChangeFeedInfo, error)
 
@@ -65,6 +68,8 @@ const (
 	QueryCaptures
 	// QueryHealth is the type of query cluster health info.
 	QueryHealth
+	// QueryChangeFeedSyncedStatus is the type of query changefeed synced status
+	QueryChangeFeedSyncedStatus
 )
 
 // Query wraps query command and return results.
@@ -122,6 +127,22 @@ func (p *ownerStatusProvider) GetAllChangeFeedInfo(ctx context.Context) (
 	return query.Data.(map[model.ChangeFeedID]*model.ChangeFeedInfo), nil
 }
 
+func (p *ownerStatusProvider) GetChangeFeedSyncedStatus(ctx context.Context,
+	changefeedID model.ChangeFeedID,
+) (*model.ChangeFeedSyncedStatusForAPI, error) {
+	query := &Query{
+		Tp:           QueryChangeFeedSyncedStatus,
+		ChangeFeedID: changefeedID,
+	}
+	if err := p.sendQueryToOwner(ctx, query); err != nil {
+		return nil, errors.Trace(err)
+	}
+	if query.Data == nil {
+		return nil, cerror.ErrChangeFeedNotExists.GenWithStackByArgs(changefeedID)
+	}
+	return query.Data.(*model.ChangeFeedSyncedStatusForAPI), nil
+}
+
 func (p *ownerStatusProvider) GetChangeFeedInfo(ctx context.Context,
 	changefeedID model.ChangeFeedID,
 ) (*model.ChangeFeedInfo, error) {
diff --git a/cdc/processor/processor.go b/cdc/processor/processor.go
index e9202dc32..cb8a79a4a 100644
--- a/cdc/processor/processor.go
+++ b/cdc/processor/processor.go
@@ -342,6 +342,7 @@ func (p *processor) GetTableSpanStatus(span tablepb.Span, collectStat bool) tabl
 		Checkpoint: tablepb.Checkpoint{
 			CheckpointTs: sinkStats.CheckpointTs,
 			ResolvedTs:   sinkStats.ResolvedTs,
+			LastSyncedTs: sinkStats.LastSyncedTs,
 		},
 		State: state,
 		Stats: stats,
@@ -819,6 +820,8 @@ func (p *processor) updateBarrierTs(barrier *schedulepb.Barrier) {
 		globalBarrierTs = schemaResolvedTs
 	}
 	log.Debug("update barrierTs",
+		zap.String("namespace", p.changefeedID.Namespace),
+		zap.String("changefeed", p.changefeedID.ID),
 		zap.Any("tableBarriers", barrier.GetTableBarriers()),
 		zap.Uint64("globalBarrierTs", globalBarrierTs))
 
@@ -841,7 +844,10 @@ func (p *processor) getTableName(ctx context.Context, tableID model.TableID) str
 		retry.WithIsRetryableErr(cerror.IsRetryableError))
 
 	if tableName == nil {
-		log.Warn("failed to get table name for metric", zap.Any("tableID", tableID))
+		log.Warn("failed to get table name for metric",
+			zap.String("namespace", p.changefeedID.Namespace),
+			zap.String("changefeed", p.changefeedID.ID),
+			zap.Any("tableID", tableID))
 		return strconv.Itoa(int(tableID))
 	}
 
@@ -929,7 +935,10 @@ func (p *processor) Close() error {
 			zap.String("namespace", p.changefeedID.Namespace),
 			zap.String("changefeed", p.changefeedID.ID))
 		if err := p.agent.Close(); err != nil {
-			log.Warn("close agent meet error", zap.Error(err))
+			log.Warn("close agent meet error",
+				zap.String("namespace", p.changefeedID.Namespace),
+				zap.String("changefeed", p.changefeedID.ID),
+				zap.Error(err))
 		}
 		log.Info("Processor closed agent successfully",
 			zap.String("namespace", p.changefeedID.Namespace),
diff --git a/cdc/processor/sinkmanager/manager.go b/cdc/processor/sinkmanager/manager.go
index 776af8cdf..c2765e72a 100644
--- a/cdc/processor/sinkmanager/manager.go
+++ b/cdc/processor/sinkmanager/manager.go
@@ -58,6 +58,7 @@ const (
 type TableStats struct {
 	CheckpointTs model.Ts
 	ResolvedTs   model.Ts
+	LastSyncedTs model.Ts
 	BarrierTs    model.Ts
 }
 
@@ -1006,6 +1007,7 @@ func (m *SinkManager) GetTableStats(span tablepb.Span) TableStats {
 	tableSink := value.(*tableSinkWrapper)
 
 	checkpointTs := tableSink.getCheckpointTs()
+	lastSyncedTs := tableSink.getLastSyncedTs()
 	m.sinkMemQuota.Release(span, checkpointTs)
 	m.redoMemQuota.Release(span, checkpointTs)
 
@@ -1048,6 +1050,7 @@ func (m *SinkManager) GetTableStats(span tablepb.Span) TableStats {
 	return TableStats{
 		CheckpointTs: checkpointTs.ResolvedMark(),
 		ResolvedTs:   resolvedTs,
+		LastSyncedTs: lastSyncedTs,
 		BarrierTs:    tableSink.barrierTs.Load(),
 	}
 }
diff --git a/cdc/processor/sinkmanager/table_sink_wrapper.go b/cdc/processor/sinkmanager/table_sink_wrapper.go
index 5e10a4bdc..2171f04bf 100644
--- a/cdc/processor/sinkmanager/table_sink_wrapper.go
+++ b/cdc/processor/sinkmanager/table_sink_wrapper.go
@@ -58,6 +58,7 @@ type tableSinkWrapper struct {
 		advanced     time.Time
 		resolvedTs   model.ResolvedTs
 		checkpointTs model.ResolvedTs
+		lastSyncedTs model.Ts
 	}
 
 	// state used to control the lifecycle of the table.
@@ -223,6 +224,15 @@ func (t *tableSinkWrapper) updateResolvedTs(ts model.ResolvedTs) error {
 	return t.tableSink.s.UpdateResolvedTs(ts)
 }
 
+func (t *tableSinkWrapper) getLastSyncedTs() uint64 {
+	t.tableSink.RLock()
+	defer t.tableSink.RUnlock()
+	if t.tableSink.s != nil {
+		return t.tableSink.s.GetLastSyncedTs()
+	}
+	return t.tableSink.lastSyncedTs
+}
+
 func (t *tableSinkWrapper) getCheckpointTs() model.ResolvedTs {
 	t.tableSink.RLock()
 	defer t.tableSink.RUnlock()
@@ -362,6 +372,7 @@ func (t *tableSinkWrapper) doTableSinkClear() {
 		t.tableSink.checkpointTs = checkpointTs
 	}
 	t.tableSink.resolvedTs = checkpointTs
+	t.tableSink.lastSyncedTs = t.tableSink.s.GetLastSyncedTs()
 	t.tableSink.advanced = time.Now()
 	t.tableSink.innerMu.Unlock()
 	t.tableSink.s = nil
diff --git a/cdc/processor/tablepb/table.pb.go b/cdc/processor/tablepb/table.pb.go
index da01e6d59..c4b86e834 100644
--- a/cdc/processor/tablepb/table.pb.go
+++ b/cdc/processor/tablepb/table.pb.go
@@ -117,6 +117,7 @@ var xxx_messageInfo_Span proto.InternalMessageInfo
 type Checkpoint struct {
 	CheckpointTs Ts `protobuf:"varint,1,opt,name=checkpoint_ts,json=checkpointTs,proto3,casttype=Ts" json:"checkpoint_ts,omitempty"`
 	ResolvedTs   Ts `protobuf:"varint,2,opt,name=resolved_ts,json=resolvedTs,proto3,casttype=Ts" json:"resolved_ts,omitempty"`
+	LastSyncedTs Ts `protobuf:"varint,3,opt,name=last_synced_ts,json=lastSyncedTs,proto3,casttype=Ts" json:"last_synced_ts,omitempty"`
 }
 
 func (m *Checkpoint) Reset()         { *m = Checkpoint{} }
@@ -166,6 +167,13 @@ func (m *Checkpoint) GetResolvedTs() Ts {
 	return 0
 }
 
+func (m *Checkpoint) GetLastSyncedTs() Ts {
+	if m != nil {
+		return m.LastSyncedTs
+	}
+	return 0
+}
+
 // Stats holds a statistic for a table.
 type Stats struct {
 	// Number of captured regions.
@@ -329,50 +337,52 @@ func init() {
 func init() { proto.RegisterFile("processor/tablepb/table.proto", fileDescriptor_ae83c9c6cf5ef75c) }
 
 var fileDescriptor_ae83c9c6cf5ef75c = []byte{
-	// 688 bytes of a gzipped FileDescriptorProto
+	// 706 bytes of a gzipped FileDescriptorProto
 	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x54, 0xbf, 0x6f, 0xd3, 0x4e,
-	0x1c, 0xb5, 0x9d, 0x5f, 0xcd, 0x27, 0xf9, 0x7e, 0xe5, 0xde, 0xb7, 0xed, 0x37, 0x44, 0x22, 0x31,
-	0x51, 0x81, 0xaa, 0x95, 0x1c, 0x28, 0x0b, 0xea, 0xd6, 0xb4, 0x80, 0xaa, 0x0a, 0x09, 0xb9, 0x81,
-	0x81, 0x25, 0xf2, 0x8f, 0xc3, 0xb5, 0x1a, 0xce, 0x96, 0xef, 0xd2, 0x2a, 0x1b, 0x23, 0xca, 0x02,
-	0x13, 0x62, 0x89, 0xd4, 0x3f, 0xa7, 0x63, 0x47, 0x06, 0x14, 0x41, 0x2a, 0x66, 0xf6, 0x4e, 0xe8,
-	0xee, 0xdc, 0xb8, 0x09, 0x0c, 0xa1, 0x4b, 0x72, 0xbe, 0xf7, 0x3e, 0xcf, 0xef, 0xbd, 0x3b, 0x19,
-	0x6e, 0x47, 0x71, 0xe8, 0x62, 0x4a, 0xc3, 0xb8, 0xc9, 0x6c, 0xa7, 0x8b, 0x23, 0x47, 0xfe, 0x9b,
-	0x51, 0x1c, 0xb2, 0x10, 0xad, 0x46, 0x01, 0xf1, 0x5d, 0x3b, 0x32, 0x59, 0xf0, 0xa6, 0x1b, 0x9e,
-	0x98, 0xae, 0xe7, 0x9a, 0x93, 0x09, 0x33, 0x99, 0xa8, 0x2e, 0xf9, 0xa1, 0x1f, 0x8a, 0x81, 0x26,
-	0x5f, 0xc9, 0xd9, 0xc6, 0x07, 0x15, 0xb2, 0x07, 0x91, 0x4d, 0xd0, 0x43, 0x58, 0x10, 0xcc, 0x4e,
-	0xe0, 0x55, 0x54, 0x43, 0x5d, 0xcb, 0xb4, 0x56, 0xc6, 0xa3, 0x7a, 0xa1, 0xcd, 0xf7, 0xf6, 0x76,
-	0x2f, 0xd3, 0xa5, 0x55, 0x10, 0xbc, 0x3d, 0x0f, 0xad, 0x42, 0x91, 0x32, 0x3b, 0x66, 0x9d, 0x23,
-	0xdc, 0xaf, 0x68, 0x86, 0xba, 0x56, 0x6e, 0x15, 0x2e, 0x47, 0xf5, 0xcc, 0x3e, 0xee, 0x5b, 0x0b,
-	0x02, 0xd9, 0xc7, 0x7d, 0x64, 0x40, 0x01, 0x13, 0x4f, 0x70, 0x32, 0xd3, 0x9c, 0x3c, 0x26, 0xde,
-	0x3e, 0xee, 0x6f, 0x95, 0xdf, 0x9f, 0xd6, 0x95, 0xcf, 0xa7, 0x75, 0xe5, 0xdd, 0x57, 0x43, 0x69,
-	0x38, 0x00, 0x3b, 0x87, 0xd8, 0x3d, 0x8a, 0xc2, 0x80, 0x30, 0xb4, 0x01, 0xff, 0xb8, 0x93, 0xa7,
-	0x0e, 0xa3, 0xc2, 0x5b, 0xb6, 0x95, 0xbf, 0x1c, 0xd5, 0xb5, 0x36, 0xb5, 0xca, 0x29, 0xd8, 0xa6,
-	0xe8, 0x3e, 0x94, 0x62, 0x4c, 0xc3, 0xee, 0x31, 0xf6, 0x38, 0x55, 0x9b, 0xa2, 0xc2, 0x15, 0xd4,
-	0xa6, 0x8d, 0x1f, 0x1a, 0xe4, 0x0e, 0x98, 0xcd, 0x28, 0xba, 0x03, 0xe5, 0x18, 0xfb, 0x41, 0x48,
-	0x3a, 0x6e, 0xd8, 0x23, 0x4c, 0xca, 0x5b, 0x25, 0xb9, 0xb7, 0xc3, 0xb7, 0xd0, 0x5d, 0x00, 0xb7,
-	0x17, 0xc7, 0x58, 0xbe, 0x7f, 0x5a, 0xb4, 0x98, 0x20, 0x6d, 0x8a, 0x18, 0x2c, 0x52, 0x66, 0xfb,
-	0xb8, 0x93, 0x5a, 0xa2, 0x95, 0x8c, 0x91, 0x59, 0x2b, 0x6d, 0x6e, 0x9b, 0xf3, 0x9c, 0x90, 0x29,
-	0x1c, 0xf1, 0x5f, 0x1f, 0xa7, 0x0d, 0xd0, 0x27, 0x84, 0xc5, 0xfd, 0x56, 0xf6, 0x6c, 0x54, 0x57,
-	0x2c, 0x9d, 0xce, 0x80, 0xdc, 0x9c, 0x63, 0xc7, 0x71, 0x80, 0x63, 0x6e, 0x2e, 0x3b, 0x6d, 0x2e,
-	0x41, 0xda, 0xb4, 0xda, 0x83, 0xe5, 0x3f, 0xea, 0x22, 0x1d, 0x32, 0xfc, 0x64, 0x78, 0xec, 0xa2,
-	0xc5, 0x97, 0xe8, 0x29, 0xe4, 0x8e, 0xed, 0x6e, 0x0f, 0x8b, 0xa4, 0xa5, 0xcd, 0x07, 0xf3, 0x79,
-	0x4f, 0x85, 0x2d, 0x39, 0xbe, 0xa5, 0x3d, 0x56, 0x1b, 0x3f, 0x35, 0x28, 0x89, 0x6b, 0xc3, 0xa3,
-	0xf5, 0xe8, 0x4d, 0x2e, 0xd9, 0x2e, 0x64, 0x69, 0x64, 0x93, 0x4a, 0x4e, 0xb8, 0x59, 0x9f, 0xb3,
-	0xc9, 0xc8, 0x26, 0x49, 0x65, 0x62, 0x9a, 0x87, 0xa2, 0xcc, 0x66, 0x32, 0xd4, 0xbf, 0xf3, 0x86,
-	0x9a, 0x58, 0xc7, 0x96, 0x1c, 0x47, 0xaf, 0x00, 0xd2, 0xe3, 0x15, 0xf7, 0xf9, 0x06, 0x0d, 0x25,
-	0xce, 0xae, 0x29, 0xa1, 0x67, 0xd2, 0x9f, 0x3c, 0xc1, 0xd2, 0xe6, 0xc6, 0x5f, 0x5c, 0x98, 0x44,
-	0x4d, 0xce, 0xaf, 0x7f, 0xd2, 0x00, 0x52, 0xdb, 0xa8, 0x01, 0x85, 0x97, 0xe4, 0x88, 0x84, 0x27,
-	0x44, 0x57, 0xaa, 0xcb, 0x83, 0xa1, 0xb1, 0x98, 0x82, 0x09, 0x80, 0x0c, 0xc8, 0x6f, 0x3b, 0x14,
-	0x13, 0xa6, 0xab, 0xd5, 0xa5, 0xc1, 0xd0, 0xd0, 0x53, 0x8a, 0xdc, 0x47, 0xf7, 0xa0, 0xf8, 0x22,
-	0xc6, 0x91, 0x1d, 0x07, 0xc4, 0xd7, 0xb5, 0xea, 0xff, 0x83, 0xa1, 0xf1, 0x5f, 0x4a, 0x9a, 0x40,
-	0x68, 0x15, 0x16, 0xe4, 0x03, 0xf6, 0xf4, 0x4c, 0x75, 0x65, 0x30, 0x34, 0xd0, 0x2c, 0x0d, 0x7b,
-	0x68, 0x1d, 0x4a, 0x16, 0x8e, 0xba, 0x81, 0x6b, 0x33, 0xae, 0x97, 0xad, 0xde, 0x1a, 0x0c, 0x8d,
-	0xe5, 0x6b, 0x5d, 0xa7, 0x20, 0x57, 0x3c, 0x60, 0x61, 0xc4, 0xdb, 0xd0, 0x73, 0xb3, 0x8a, 0x57,
-	0x08, 0x4f, 0x29, 0xd6, 0xd8, 0xd3, 0xf3, 0xb3, 0x29, 0x13, 0xa0, 0xf5, 0xfc, 0xfc, 0x7b, 0x4d,
-	0x39, 0x1b, 0xd7, 0xd4, 0xf3, 0x71, 0x4d, 0xfd, 0x36, 0xae, 0xa9, 0x1f, 0x2f, 0x6a, 0xca, 0xf9,
-	0x45, 0x4d, 0xf9, 0x72, 0x51, 0x53, 0x5e, 0x37, 0xfd, 0x80, 0x1d, 0xf6, 0x1c, 0xd3, 0x0d, 0xdf,
-	0x36, 0x93, 0xea, 0x9b, 0xb2, 0xfa, 0xa6, 0xeb, 0xb9, 0xcd, 0xdf, 0xbe, 0xbf, 0x4e, 0x5e, 0x7c,
-	0x3e, 0x1f, 0xfd, 0x0a, 0x00, 0x00, 0xff, 0xff, 0x80, 0x0e, 0x45, 0x99, 0x9b, 0x05, 0x00, 0x00,
+	0x1c, 0xb5, 0x9d, 0x5f, 0xcd, 0x27, 0xf9, 0x56, 0xee, 0x7d, 0xdb, 0x52, 0x22, 0x91, 0x98, 0xa8,
+	0x40, 0xd5, 0x22, 0x07, 0xca, 0x82, 0xba, 0x35, 0x2d, 0xa0, 0xaa, 0x42, 0x42, 0x4e, 0x60, 0x60,
+	0x89, 0x1c, 0xfb, 0x70, 0xad, 0x86, 0xb3, 0xe5, 0xbb, 0xb4, 0xca, 0xc6, 0x88, 0xb2, 0xd0, 0x09,
+	0xb1, 0x44, 0xea, 0x9f, 0xd3, 0xb1, 0x23, 0x03, 0x8a, 0x20, 0x15, 0x33, 0x7b, 0x27, 0x74, 0x77,
+	0x6e, 0xdc, 0x04, 0x86, 0xd0, 0x25, 0x39, 0xdf, 0x7b, 0x9f, 0xa7, 0xf7, 0xde, 0x9d, 0x0e, 0xee,
+	0x84, 0x51, 0xe0, 0x60, 0x4a, 0x83, 0xa8, 0xc6, 0xec, 0x76, 0x07, 0x87, 0x6d, 0xf9, 0x6f, 0x86,
+	0x51, 0xc0, 0x02, 0xb4, 0x1a, 0xfa, 0xc4, 0x73, 0xec, 0xd0, 0x64, 0xfe, 0xbb, 0x4e, 0x70, 0x6c,
+	0x3a, 0xae, 0x63, 0x8e, 0x27, 0xcc, 0x78, 0xa2, 0xb4, 0xe8, 0x05, 0x5e, 0x20, 0x06, 0x6a, 0x7c,
+	0x25, 0x67, 0xab, 0x9f, 0x54, 0x48, 0x37, 0x42, 0x9b, 0xa0, 0xc7, 0x30, 0x27, 0x98, 0x2d, 0xdf,
+	0x5d, 0x51, 0x0d, 0x75, 0x2d, 0x55, 0x5f, 0x1e, 0x0d, 0x2b, 0xb9, 0x26, 0xdf, 0xdb, 0xdb, 0xbd,
+	0x4c, 0x96, 0x56, 0x4e, 0xf0, 0xf6, 0x5c, 0xb4, 0x0a, 0x79, 0xca, 0xec, 0x88, 0xb5, 0x0e, 0x71,
+	0x6f, 0x45, 0x33, 0xd4, 0xb5, 0x62, 0x3d, 0x77, 0x39, 0xac, 0xa4, 0xf6, 0x71, 0xcf, 0x9a, 0x13,
+	0xc8, 0x3e, 0xee, 0x21, 0x03, 0x72, 0x98, 0xb8, 0x82, 0x93, 0x9a, 0xe4, 0x64, 0x31, 0x71, 0xf7,
+	0x71, 0x6f, 0xab, 0xf8, 0xf1, 0xb4, 0xa2, 0x7c, 0x39, 0xad, 0x28, 0x1f, 0xbe, 0x19, 0x4a, 0xf5,
+	0x44, 0x05, 0xd8, 0x39, 0xc0, 0xce, 0x61, 0x18, 0xf8, 0x84, 0xa1, 0x0d, 0xf8, 0xcf, 0x19, 0x7f,
+	0xb5, 0x18, 0x15, 0xe6, 0xd2, 0xf5, 0xec, 0xe5, 0xb0, 0xa2, 0x35, 0xa9, 0x55, 0x4c, 0xc0, 0x26,
+	0x45, 0x0f, 0xa0, 0x10, 0x61, 0x1a, 0x74, 0x8e, 0xb0, 0xcb, 0xa9, 0xda, 0x04, 0x15, 0xae, 0xa0,
+	0x26, 0x45, 0x0f, 0x61, 0xbe, 0x63, 0x53, 0xd6, 0xa2, 0x3d, 0xe2, 0x48, 0x6e, 0x6a, 0x52, 0x96,
+	0xa3, 0x0d, 0x01, 0x36, 0x69, 0xf5, 0xa7, 0x06, 0x99, 0x06, 0xb3, 0x19, 0x45, 0x77, 0xa1, 0x18,
+	0x61, 0xcf, 0x0f, 0x48, 0xcb, 0x09, 0xba, 0x84, 0x49, 0x33, 0x56, 0x41, 0xee, 0xed, 0xf0, 0x2d,
+	0x74, 0x0f, 0xc0, 0xe9, 0x46, 0x11, 0x96, 0x6e, 0x27, 0x2d, 0xe4, 0x63, 0xa4, 0x49, 0x11, 0x83,
+	0x05, 0xca, 0x6c, 0x0f, 0xb7, 0x92, 0x00, 0xdc, 0x44, 0x6a, 0xad, 0xb0, 0xb9, 0x6d, 0xce, 0x72,
+	0xa0, 0xa6, 0x70, 0xc4, 0x7f, 0x3d, 0x9c, 0xf4, 0x45, 0x9f, 0x11, 0x16, 0xf5, 0xea, 0xe9, 0xb3,
+	0x61, 0x45, 0xb1, 0x74, 0x3a, 0x05, 0x72, 0x73, 0x6d, 0x3b, 0x8a, 0x7c, 0x1c, 0x71, 0x73, 0xe9,
+	0x49, 0x73, 0x31, 0xd2, 0xa4, 0xa5, 0x2e, 0x2c, 0xfd, 0x55, 0x17, 0xe9, 0x90, 0xe2, 0x07, 0xc9,
+	0x63, 0xe7, 0x2d, 0xbe, 0x44, 0xcf, 0x21, 0x73, 0x64, 0x77, 0xba, 0x58, 0x24, 0x2d, 0x6c, 0x3e,
+	0x9a, 0xcd, 0x7b, 0x22, 0x6c, 0xc9, 0xf1, 0x2d, 0xed, 0xa9, 0x5a, 0xfd, 0xa5, 0x41, 0x41, 0xdc,
+	0x32, 0x1e, 0xad, 0x4b, 0x6f, 0x72, 0x27, 0x77, 0x21, 0x4d, 0x43, 0x9b, 0xac, 0x64, 0x84, 0x9b,
+	0xf5, 0x19, 0x9b, 0x0c, 0x6d, 0x12, 0x57, 0x26, 0xa6, 0x79, 0x28, 0xca, 0x6c, 0x26, 0x43, 0xcd,
+	0xcf, 0x1a, 0x6a, 0x6c, 0x1d, 0x5b, 0x72, 0x1c, 0xbd, 0x01, 0x48, 0x8e, 0x57, 0x5c, 0xb1, 0x1b,
+	0x34, 0x14, 0x3b, 0xbb, 0xa6, 0x84, 0x5e, 0x48, 0x7f, 0xf2, 0x04, 0x0b, 0x9b, 0x1b, 0xff, 0x70,
+	0x61, 0x62, 0x35, 0x39, 0xbf, 0xfe, 0x59, 0x03, 0x48, 0x6c, 0xa3, 0x2a, 0xe4, 0x5e, 0x93, 0x43,
+	0x12, 0x1c, 0x13, 0x5d, 0x29, 0x2d, 0xf5, 0x07, 0xc6, 0x42, 0x02, 0xc6, 0x00, 0x32, 0x20, 0xbb,
+	0xdd, 0xa6, 0x98, 0x30, 0x5d, 0x2d, 0x2d, 0xf6, 0x07, 0x86, 0x9e, 0x50, 0xe4, 0x3e, 0xba, 0x0f,
+	0xf9, 0x57, 0x11, 0x0e, 0xed, 0xc8, 0x27, 0x9e, 0xae, 0x95, 0x6e, 0xf5, 0x07, 0xc6, 0xff, 0x09,
+	0x69, 0x0c, 0xa1, 0x55, 0x98, 0x93, 0x1f, 0xd8, 0xd5, 0x53, 0xa5, 0xe5, 0xfe, 0xc0, 0x40, 0xd3,
+	0x34, 0xec, 0xa2, 0x75, 0x28, 0x58, 0x38, 0xec, 0xf8, 0x8e, 0xcd, 0xb8, 0x5e, 0xba, 0x74, 0xbb,
+	0x3f, 0x30, 0x96, 0xae, 0x75, 0x9d, 0x80, 0x5c, 0xb1, 0xc1, 0x82, 0x90, 0xb7, 0xa1, 0x67, 0xa6,
+	0x15, 0xaf, 0x10, 0x9e, 0x52, 0xac, 0xb1, 0xab, 0x67, 0xa7, 0x53, 0xc6, 0x40, 0xfd, 0xe5, 0xf9,
+	0x8f, 0xb2, 0x72, 0x36, 0x2a, 0xab, 0xe7, 0xa3, 0xb2, 0xfa, 0x7d, 0x54, 0x56, 0x4f, 0x2e, 0xca,
+	0xca, 0xf9, 0x45, 0x59, 0xf9, 0x7a, 0x51, 0x56, 0xde, 0xd6, 0x3c, 0x9f, 0x1d, 0x74, 0xdb, 0xa6,
+	0x13, 0xbc, 0xaf, 0xc5, 0xd5, 0xd7, 0x64, 0xf5, 0x35, 0xc7, 0x75, 0x6a, 0x7f, 0x3c, 0xd7, 0xed,
+	0xac, 0x78, 0x6d, 0x9f, 0xfc, 0x0e, 0x00, 0x00, 0xff, 0xff, 0xa3, 0xeb, 0x06, 0x9b, 0xca, 0x05,
+	0x00, 0x00,
 }
 
 func (m *Span) Marshal() (dAtA []byte, err error) {
@@ -437,6 +447,11 @@ func (m *Checkpoint) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	_ = i
 	var l int
 	_ = l
+	if m.LastSyncedTs != 0 {
+		i = encodeVarintTable(dAtA, i, uint64(m.LastSyncedTs))
+		i--
+		dAtA[i] = 0x18
+	}
 	if m.ResolvedTs != 0 {
 		i = encodeVarintTable(dAtA, i, uint64(m.ResolvedTs))
 		i--
@@ -618,6 +633,9 @@ func (m *Checkpoint) Size() (n int) {
 	if m.ResolvedTs != 0 {
 		n += 1 + sovTable(uint64(m.ResolvedTs))
 	}
+	if m.LastSyncedTs != 0 {
+		n += 1 + sovTable(uint64(m.LastSyncedTs))
+	}
 	return n
 }
 
@@ -879,6 +897,25 @@ func (m *Checkpoint) Unmarshal(dAtA []byte) error {
 					break
 				}
 			}
+		case 3:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field LastSyncedTs", wireType)
+			}
+			m.LastSyncedTs = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowTable
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.LastSyncedTs |= Ts(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
 		default:
 			iNdEx = preIndex
 			skippy, err := skipTable(dAtA[iNdEx:])
diff --git a/cdc/processor/tablepb/table.proto b/cdc/processor/tablepb/table.proto
index 97de3a23e..3727edc0c 100644
--- a/cdc/processor/tablepb/table.proto
+++ b/cdc/processor/tablepb/table.proto
@@ -57,6 +57,7 @@ enum TableState {
 message Checkpoint {
     uint64 checkpoint_ts = 1 [(gogoproto.casttype) = "Ts"];
     uint64 resolved_ts = 2 [(gogoproto.casttype) = "Ts"];
+    uint64 last_synced_ts = 3 [(gogoproto.casttype) = "Ts"];
 }
 
 // Stats holds a statistic for a table.
diff --git a/cdc/puller/ddl_puller.go b/cdc/puller/ddl_puller.go
index 97db61d71..8fa8e30ea 100644
--- a/cdc/puller/ddl_puller.go
+++ b/cdc/puller/ddl_puller.go
@@ -261,11 +261,7 @@ func (p *ddlJobPullerImpl) handleRenameTables(job *timodel.Job) (skip bool, err
 		if !ok {
 			shouldDiscardOldTable = true
 		} else {
-			shouldDiscardOldTable, err = p.filter.ShouldDiscardDDL(job.StartTS,
-				job.Type, oldSchemaNames[i].O, oldTable.Name.O, job.Query)
-			if err != nil {
-				return true, errors.Trace(err)
-			}
+			shouldDiscardOldTable = p.filter.ShouldDiscardDDL(job.Type, oldSchemaNames[i].O, oldTable.Name.O)
 		}
 
 		newSchemaName, ok := snap.SchemaByID(newSchemaIDs[i])
@@ -273,11 +269,7 @@ func (p *ddlJobPullerImpl) handleRenameTables(job *timodel.Job) (skip bool, err
 			// the new table name does not hit the filter rule, so we should discard the table.
 			shouldDiscardNewTable = true
 		} else {
-			shouldDiscardNewTable, err = p.filter.ShouldDiscardDDL(job.StartTS,
-				job.Type, newSchemaName.Name.O, newTableNames[i].O, job.Query)
-			if err != nil {
-				return true, errors.Trace(err)
-			}
+			shouldDiscardNewTable = p.filter.ShouldDiscardDDL(job.Type, newSchemaName.Name.O, newTableNames[i].O)
 		}
 
 		if shouldDiscardOldTable && shouldDiscardNewTable {
@@ -389,12 +381,7 @@ func (p *ddlJobPullerImpl) handleJob(job *timodel.Job) (skip bool, err error) {
 	snap := p.schemaStorage.GetLastSnapshot()
 	if err := snap.FillSchemaName(job); err != nil {
 		log.Info("failed to fill schema name for ddl job", zap.Error(err))
-		discard, fErr := p.filter.
-			ShouldDiscardDDL(job.StartTS, job.Type, job.SchemaName, job.TableName, job.Query)
-		if fErr != nil {
-			return false, errors.Trace(fErr)
-		}
-		if discard {
+		if p.filter.ShouldDiscardDDL(job.Type, job.SchemaName, job.TableName) {
 			return true, nil
 		}
 		return true, errors.Trace(err)
@@ -417,11 +404,7 @@ func (p *ddlJobPullerImpl) handleJob(job *timodel.Job) (skip bool, err error) {
 		oldTable, ok := snap.PhysicalTableByID(job.TableID)
 		if !ok {
 			// 1. If we can not find the old table, and the new table name is in filter rule, return error.
-			discard, err := p.filter.
-				ShouldDiscardDDL(job.StartTS, job.Type, job.SchemaName, job.BinlogInfo.TableInfo.Name.O, job.Query)
-			if err != nil {
-				return true, errors.Trace(err)
-			}
+			discard := p.filter.ShouldDiscardDDL(job.Type, job.SchemaName, job.BinlogInfo.TableInfo.Name.O)
 			if !discard {
 				return true, cerror.ErrSyncRenameTableFailed.GenWithStackByArgs(job.TableID, job.Query)
 			}
@@ -432,16 +415,8 @@ func (p *ddlJobPullerImpl) handleJob(job *timodel.Job) (skip bool, err error) {
 				zap.String("oldSchemaName", oldTable.TableName.Schema))
 			// since we can find the old table, we must can find the old schema.
 			// 2. If we can find the preTableInfo, we filter it by the old table name.
-			skipByOldTableName, err := p.filter.ShouldDiscardDDL(job.StartTS,
-				job.Type, oldTable.TableName.Schema, oldTable.TableName.Table, job.Query)
-			if err != nil {
-				return true, errors.Trace(err)
-			}
-			skipByNewTableName, err := p.filter.ShouldDiscardDDL(job.StartTS,
-				job.Type, job.SchemaName, job.BinlogInfo.TableInfo.Name.O, job.Query)
-			if err != nil {
-				return true, errors.Trace(err)
-			}
+			skipByOldTableName := p.filter.ShouldDiscardDDL(job.Type, oldTable.TableName.Schema, oldTable.TableName.Table)
+			skipByNewTableName := p.filter.ShouldDiscardDDL(job.Type, job.SchemaName, job.BinlogInfo.TableInfo.Name.O)
 			// 3. If its old table name is not in filter rule, and its new table name in filter rule, return error.
 			if skipByOldTableName && !skipByNewTableName {
 				return true, cerror.ErrSyncRenameTableFailed.GenWithStackByArgs(job.TableID, job.Query)
@@ -456,11 +431,7 @@ func (p *ddlJobPullerImpl) handleJob(job *timodel.Job) (skip bool, err error) {
 		if job.BinlogInfo.TableInfo != nil {
 			job.TableName = job.BinlogInfo.TableInfo.Name.O
 		}
-		skip, err = p.filter.
-			ShouldDiscardDDL(job.StartTS, job.Type, job.SchemaName, job.TableName, job.Query)
-		if err != nil {
-			return false, errors.Trace(err)
-		}
+		skip = p.filter.ShouldDiscardDDL(job.Type, job.SchemaName, job.TableName)
 	}
 
 	if skip {
diff --git a/cdc/puller/frontier/frontier.go b/cdc/puller/frontier/frontier.go
index c4e13f9f5..aa2bfabad 100644
--- a/cdc/puller/frontier/frontier.go
+++ b/cdc/puller/frontier/frontier.go
@@ -80,7 +80,7 @@ func (s *spanFrontier) Frontier() uint64 {
 func (s *spanFrontier) Forward(regionID uint64, span tablepb.Span, ts uint64) {
 	// it's the fast part to detect if the region is split or merged,
 	// if not we can update the minTsHeap with use new ts directly
-	if n, ok := s.cachedRegions[regionID]; ok && n.regionID != fakeRegionID && n.end != nil {
+	if n, ok := s.cachedRegions[regionID]; ok && n.regionID == regionID && n.end != nil {
 		if bytes.Equal(n.Key(), span.StartKey) && bytes.Equal(n.End(), span.EndKey) {
 			s.minTsHeap.UpdateKey(n.Value(), ts)
 			return
@@ -102,6 +102,7 @@ func (s *spanFrontier) insert(regionID uint64, span tablepb.Span, ts uint64) {
 		if bytes.Equal(seekRes.Node().Key(), span.StartKey) &&
 			bytes.Equal(next.Key(), span.EndKey) {
 			s.minTsHeap.UpdateKey(seekRes.Node().Value(), ts)
+			delete(s.cachedRegions, seekRes.Node().regionID)
 			if regionID != fakeRegionID {
 				s.cachedRegions[regionID] = seekRes.Node()
 				s.cachedRegions[regionID].regionID = regionID
diff --git a/cdc/redo/manager.go b/cdc/redo/manager.go
index 7d2370c0c..af3b77fc2 100644
--- a/cdc/redo/manager.go
+++ b/cdc/redo/manager.go
@@ -374,7 +374,10 @@ func (m *logManager) GetResolvedTs(span tablepb.Span) model.Ts {
 func (m *logManager) AddTable(span tablepb.Span, startTs uint64) {
 	_, loaded := m.rtsMap.LoadOrStore(span, &statefulRts{flushed: startTs, unflushed: startTs})
 	if loaded {
-		log.Warn("add duplicated table in redo log manager", zap.Stringer("span", &span))
+		log.Warn("add duplicated table in redo log manager",
+			zap.String("namespace", m.cfg.ChangeFeedID.Namespace),
+			zap.String("changefeed", m.cfg.ChangeFeedID.ID),
+			zap.Stringer("span", &span))
 		return
 	}
 }
@@ -382,7 +385,10 @@ func (m *logManager) AddTable(span tablepb.Span, startTs uint64) {
 // RemoveTable removes a table from redo log manager
 func (m *logManager) RemoveTable(span tablepb.Span) {
 	if _, ok := m.rtsMap.LoadAndDelete(span); !ok {
-		log.Warn("remove a table not maintained in redo log manager", zap.Stringer("span", &span))
+		log.Warn("remove a table not maintained in redo log manager",
+			zap.String("namespace", m.cfg.ChangeFeedID.Namespace),
+			zap.String("changefeed", m.cfg.ChangeFeedID.ID),
+			zap.Stringer("span", &span))
 		return
 	}
 }
@@ -408,6 +414,8 @@ func (m *logManager) postFlush(tableRtsMap *spanz.HashMap[model.Ts]) {
 			changed := value.(*statefulRts).checkAndSetFlushed(flushed)
 			if !changed {
 				log.Debug("flush redo with regressed resolved ts",
+					zap.String("namespace", m.cfg.ChangeFeedID.Namespace),
+					zap.String("changefeed", m.cfg.ChangeFeedID.ID),
 					zap.Stringer("span", &span),
 					zap.Uint64("flushed", flushed),
 					zap.Uint64("current", value.(*statefulRts).getFlushed()))
@@ -425,12 +433,15 @@ func (m *logManager) flushLog(
 		*workTimeSlice += time.Since(start)
 	}()
 	if !atomic.CompareAndSwapInt64(&m.flushing, 0, 1) {
-		log.Debug("Fail to update flush flag, " +
-			"the previous flush operation hasn't finished yet")
+		log.Debug("Fail to update flush flag, "+
+			"the previous flush operation hasn't finished yet",
+			zap.String("namespace", m.cfg.ChangeFeedID.Namespace),
+			zap.String("changefeed", m.cfg.ChangeFeedID.ID))
 		if time.Since(m.lastFlushTime) > redo.FlushWarnDuration {
 			log.Warn("flushLog blocking too long, the redo manager may be stuck",
-				zap.Duration("duration", time.Since(m.lastFlushTime)),
-				zap.Any("changfeed", m.cfg.ChangeFeedID))
+				zap.String("namespace", m.cfg.ChangeFeedID.Namespace),
+				zap.String("changefeed", m.cfg.ChangeFeedID.ID),
+				zap.Duration("duration", time.Since(m.lastFlushTime)))
 		}
 		return
 	}
diff --git a/cdc/redo/meta_manager.go b/cdc/redo/meta_manager.go
index f25f065a7..3ecde8a2a 100644
--- a/cdc/redo/meta_manager.go
+++ b/cdc/redo/meta_manager.go
@@ -106,6 +106,8 @@ func NewMetaManager(
 
 	if m.flushIntervalInMs < redo.MinFlushIntervalInMs {
 		log.Warn("redo flush interval is too small, use default value",
+			zap.String("namespace", m.changeFeedID.Namespace),
+			zap.String("changefeed", m.changeFeedID.ID),
 			zap.Int64("interval", m.flushIntervalInMs))
 		m.flushIntervalInMs = redo.DefaultMetaFlushIntervalInMs
 	}
@@ -263,6 +265,8 @@ func (m *metaManager) initMeta(ctx context.Context) error {
 	common.ParseMeta(metas, &checkpointTs, &resolvedTs)
 	if checkpointTs == 0 || resolvedTs == 0 {
 		log.Panic("checkpointTs or resolvedTs is 0 when initializing redo meta in owner",
+			zap.String("namespace", m.changeFeedID.Namespace),
+			zap.String("changefeed", m.changeFeedID.ID),
 			zap.Uint64("checkpointTs", checkpointTs),
 			zap.Uint64("resolvedTs", resolvedTs))
 	}
@@ -324,11 +328,17 @@ func (m *metaManager) shouldRemoved(path string, checkPointTs uint64) bool {
 
 	commitTs, fileType, err := redo.ParseLogFileName(path)
 	if err != nil {
-		log.Error("parse file name failed", zap.String("path", path), zap.Error(err))
+		log.Error("parse file name failed",
+			zap.String("namespace", m.changeFeedID.Namespace),
+			zap.String("changefeed", m.changeFeedID.ID),
+			zap.String("path", path), zap.Error(err))
 		return false
 	}
 	if fileType != redo.RedoDDLLogFileType && fileType != redo.RedoRowLogFileType {
-		log.Panic("unknown file type", zap.String("path", path), zap.Any("fileType", fileType))
+		log.Panic("unknown file type",
+			zap.String("namespace", m.changeFeedID.Namespace),
+			zap.String("changefeed", m.changeFeedID.ID),
+			zap.String("path", path), zap.Any("fileType", fileType))
 	}
 
 	// if commitTs == checkPointTs, the DDL may be executed in the owner,
@@ -450,6 +460,8 @@ func (m *metaManager) flush(ctx context.Context, meta common.LogMeta) error {
 	m.preMetaFile = metaFile
 
 	log.Debug("flush meta to s3",
+		zap.String("namespace", m.changeFeedID.Namespace),
+		zap.String("changefeed", m.changeFeedID.ID),
 		zap.String("metaFile", metaFile),
 		zap.Any("cost", time.Since(start).Milliseconds()))
 	m.metricFlushLogDuration.Observe(time.Since(start).Seconds())
diff --git a/cdc/redo/writer/memory/encoding_worker.go b/cdc/redo/writer/memory/encoding_worker.go
index bd878f65b..07e9824d1 100644
--- a/cdc/redo/writer/memory/encoding_worker.go
+++ b/cdc/redo/writer/memory/encoding_worker.go
@@ -98,6 +98,7 @@ func (e *polymorphicRedoEvent) encode() (err error) {
 }
 
 type encodingWorkerGroup struct {
+	changefeed model.ChangeFeedID
 	outputCh   chan *polymorphicRedoEvent
 	inputChs   []chan *polymorphicRedoEvent
 	workerNum  int
@@ -106,7 +107,8 @@ type encodingWorkerGroup struct {
 	closed chan struct{}
 }
 
-func newEncodingWorkerGroup(workerNum int) *encodingWorkerGroup {
+func newEncodingWorkerGroup(cfg *writer.LogWriterConfig) *encodingWorkerGroup {
+	workerNum := cfg.EncodingWorkerNum
 	if workerNum <= 0 {
 		workerNum = redo.DefaultEncodingWorkerNum
 	}
@@ -115,10 +117,11 @@ func newEncodingWorkerGroup(workerNum int) *encodingWorkerGroup {
 		inputChs[i] = make(chan *polymorphicRedoEvent, redo.DefaultEncodingInputChanSize)
 	}
 	return &encodingWorkerGroup{
-		inputChs:  inputChs,
-		outputCh:  make(chan *polymorphicRedoEvent, redo.DefaultEncodingOutputChanSize),
-		workerNum: workerNum,
-		closed:    make(chan struct{}),
+		changefeed: cfg.ChangeFeedID,
+		inputChs:   inputChs,
+		outputCh:   make(chan *polymorphicRedoEvent, redo.DefaultEncodingOutputChanSize),
+		workerNum:  workerNum,
+		closed:     make(chan struct{}),
 	}
 }
 
@@ -126,7 +129,10 @@ func (e *encodingWorkerGroup) Run(ctx context.Context) (err error) {
 	defer func() {
 		close(e.closed)
 		if err != nil && errors.Cause(err) != context.Canceled {
-			log.Warn("redo fileWorkerGroup closed with error", zap.Error(err))
+			log.Warn("redo fileWorkerGroup closed with error",
+				zap.String("namespace", e.changefeed.Namespace),
+				zap.String("changefeed", e.changefeed.ID),
+				zap.Error(err))
 		}
 	}()
 	eg, egCtx := errgroup.WithContext(ctx)
@@ -136,7 +142,10 @@ func (e *encodingWorkerGroup) Run(ctx context.Context) (err error) {
 			return e.runWorker(egCtx, idx)
 		})
 	}
-	log.Info("redo log encoding workers started", zap.Int("workerNum", e.workerNum))
+	log.Info("redo log encoding workers started",
+		zap.String("namespace", e.changefeed.Namespace),
+		zap.String("changefeed", e.changefeed.ID),
+		zap.Int("workerNum", e.workerNum))
 	return eg.Wait()
 }
 
diff --git a/cdc/redo/writer/memory/file_worker.go b/cdc/redo/writer/memory/file_worker.go
index 0816f2b0d..d566ddf66 100644
--- a/cdc/redo/writer/memory/file_worker.go
+++ b/cdc/redo/writer/memory/file_worker.go
@@ -156,7 +156,10 @@ func (f *fileWorkerGroup) Run(
 	defer func() {
 		f.close()
 		if err != nil && errors.Cause(err) != context.Canceled {
-			log.Warn("redo file workers closed with error", zap.Error(err))
+			log.Warn("redo file workers closed with error",
+				zap.String("namespace", f.cfg.ChangeFeedID.Namespace),
+				zap.String("changefeed", f.cfg.ChangeFeedID.ID),
+				zap.Error(err))
 		}
 	}()
 
@@ -169,7 +172,10 @@ func (f *fileWorkerGroup) Run(
 			return f.bgFlushFileCache(egCtx)
 		})
 	}
-	log.Info("redo file workers started", zap.Int("workerNum", f.workerNum))
+	log.Info("redo file workers started",
+		zap.String("namespace", f.cfg.ChangeFeedID.Namespace),
+		zap.String("changefeed", f.cfg.ChangeFeedID.ID),
+		zap.Int("workerNum", f.workerNum))
 	return eg.Wait()
 }
 
diff --git a/cdc/redo/writer/memory/mem_log_writer.go b/cdc/redo/writer/memory/mem_log_writer.go
index 25e968c51..32b5076ed 100644
--- a/cdc/redo/writer/memory/mem_log_writer.go
+++ b/cdc/redo/writer/memory/mem_log_writer.go
@@ -62,7 +62,7 @@ func NewLogWriter(
 		cancel: lwCancel,
 	}
 
-	lw.encodeWorkers = newEncodingWorkerGroup(cfg.EncodingWorkerNum)
+	lw.encodeWorkers = newEncodingWorkerGroup(cfg)
 	eg.Go(func() error {
 		return lw.encodeWorkers.Run(lwCtx)
 	})
@@ -78,6 +78,8 @@ func (l *memoryLogWriter) WriteEvents(ctx context.Context, events ...writer.Redo
 	for _, event := range events {
 		if event == nil {
 			log.Warn("writing nil event to redo log, ignore this",
+				zap.String("namespace", l.cfg.ChangeFeedID.Namespace),
+				zap.String("changefeed", l.cfg.ChangeFeedID.ID),
 				zap.String("capture", l.cfg.CaptureID))
 			continue
 		}
diff --git a/cdc/scheduler/internal/scheduler.go b/cdc/scheduler/internal/scheduler.go
index 111911b56..28e241b9d 100644
--- a/cdc/scheduler/internal/scheduler.go
+++ b/cdc/scheduler/internal/scheduler.go
@@ -47,7 +47,7 @@ type Scheduler interface {
 		// ddl jobs that need to be replicated. The Scheduler will
 		// broadcast the barrierTs to all captures through the Heartbeat.
 		barrier *schedulepb.BarrierWithMinTs,
-	) (newCheckpointTs, newResolvedTs model.Ts, err error)
+	) (watermark schedulepb.Watermark, err error)
 
 	// MoveTable requests that a table be moved to target.
 	// It is thread-safe.
diff --git a/cdc/scheduler/internal/v3/coordinator.go b/cdc/scheduler/internal/v3/coordinator.go
index 2a98c7be9..cb184cdf6 100644
--- a/cdc/scheduler/internal/v3/coordinator.go
+++ b/cdc/scheduler/internal/v3/coordinator.go
@@ -127,7 +127,7 @@ func (c *coordinator) Tick(
 	// All captures that are alive according to the latest Etcd states.
 	aliveCaptures map[model.CaptureID]*model.CaptureInfo,
 	barrier *schedulepb.BarrierWithMinTs,
-) (newCheckpointTs, newResolvedTs model.Ts, err error) {
+) (watermark schedulepb.Watermark, err error) {
 	startTime := time.Now()
 	defer func() {
 		costTime := time.Since(startTime)
@@ -273,7 +273,7 @@ func (c *coordinator) poll(
 	currentTables []model.TableID,
 	aliveCaptures map[model.CaptureID]*model.CaptureInfo,
 	barrier *schedulepb.BarrierWithMinTs,
-) (newCheckpointTs, newResolvedTs model.Ts, err error) {
+) (watermark schedulepb.Watermark, err error) {
 	c.maybeCollectMetrics()
 	if c.compat.UpdateCaptureInfo(aliveCaptures) {
 		spanReplicationEnabled := c.compat.CheckSpanReplicationEnabled()
@@ -284,7 +284,12 @@ func (c *coordinator) poll(
 
 	recvMsgs, err := c.recvMsgs(ctx)
 	if err != nil {
-		return checkpointCannotProceed, checkpointCannotProceed, errors.Trace(err)
+		return schedulepb.Watermark{
+			CheckpointTs:     checkpointCannotProceed,
+			ResolvedTs:       checkpointCannotProceed,
+			LastSyncedTs:     checkpointCannotProceed,
+			PullerResolvedTs: checkpointCannotProceed,
+		}, errors.Trace(err)
 	}
 
 	var msgBuf []*schedulepb.Message
@@ -296,7 +301,12 @@ func (c *coordinator) poll(
 	// Handle received messages to advance replication set.
 	msgs, err = c.replicationM.HandleMessage(recvMsgs)
 	if err != nil {
-		return checkpointCannotProceed, checkpointCannotProceed, errors.Trace(err)
+		return schedulepb.Watermark{
+			CheckpointTs:     checkpointCannotProceed,
+			ResolvedTs:       checkpointCannotProceed,
+			LastSyncedTs:     checkpointCannotProceed,
+			PullerResolvedTs: checkpointCannotProceed,
+		}, errors.Trace(err)
 	}
 	msgBuf = append(msgBuf, msgs...)
 
@@ -313,13 +323,13 @@ func (c *coordinator) poll(
 	if !c.captureM.CheckAllCaptureInitialized() {
 		// Skip generating schedule tasks for replication manager,
 		// as not all capture are initialized.
-		newCheckpointTs, newResolvedTs = c.replicationM.AdvanceCheckpoint(&c.tableRanges, pdTime, barrier, c.redoMetaManager)
+		watermark = c.replicationM.AdvanceCheckpoint(&c.tableRanges, pdTime, barrier, c.redoMetaManager)
 		// tick capture manager after checkpoint calculation to take account resolvedTs in barrier
 		// when redo is enabled
 		msgs = c.captureM.Tick(c.replicationM.ReplicationSets(),
 			c.schedulerM.DrainingTarget(), barrier.Barrier)
 		msgBuf = append(msgBuf, msgs...)
-		return newCheckpointTs, newResolvedTs, c.sendMsgs(ctx, msgBuf)
+		return watermark, c.sendMsgs(ctx, msgBuf)
 	}
 
 	// Handle capture membership changes.
@@ -327,7 +337,12 @@ func (c *coordinator) poll(
 		msgs, err = c.replicationM.HandleCaptureChanges(
 			changes.Init, changes.Removed, checkpointTs)
 		if err != nil {
-			return checkpointCannotProceed, checkpointCannotProceed, errors.Trace(err)
+			return schedulepb.Watermark{
+				CheckpointTs:     checkpointCannotProceed,
+				ResolvedTs:       checkpointCannotProceed,
+				LastSyncedTs:     checkpointCannotProceed,
+				PullerResolvedTs: checkpointCannotProceed,
+			}, errors.Trace(err)
 		}
 		msgBuf = append(msgBuf, msgs...)
 	}
@@ -343,12 +358,17 @@ func (c *coordinator) poll(
 	// Handle generated schedule tasks.
 	msgs, err = c.replicationM.HandleTasks(allTasks)
 	if err != nil {
-		return checkpointCannotProceed, checkpointCannotProceed, errors.Trace(err)
+		return schedulepb.Watermark{
+			CheckpointTs:     checkpointCannotProceed,
+			ResolvedTs:       checkpointCannotProceed,
+			LastSyncedTs:     checkpointCannotProceed,
+			PullerResolvedTs: checkpointCannotProceed,
+		}, errors.Trace(err)
 	}
 	msgBuf = append(msgBuf, msgs...)
 
 	// Checkpoint calculation
-	newCheckpointTs, newResolvedTs = c.replicationM.AdvanceCheckpoint(&c.tableRanges, pdTime, barrier, c.redoMetaManager)
+	watermark = c.replicationM.AdvanceCheckpoint(&c.tableRanges, pdTime, barrier, c.redoMetaManager)
 
 	// tick capture manager after checkpoint calculation to take account resolvedTs in barrier
 	// when redo is enabled
@@ -359,10 +379,15 @@ func (c *coordinator) poll(
 	// Send new messages.
 	err = c.sendMsgs(ctx, msgBuf)
 	if err != nil {
-		return checkpointCannotProceed, checkpointCannotProceed, errors.Trace(err)
+		return schedulepb.Watermark{
+			CheckpointTs:     checkpointCannotProceed,
+			ResolvedTs:       checkpointCannotProceed,
+			LastSyncedTs:     checkpointCannotProceed,
+			PullerResolvedTs: checkpointCannotProceed,
+		}, errors.Trace(err)
 	}
 
-	return newCheckpointTs, newResolvedTs, nil
+	return watermark, nil
 }
 
 func (c *coordinator) recvMsgs(ctx context.Context) ([]*schedulepb.Message, error) {
diff --git a/cdc/scheduler/internal/v3/replication/replication_manager.go b/cdc/scheduler/internal/v3/replication/replication_manager.go
index 4228cc0ec..8196a06a0 100644
--- a/cdc/scheduler/internal/v3/replication/replication_manager.go
+++ b/cdc/scheduler/internal/v3/replication/replication_manager.go
@@ -557,9 +557,9 @@ func (r *Manager) AdvanceCheckpoint(
 	currentPDTime time.Time,
 	barrier *schedulepb.BarrierWithMinTs,
 	redoMetaManager redo.MetaManager,
-) (newCheckpointTs, newResolvedTs model.Ts) {
+) (watermark schedulepb.Watermark) {
 	var redoFlushedResolvedTs model.Ts
-	limitBarrierWithRedo := func(newCheckpointTs, newResolvedTs uint64) (uint64, uint64) {
+	limitBarrierWithRedo := func(watermark *schedulepb.Watermark) {
 		flushedMeta := redoMetaManager.GetFlushedMeta()
 		redoFlushedResolvedTs = flushedMeta.ResolvedTs
 		log.Debug("owner gets flushed redo meta",
@@ -567,18 +567,17 @@ func (r *Manager) AdvanceCheckpoint(
 			zap.String("changefeed", r.changefeedID.ID),
 			zap.Uint64("flushedCheckpointTs", flushedMeta.CheckpointTs),
 			zap.Uint64("flushedResolvedTs", flushedMeta.ResolvedTs))
-		if flushedMeta.ResolvedTs < newResolvedTs {
-			newResolvedTs = flushedMeta.ResolvedTs
+		if flushedMeta.ResolvedTs < watermark.ResolvedTs {
+			watermark.ResolvedTs = flushedMeta.ResolvedTs
 		}
 
-		if newCheckpointTs > newResolvedTs {
-			newCheckpointTs = newResolvedTs
+		if watermark.CheckpointTs > watermark.ResolvedTs {
+			watermark.CheckpointTs = watermark.ResolvedTs
 		}
 
-		if barrier.GlobalBarrierTs > newResolvedTs {
-			barrier.GlobalBarrierTs = newResolvedTs
+		if barrier.GlobalBarrierTs > watermark.ResolvedTs {
+			barrier.GlobalBarrierTs = watermark.ResolvedTs
 		}
-		return newCheckpointTs, newResolvedTs
 	}
 	defer func() {
 		if redoFlushedResolvedTs != 0 && barrier.GlobalBarrierTs > redoFlushedResolvedTs {
@@ -592,9 +591,14 @@ func (r *Manager) AdvanceCheckpoint(
 
 	r.slowestPuller = tablepb.Span{}
 	r.slowestSink = tablepb.Span{}
-	var slowestPullerResolvedTs uint64 = math.MaxUint64
 
-	newCheckpointTs, newResolvedTs = math.MaxUint64, math.MaxUint64
+	watermark = schedulepb.Watermark{
+		CheckpointTs:     math.MaxUint64,
+		ResolvedTs:       math.MaxUint64,
+		LastSyncedTs:     0,
+		PullerResolvedTs: math.MaxUint64,
+	}
+
 	cannotProceed := false
 	currentTables.Iter(func(tableID model.TableID, tableStart, tableEnd tablepb.Span) bool {
 		tableSpanFound, tableHasHole := false, false
@@ -621,21 +625,26 @@ func (r *Manager) AdvanceCheckpoint(
 				}
 
 				// Find the minimum checkpoint ts and resolved ts.
-				if newCheckpointTs > table.Checkpoint.CheckpointTs {
-					newCheckpointTs = table.Checkpoint.CheckpointTs
+				if watermark.CheckpointTs > table.Checkpoint.CheckpointTs {
+					watermark.CheckpointTs = table.Checkpoint.CheckpointTs
 					r.slowestSink = span
 				}
-				if newResolvedTs > table.Checkpoint.ResolvedTs {
-					newResolvedTs = table.Checkpoint.ResolvedTs
+				if watermark.ResolvedTs > table.Checkpoint.ResolvedTs {
+					watermark.ResolvedTs = table.Checkpoint.ResolvedTs
 				}
 
+				// Find the max lastSyncedTs of all tables.
+				if watermark.LastSyncedTs < table.Checkpoint.LastSyncedTs {
+					watermark.LastSyncedTs = table.Checkpoint.LastSyncedTs
+				}
 				// Find the minimum puller resolved ts.
 				if pullerCkpt, ok := table.Stats.StageCheckpoints["puller-egress"]; ok {
-					if slowestPullerResolvedTs > pullerCkpt.ResolvedTs {
-						slowestPullerResolvedTs = pullerCkpt.ResolvedTs
+					if watermark.PullerResolvedTs > pullerCkpt.ResolvedTs {
+						watermark.PullerResolvedTs = pullerCkpt.ResolvedTs
 						r.slowestPuller = span
 					}
 				}
+
 				return true
 			})
 		if !tableSpanFound || !tableSpanStartFound || !tableSpanEndFound || tableHasHole {
@@ -663,32 +672,39 @@ func (r *Manager) AdvanceCheckpoint(
 	if cannotProceed {
 		if redoMetaManager.Enabled() {
 			// If redo is enabled, GlobalBarrierTs should be limited by redo flushed meta.
-			newResolvedTs = barrier.RedoBarrierTs
-			limitBarrierWithRedo(newCheckpointTs, newResolvedTs)
+			watermark.ResolvedTs = barrier.RedoBarrierTs
+			watermark.LastSyncedTs = checkpointCannotProceed
+			watermark.PullerResolvedTs = checkpointCannotProceed
+			limitBarrierWithRedo(&watermark)
+		}
+		return schedulepb.Watermark{
+			CheckpointTs:     checkpointCannotProceed,
+			ResolvedTs:       checkpointCannotProceed,
+			LastSyncedTs:     checkpointCannotProceed,
+			PullerResolvedTs: checkpointCannotProceed,
 		}
-		return checkpointCannotProceed, checkpointCannotProceed
 	}
 
 	// If currentTables is empty, we should advance newResolvedTs to global barrier ts and
 	// advance newCheckpointTs to min table barrier ts.
-	if newResolvedTs == math.MaxUint64 || newCheckpointTs == math.MaxUint64 {
-		if newCheckpointTs != newResolvedTs || currentTables.Len() != 0 {
+	if watermark.ResolvedTs == math.MaxUint64 || watermark.CheckpointTs == math.MaxUint64 {
+		if watermark.CheckpointTs != watermark.ResolvedTs || currentTables.Len() != 0 {
 			log.Panic("schedulerv3: newCheckpointTs and newResolvedTs should be both maxUint64 "+
 				"if currentTables is empty",
-				zap.Uint64("newCheckpointTs", newCheckpointTs),
-				zap.Uint64("newResolvedTs", newResolvedTs),
+				zap.Uint64("newCheckpointTs", watermark.CheckpointTs),
+				zap.Uint64("newResolvedTs", watermark.ResolvedTs),
 				zap.Any("currentTables", currentTables))
 		}
-		newResolvedTs = barrier.GlobalBarrierTs
-		newCheckpointTs = barrier.MinTableBarrierTs
+		watermark.ResolvedTs = barrier.GlobalBarrierTs
+		watermark.CheckpointTs = barrier.MinTableBarrierTs
 	}
 
-	if newCheckpointTs > barrier.MinTableBarrierTs {
-		newCheckpointTs = barrier.MinTableBarrierTs
+	if watermark.CheckpointTs > barrier.MinTableBarrierTs {
+		watermark.CheckpointTs = barrier.MinTableBarrierTs
 		// TODO: add panic after we fix the bug that newCheckpointTs > minTableBarrierTs.
 		// log.Panic("schedulerv3: newCheckpointTs should not be larger than minTableBarrierTs",
-		// 	zap.Uint64("newCheckpointTs", newCheckpointTs),
-		// 	zap.Uint64("newResolvedTs", newResolvedTs),
+		// 	zap.Uint64("newCheckpointTs", watermark.CheckpointTs),
+		// 	zap.Uint64("newResolvedTs", watermark.ResolvedTs),
 		// 	zap.Any("currentTables", currentTables.currentTables),
 		// 	zap.Any("barrier", barrier.Barrier),
 		// 	zap.Any("minTableBarrierTs", barrier.MinTableBarrierTs))
@@ -697,7 +713,7 @@ func (r *Manager) AdvanceCheckpoint(
 	// If changefeed's checkpoint lag is larger than 30s,
 	// log the 4 slowlest table infos every minute, which can
 	// help us find the problematic tables.
-	checkpointLag := currentPDTime.Sub(oracle.GetTimeFromTS(newCheckpointTs))
+	checkpointLag := currentPDTime.Sub(oracle.GetTimeFromTS(watermark.CheckpointTs))
 	if checkpointLag > logSlowTablesLagThreshold &&
 		time.Since(r.lastLogSlowTablesTime) > logSlowTablesInterval {
 		r.logSlowTableInfo(currentPDTime)
@@ -705,19 +721,19 @@ func (r *Manager) AdvanceCheckpoint(
 	}
 
 	if redoMetaManager.Enabled() {
-		if newResolvedTs > barrier.RedoBarrierTs {
-			newResolvedTs = barrier.RedoBarrierTs
+		if watermark.ResolvedTs > barrier.RedoBarrierTs {
+			watermark.ResolvedTs = barrier.RedoBarrierTs
 		}
-		redoMetaManager.UpdateMeta(newCheckpointTs, newResolvedTs)
+		redoMetaManager.UpdateMeta(watermark.CheckpointTs, watermark.ResolvedTs)
 		log.Debug("owner updates redo meta",
 			zap.String("namespace", r.changefeedID.Namespace),
 			zap.String("changefeed", r.changefeedID.ID),
-			zap.Uint64("newCheckpointTs", newCheckpointTs),
-			zap.Uint64("newResolvedTs", newResolvedTs))
-		return limitBarrierWithRedo(newCheckpointTs, newResolvedTs)
+			zap.Uint64("newCheckpointTs", watermark.CheckpointTs),
+			zap.Uint64("newResolvedTs", watermark.ResolvedTs))
+		limitBarrierWithRedo(&watermark)
 	}
 
-	return newCheckpointTs, newResolvedTs
+	return watermark
 }
 
 func (r *Manager) logSlowTableInfo(currentPDTime time.Time) {
diff --git a/cdc/scheduler/internal/v3/replication/replication_set.go b/cdc/scheduler/internal/v3/replication/replication_set.go
index eab841b24..158fe5271 100644
--- a/cdc/scheduler/internal/v3/replication/replication_set.go
+++ b/cdc/scheduler/internal/v3/replication/replication_set.go
@@ -1021,6 +1021,10 @@ func (r *ReplicationSet) updateCheckpointAndStats(
 			zap.Any("resolvedTs", r.Checkpoint.ResolvedTs))
 	}
 
+	if r.Checkpoint.LastSyncedTs < checkpoint.LastSyncedTs {
+		r.Checkpoint.LastSyncedTs = checkpoint.LastSyncedTs
+	}
+
 	// we only update stats when stats is not empty, because we only collect stats every 10s.
 	if stats.Size() > 0 {
 		r.Stats = stats
diff --git a/cdc/scheduler/schedulepb/watermark.go b/cdc/scheduler/schedulepb/watermark.go
new file mode 100644
index 000000000..81123734e
--- /dev/null
+++ b/cdc/scheduler/schedulepb/watermark.go
@@ -0,0 +1,24 @@
+// Copyright 2023 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package schedulepb
+
+import "github.com/pingcap/tiflow/cdc/model"
+
+// Watermark contains various ts variables to make code easier
+type Watermark struct {
+	CheckpointTs     model.Ts
+	ResolvedTs       model.Ts
+	LastSyncedTs     model.Ts
+	PullerResolvedTs model.Ts
+}
diff --git a/cdc/sink/ddlsink/mq/kafka_ddl_sink.go b/cdc/sink/ddlsink/mq/kafka_ddl_sink.go
index e38b5f07d..853d076fd 100644
--- a/cdc/sink/ddlsink/mq/kafka_ddl_sink.go
+++ b/cdc/sink/ddlsink/mq/kafka_ddl_sink.go
@@ -86,7 +86,7 @@ func NewKafkaDDLSink(
 		return nil, errors.Trace(err)
 	}
 
-	eventRouter, err := dispatcher.NewEventRouter(replicaConfig, topic)
+	eventRouter, err := dispatcher.NewEventRouter(replicaConfig, topic, protocol)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
diff --git a/cdc/sink/dmlsink/cloudstorage/dml_worker.go b/cdc/sink/dmlsink/cloudstorage/dml_worker.go
index 1d0174cc1..07e753c6c 100644
--- a/cdc/sink/dmlsink/cloudstorage/dml_worker.go
+++ b/cdc/sink/dmlsink/cloudstorage/dml_worker.go
@@ -230,23 +230,24 @@ func (d *dmlWorker) writeDataFile(ctx context.Context, path string, task *single
 	var callbacks []func()
 	buf := bytes.NewBuffer(make([]byte, 0, task.size))
 	rowsCnt := 0
+	bytesCnt := int64(0)
 	for _, msg := range task.msgs {
-		d.metricWriteBytes.Add(float64(len(msg.Value)))
+		bytesCnt += int64(len(msg.Value))
 		rowsCnt += msg.GetRowsCount()
 		buf.Write(msg.Value)
 		callbacks = append(callbacks, msg.Callback)
 	}
 
-	if err := d.statistics.RecordBatchExecution(func() (int, error) {
+	if err := d.statistics.RecordBatchExecution(func() (int, int64, error) {
 		if d.config.FlushConcurrency <= 1 {
-			return rowsCnt, d.storage.WriteFile(ctx, path, buf.Bytes())
+			return rowsCnt, bytesCnt, d.storage.WriteFile(ctx, path, buf.Bytes())
 		}
 
 		writer, inErr := d.storage.Create(ctx, path, &storage.WriterOption{
 			Concurrency: d.config.FlushConcurrency,
 		})
 		if inErr != nil {
-			return 0, inErr
+			return 0, 0, inErr
 		}
 
 		defer func() {
@@ -263,13 +264,14 @@ func (d *dmlWorker) writeDataFile(ctx context.Context, path string, task *single
 			}
 		}()
 		if _, inErr = writer.Write(ctx, buf.Bytes()); inErr != nil {
-			return 0, inErr
+			return 0, 0, inErr
 		}
-		return rowsCnt, nil
+		return rowsCnt, bytesCnt, nil
 	}); err != nil {
 		return err
 	}
 
+	d.metricWriteBytes.Add(float64(bytesCnt))
 	d.metricFileCount.Add(1)
 	for _, cb := range callbacks {
 		if cb != nil {
diff --git a/cdc/sink/dmlsink/mq/dispatcher/event_router.go b/cdc/sink/dmlsink/mq/dispatcher/event_router.go
index 7e6146b5e..5366f719e 100644
--- a/cdc/sink/dmlsink/mq/dispatcher/event_router.go
+++ b/cdc/sink/dmlsink/mq/dispatcher/event_router.go
@@ -78,7 +78,7 @@ type EventRouter struct {
 }
 
 // NewEventRouter creates a new EventRouter.
-func NewEventRouter(cfg *config.ReplicaConfig, defaultTopic string) (*EventRouter, error) {
+func NewEventRouter(cfg *config.ReplicaConfig, defaultTopic string, protocol config.Protocol) (*EventRouter, error) {
 	// If an event does not match any dispatching rules in the config file,
 	// it will be dispatched by the default partition dispatcher and
 	// static topic dispatcher because it matches *.* rule.
@@ -103,7 +103,7 @@ func NewEventRouter(cfg *config.ReplicaConfig, defaultTopic string) (*EventRoute
 		}
 
 		d := getPartitionDispatcher(ruleConfig, cfg.EnableOldValue)
-		t, err := getTopicDispatcher(ruleConfig, defaultTopic, cfg.Sink.Protocol)
+		t, err := getTopicDispatcher(ruleConfig.TopicRule, defaultTopic, protocol)
 		if err != nil {
 			return nil, err
 		}
@@ -251,32 +251,31 @@ func getPartitionDispatcher(
 
 // getTopicDispatcher returns the topic dispatcher for a specific topic rule (aka topic expression).
 func getTopicDispatcher(
-	ruleConfig *config.DispatchRule, defaultTopic string, protocol string,
+	rule string, defaultTopic string, protocol config.Protocol,
 ) (topic.Dispatcher, error) {
-	if ruleConfig.TopicRule == "" {
+	if rule == "" {
 		return topic.NewStaticTopicDispatcher(defaultTopic), nil
 	}
 
-	// check if this rule is a valid topic expression
-	topicExpr := topic.Expression(ruleConfig.TopicRule)
-
-	if protocol != "" {
-		p, err := config.ParseSinkProtocolFromString(protocol)
-		if err != nil {
-			return nil, cerror.WrapError(cerror.ErrKafkaInvalidConfig, err)
-		}
+	if topic.IsHardCode(rule) {
+		return topic.NewStaticTopicDispatcher(rule), nil
+	}
 
-		if p == config.ProtocolAvro {
-			err := topicExpr.ValidateForAvro()
-			if err != nil {
-				return nil, err
-			}
-		} else {
-			err := topicExpr.Validate()
-			if err != nil {
-				return nil, err
-			}
-		}
+	// check if this rule is a valid topic expression
+	topicExpr := topic.Expression(rule)
+	err := validateTopicExpression(topicExpr, protocol)
+	if err != nil {
+		return nil, err
 	}
 	return topic.NewDynamicTopicDispatcher(topicExpr), nil
 }
+
+func validateTopicExpression(expr topic.Expression, protocol config.Protocol) error {
+	switch protocol {
+	case config.ProtocolAvro:
+		return expr.ValidateForAvro()
+	default:
+	}
+
+	return expr.Validate()
+}
diff --git a/cdc/sink/dmlsink/mq/dispatcher/topic/dispatcher.go b/cdc/sink/dmlsink/mq/dispatcher/topic/dispatcher.go
index 90a543b2b..4053222d5 100644
--- a/cdc/sink/dmlsink/mq/dispatcher/topic/dispatcher.go
+++ b/cdc/sink/dmlsink/mq/dispatcher/topic/dispatcher.go
@@ -23,28 +23,28 @@ type Dispatcher interface {
 	Substitute(schema, table string) string
 }
 
-// StaticTopicDispatcher is a topic dispatcher which dispatches rows and ddls to the default topic.
+// StaticTopicDispatcher is a topic dispatcher which dispatches rows and DDL to the specific topic.
 type StaticTopicDispatcher struct {
-	defaultTopic string
+	topic string
 }
 
 // NewStaticTopicDispatcher returns a StaticTopicDispatcher.
 func NewStaticTopicDispatcher(defaultTopic string) *StaticTopicDispatcher {
 	return &StaticTopicDispatcher{
-		defaultTopic: defaultTopic,
+		topic: defaultTopic,
 	}
 }
 
 // Substitute converts schema/table name in a topic expression to kafka topic name.
 func (s *StaticTopicDispatcher) Substitute(schema, table string) string {
-	return s.defaultTopic
+	return s.topic
 }
 
 func (s *StaticTopicDispatcher) String() string {
-	return s.defaultTopic
+	return s.topic
 }
 
-// DynamicTopicDispatcher is a topic dispatcher which dispatches rows and ddls
+// DynamicTopicDispatcher is a topic dispatcher which dispatches rows and DDLs
 // dynamically to the target topics.
 type DynamicTopicDispatcher struct {
 	expression Expression
diff --git a/cdc/sink/dmlsink/mq/dispatcher/topic/expression.go b/cdc/sink/dmlsink/mq/dispatcher/topic/expression.go
index a6dde858a..e8f3e279e 100644
--- a/cdc/sink/dmlsink/mq/dispatcher/topic/expression.go
+++ b/cdc/sink/dmlsink/mq/dispatcher/topic/expression.go
@@ -20,9 +20,12 @@ import (
 )
 
 var (
+	// hardCodeTopicNameRe is used to match a topic name which is hard code in the config
+	hardCodeTopicNameRe = regexp.MustCompile(`^([A-Za-z0-9\._\-]+)$`)
+
 	// topicNameRE is used to match a valid topic expression
 	topicNameRE = regexp.MustCompile(
-		`^[A-Za-z0-9\._\-]*\{schema\}([A-Za-z0-9\._\-]*\{table\})?[A-Za-z0-9\._\-]*$`,
+		`^[A-Za-z0-9\._\-]*(\{schema\})?([A-Za-z0-9\._\-]*\{table\})?[A-Za-z0-9\._\-]*$`,
 	)
 	// kafkaForbidRE is used to reject the characters which are forbidden in kafka topic name
 	kafkaForbidRE = regexp.MustCompile(`[^a-zA-Z0-9\._\-]`)
@@ -48,13 +51,14 @@ const kafkaTopicNameMaxLength = 249
 type Expression string
 
 // Validate checks whether a kafka topic name is valid or not.
+// return true if the expression is hard coded.
 func (e Expression) Validate() error {
 	// validate the topic expression
-	if ok := topicNameRE.MatchString(string(e)); !ok {
-		return errors.ErrKafkaInvalidTopicExpression.GenWithStackByArgs()
+	if ok := topicNameRE.MatchString(string(e)); ok {
+		return nil
 	}
 
-	return nil
+	return errors.ErrKafkaInvalidTopicExpression.GenWithStackByArgs()
 }
 
 // ValidateForAvro checks whether topic pattern is {schema}_{table}, the only allowed
@@ -94,3 +98,8 @@ func (e Expression) Substitute(schema, table string) string {
 		return topicName
 	}
 }
+
+// IsHardCode checks whether a topic name is hard code or not.
+func IsHardCode(topicName string) bool {
+	return hardCodeTopicNameRe.MatchString(topicName)
+}
diff --git a/cdc/sink/dmlsink/mq/kafka_dml_sink.go b/cdc/sink/dmlsink/mq/kafka_dml_sink.go
index 23aff7dfe..69075fe3f 100644
--- a/cdc/sink/dmlsink/mq/kafka_dml_sink.go
+++ b/cdc/sink/dmlsink/mq/kafka_dml_sink.go
@@ -88,7 +88,7 @@ func NewKafkaDMLSink(
 		return nil, errors.Trace(err)
 	}
 
-	eventRouter, err := dispatcher.NewEventRouter(replicaConfig, topic)
+	eventRouter, err := dispatcher.NewEventRouter(replicaConfig, topic, protocol)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
diff --git a/cdc/sink/dmlsink/mq/worker.go b/cdc/sink/dmlsink/mq/worker.go
index 4e772770b..6f64e577b 100644
--- a/cdc/sink/dmlsink/mq/worker.go
+++ b/cdc/sink/dmlsink/mq/worker.go
@@ -296,11 +296,11 @@ func (w *worker) sendMessages(ctx context.Context) error {
 			}
 			for _, message := range future.Messages {
 				start := time.Now()
-				if err := w.statistics.RecordBatchExecution(func() (int, error) {
+				if err := w.statistics.RecordBatchExecution(func() (int, int64, error) {
 					if err := w.producer.AsyncSendMessage(ctx, future.Topic, future.Partition, message); err != nil {
-						return 0, err
+						return 0, 0, err
 					}
-					return message.GetRowsCount(), nil
+					return message.GetRowsCount(), int64(message.Length()), nil
 				}); err != nil {
 					return err
 				}
diff --git a/cdc/sink/dmlsink/txn/event.go b/cdc/sink/dmlsink/txn/event.go
index 18ea2090f..52dcf90e3 100644
--- a/cdc/sink/dmlsink/txn/event.go
+++ b/cdc/sink/dmlsink/txn/event.go
@@ -41,11 +41,44 @@ func (e *txnEvent) OnConflictResolved() {
 	e.conflictResolved = time.Now()
 }
 
-// ConflictKeys implements causality.txnEvent interface.
-func (e *txnEvent) ConflictKeys(numSlots uint64) []uint64 {
-	keys := genTxnKeys(e.TxnCallbackableEvent.Event)
-	sort.Slice(keys, func(i, j int) bool { return keys[i]%numSlots < keys[j]%numSlots })
-	return keys
+// GenSortedDedupKeysHash implements causality.txnEvent interface.
+func (e *txnEvent) GenSortedDedupKeysHash(numSlots uint64) []uint64 {
+	hashes := genTxnKeys(e.TxnCallbackableEvent.Event)
+
+	// Sort and dedup hashes.
+	// Sort hashes by `hash % numSlots` to avoid deadlock, and then dedup
+	// hashes, so the same txn will not check confict with the same hash twice to
+	// prevent potential cyclic self dependency in the causality dependency
+	// graph.
+	return sortAndDedupHashes(hashes, numSlots)
+}
+
+func sortAndDedupHashes(hashes []uint64, numSlots uint64) []uint64 {
+	if len(hashes) == 0 {
+		return nil
+	}
+
+	// Sort hashes by `hash % numSlots` to avoid deadlock.
+	sort.Slice(hashes, func(i, j int) bool { return hashes[i]%numSlots < hashes[j]%numSlots })
+
+	// Dedup hashes
+	last := hashes[0]
+	j := 1
+	for i, hash := range hashes {
+		if i == 0 {
+			// skip first one, start checking duplication from 2nd one
+			continue
+		}
+		if hash == last {
+			continue
+		}
+		last = hash
+		hashes[j] = hash
+		j++
+	}
+	hashes = hashes[:j]
+
+	return hashes
 }
 
 // genTxnKeys returns hash keys for `txn`.
diff --git a/cdc/sink/dmlsink/txn/mysql/mysql.go b/cdc/sink/dmlsink/txn/mysql/mysql.go
index 31d3b45bd..97d9bd071 100644
--- a/cdc/sink/dmlsink/txn/mysql/mysql.go
+++ b/cdc/sink/dmlsink/txn/mysql/mysql.go
@@ -171,7 +171,9 @@ func NewMySQLBackends(
 	var maxAllowedPacket int64
 	maxAllowedPacket, err = pmysql.QueryMaxAllowedPacket(ctx, db)
 	if err != nil {
-		log.Warn("failed to query max_allowed_packet, use default value", zap.Error(err))
+		log.Warn("failed to query max_allowed_packet, use default value",
+			zap.String("changefeed", changefeed),
+			zap.Error(err))
 		maxAllowedPacket = int64(variable.DefMaxAllowedPacket)
 	}
 
@@ -228,13 +230,13 @@ func (s *mysqlBackend) Flush(ctx context.Context) (err error) {
 	}
 
 	dmls := s.prepareDMLs()
-	log.Debug("prepare DMLs", zap.Any("rows", s.rows),
+	log.Debug("prepare DMLs", zap.String("changefeed", s.changefeed), zap.Any("rows", s.rows),
 		zap.Strings("sqls", dmls.sqls), zap.Any("values", dmls.values))
 
 	start := time.Now()
 	if err := s.execDMLWithMaxRetries(ctx, dmls); err != nil {
 		if errors.Cause(err) != context.Canceled {
-			log.Error("execute DMLs failed", zap.Error(err))
+			log.Error("execute DMLs failed", zap.String("changefeed", s.changefeed), zap.Error(err))
 		}
 		return errors.Trace(err)
 	}
@@ -554,6 +556,7 @@ func (s *mysqlBackend) prepareDMLs() *preparedDMLs {
 		// replicated before, and there is no such row in downstream MySQL.
 		translateToInsert = translateToInsert && firstRow.CommitTs > firstRow.ReplicatingTs
 		log.Debug("translate to insert",
+			zap.String("changefeed", s.changefeed),
 			zap.Bool("translateToInsert", translateToInsert),
 			zap.Uint64("firstRowCommitTs", firstRow.CommitTs),
 			zap.Uint64("firstRowReplicatingTs", firstRow.ReplicatingTs),
@@ -662,7 +665,7 @@ func (s *mysqlBackend) multiStmtExecute(
 	}
 	multiStmtSQL := strings.Join(dmls.sqls, ";")
 
-	log.Debug("exec row", zap.Int("workerID", s.workerID),
+	log.Debug("exec row", zap.String("changefeed", s.changefeed), zap.Int("workerID", s.workerID),
 		zap.String("sql", multiStmtSQL), zap.Any("args", multiStmtArgs))
 	ctx, cancel := context.WithTimeout(ctx, writeTimeout)
 	defer cancel()
@@ -674,7 +677,7 @@ func (s *mysqlBackend) multiStmtExecute(
 			start, s.changefeed, multiStmtSQL, dmls.rowCount, dmls.startTs)
 		if rbErr := tx.Rollback(); rbErr != nil {
 			if errors.Cause(rbErr) != context.Canceled {
-				log.Warn("failed to rollback txn", zap.Error(rbErr))
+				log.Warn("failed to rollback txn", zap.String("changefeed", s.changefeed), zap.Error(rbErr))
 			}
 		}
 		return err
@@ -689,7 +692,7 @@ func (s *mysqlBackend) sequenceExecute(
 	start := time.Now()
 	for i, query := range dmls.sqls {
 		args := dmls.values[i]
-		log.Debug("exec row", zap.Int("workerID", s.workerID),
+		log.Debug("exec row", zap.String("changefeed", s.changefeed), zap.Int("workerID", s.workerID),
 			zap.String("sql", query), zap.Any("args", args))
 		ctx, cancelFunc := context.WithTimeout(ctx, writeTimeout)
 
@@ -721,7 +724,7 @@ func (s *mysqlBackend) sequenceExecute(
 				start, s.changefeed, query, dmls.rowCount, dmls.startTs)
 			if rbErr := tx.Rollback(); rbErr != nil {
 				if errors.Cause(rbErr) != context.Canceled {
-					log.Warn("failed to rollback txn", zap.Error(rbErr))
+					log.Warn("failed to rollback txn", zap.String("changefeed", s.changefeed), zap.Error(rbErr))
 				}
 			}
 			cancelFunc()
@@ -735,6 +738,7 @@ func (s *mysqlBackend) sequenceExecute(
 func (s *mysqlBackend) execDMLWithMaxRetries(pctx context.Context, dmls *preparedDMLs) error {
 	if len(dmls.sqls) != len(dmls.values) {
 		log.Panic("unexpected number of sqls and values",
+			zap.String("changefeed", s.changefeed),
 			zap.Strings("sqls", dmls.sqls),
 			zap.Any("values", dmls.values))
 	}
@@ -754,10 +758,10 @@ func (s *mysqlBackend) execDMLWithMaxRetries(pctx context.Context, dmls *prepare
 		})
 		failpoint.Inject("MySQLSinkHangLongTime", func() { _ = util.Hang(pctx, time.Hour) })
 
-		err := s.statistics.RecordBatchExecution(func() (int, error) {
+		err := s.statistics.RecordBatchExecution(func() (int, int64, error) {
 			tx, err := s.db.BeginTx(pctx, nil)
 			if err != nil {
-				return 0, logDMLTxnErr(
+				return 0, 0, logDMLTxnErr(
 					cerror.WrapError(cerror.ErrMySQLTxnError, err),
 					start, s.changefeed, "BEGIN", dmls.rowCount, dmls.startTs)
 			}
@@ -771,12 +775,12 @@ func (s *mysqlBackend) execDMLWithMaxRetries(pctx context.Context, dmls *prepare
 				err = s.multiStmtExecute(pctx, dmls, tx, writeTimeout)
 				if err != nil {
 					fallbackToSeqWay = true
-					return 0, err
+					return 0, 0, err
 				}
 			} else {
 				err = s.sequenceExecute(pctx, dmls, tx, writeTimeout)
 				if err != nil {
-					return 0, err
+					return 0, 0, err
 				}
 			}
 
@@ -791,25 +795,25 @@ func (s *mysqlBackend) execDMLWithMaxRetries(pctx context.Context, dmls *prepare
 					dmls.rowCount, dmls.startTs)
 				if rbErr := tx.Rollback(); rbErr != nil {
 					if errors.Cause(rbErr) != context.Canceled {
-						log.Warn("failed to rollback txn", zap.Error(rbErr))
+						log.Warn("failed to rollback txn", zap.String("changefeed", s.changefeed), zap.Error(rbErr))
 					}
 				}
-				return 0, err
+				return 0, 0, err
 			}
 
 			if err = tx.Commit(); err != nil {
-				return 0, logDMLTxnErr(
+				return 0, 0, logDMLTxnErr(
 					cerror.WrapError(cerror.ErrMySQLTxnError, err),
 					start, s.changefeed, "COMMIT", dmls.rowCount, dmls.startTs)
 			}
-			return dmls.rowCount, nil
+			return dmls.rowCount, dmls.approximateSize, nil
 		})
 		if err != nil {
 			return errors.Trace(err)
 		}
 		log.Debug("Exec Rows succeeded",
-			zap.Int("workerID", s.workerID),
 			zap.String("changefeed", s.changefeed),
+			zap.Int("workerID", s.workerID),
 			zap.Int("numOfRows", dmls.rowCount))
 		return nil
 	}, retry.WithBackoffBaseDelay(pmysql.BackoffBaseDelay.Milliseconds()),
diff --git a/cdc/sink/dmlsink/txn/worker.go b/cdc/sink/dmlsink/txn/worker.go
index f911a2b94..beed5d33b 100644
--- a/cdc/sink/dmlsink/txn/worker.go
+++ b/cdc/sink/dmlsink/txn/worker.go
@@ -29,7 +29,7 @@ import (
 
 type txnWithNotifier struct {
 	*txnEvent
-	wantMore func()
+	postTxnExecuted func()
 }
 
 type worker struct {
@@ -49,9 +49,9 @@ type worker struct {
 	metricTxnWorkerHandledRows   prometheus.Counter
 
 	// Fields only used in the background loop.
-	flushInterval     time.Duration
-	hasPending        bool
-	wantMoreCallbacks []func()
+	flushInterval            time.Duration
+	hasPending               bool
+	postTxnExecutedCallbacks []func()
 }
 
 func newWorker(ctx context.Context, ID int, backend backend, workerCount int) *worker {
@@ -72,17 +72,19 @@ func newWorker(ctx context.Context, ID int, backend backend, workerCount int) *w
 		metricTxnWorkerBusyRatio:     txn.WorkerBusyRatio.WithLabelValues(changefeedID.Namespace, changefeedID.ID),
 		metricTxnWorkerHandledRows:   txn.WorkerHandledRows.WithLabelValues(changefeedID.Namespace, changefeedID.ID, wid),
 
-		flushInterval:     backend.MaxFlushInterval(),
-		hasPending:        false,
-		wantMoreCallbacks: make([]func(), 0, 1024),
+		flushInterval:            backend.MaxFlushInterval(),
+		hasPending:               false,
+		postTxnExecutedCallbacks: make([]func(), 0, 1024),
 	}
 }
 
 // Add adds a txnEvent to the worker.
-// The worker will call unlock() when it's ready to receive more events.
-// In other words, it maybe advances the conflict detector.
-func (w *worker) Add(txn *txnEvent, unlock func()) {
-	w.txnCh.In() <- txnWithNotifier{txn, unlock}
+// The worker will call postTxnExecuted() after the txn executed.
+// The postTxnExecuted will remove the txn related Node in the conflict detector's
+// dependency graph and resolve related dependencies for these transacitons
+// which depend on this executed txn.
+func (w *worker) Add(txn *txnEvent, postTxnExecuted func()) {
+	w.txnCh.In() <- txnWithNotifier{txn, postTxnExecuted}
 }
 
 func (w *worker) close() {
@@ -153,15 +155,15 @@ func (w *worker) onEvent(txn txnWithNotifier) bool {
 		// The table where the event comes from is in stopping, so it's safe
 		// to drop the event directly.
 		txn.txnEvent.Callback()
-		// Still necessary to append the wantMore callback into the pending list.
-		w.wantMoreCallbacks = append(w.wantMoreCallbacks, txn.wantMore)
+		// Still necessary to append the callbacks into the pending list.
+		w.postTxnExecutedCallbacks = append(w.postTxnExecutedCallbacks, txn.postTxnExecuted)
 		return false
 	}
 
 	w.metricConflictDetectDuration.Observe(txn.conflictResolved.Sub(txn.start).Seconds())
 	w.metricQueueDuration.Observe(time.Since(txn.start).Seconds())
 	w.metricTxnWorkerHandledRows.Add(float64(len(txn.Event.Rows)))
-	w.wantMoreCallbacks = append(w.wantMoreCallbacks, txn.wantMore)
+	w.postTxnExecutedCallbacks = append(w.postTxnExecutedCallbacks, txn.postTxnExecuted)
 	return w.backend.OnTxnEvent(txn.txnEvent.TxnCallbackableEvent)
 }
 
@@ -184,13 +186,13 @@ func (w *worker) doFlush(flushTimeSlice *time.Duration) error {
 			return err
 		}
 		// Flush successfully, call callbacks to notify conflict detector.
-		for _, wantMore := range w.wantMoreCallbacks {
-			wantMore()
+		for _, postTxnExecuted := range w.postTxnExecutedCallbacks {
+			postTxnExecuted()
 		}
-		w.wantMoreCallbacks = w.wantMoreCallbacks[:0]
-		if cap(w.wantMoreCallbacks) > 1024 {
+		w.postTxnExecutedCallbacks = w.postTxnExecutedCallbacks[:0]
+		if cap(w.postTxnExecutedCallbacks) > 1024 {
 			// Resize the buffer if it's too big.
-			w.wantMoreCallbacks = make([]func(), 0, 1024)
+			w.postTxnExecutedCallbacks = make([]func(), 0, 1024)
 		}
 	}
 
diff --git a/cdc/sink/metrics/metrics.go b/cdc/sink/metrics/metrics.go
index 0fdf258e8..1f7cfa786 100644
--- a/cdc/sink/metrics/metrics.go
+++ b/cdc/sink/metrics/metrics.go
@@ -36,6 +36,15 @@ var (
 			Buckets:   prometheus.ExponentialBuckets(1, 2, 18),
 		}, []string{"namespace", "changefeed", "type"}) // type is for `sinkType`
 
+	// ExecWriteBytesGauge records the total number of bytes written by sink.
+	TotalWriteBytesCounter = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: "ticdc",
+			Subsystem: "sink",
+			Name:      "write_bytes_total",
+			Help:      "Total number of bytes written by sink",
+		}, []string{"namespace", "changefeed", "type"}) // type is for `sinkType`
+
 	// LargeRowSizeHistogram records size of large rows.
 	LargeRowSizeHistogram = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
@@ -69,6 +78,7 @@ var (
 // InitMetrics registers all metrics in this file.
 func InitMetrics(registry *prometheus.Registry) {
 	registry.MustRegister(ExecBatchHistogram)
+	registry.MustRegister(TotalWriteBytesCounter)
 	registry.MustRegister(ExecDDLHistogram)
 	registry.MustRegister(LargeRowSizeHistogram)
 	registry.MustRegister(ExecutionErrorCounter)
diff --git a/cdc/sink/metrics/statistics.go b/cdc/sink/metrics/statistics.go
index 520b68d39..8f8cb364f 100644
--- a/cdc/sink/metrics/statistics.go
+++ b/cdc/sink/metrics/statistics.go
@@ -36,6 +36,7 @@ func NewStatistics(ctx context.Context, sinkType sink.Type) *Statistics {
 	s := sinkType.String()
 	statistics.metricExecDDLHis = ExecDDLHistogram.WithLabelValues(namespcae, changefeedID, s)
 	statistics.metricExecBatchHis = ExecBatchHistogram.WithLabelValues(namespcae, changefeedID, s)
+	statistics.metricTotalWriteBytesCnt = TotalWriteBytesCounter.WithLabelValues(namespcae, changefeedID, s)
 	statistics.metricRowSizeHis = LargeRowSizeHistogram.WithLabelValues(namespcae, changefeedID, s)
 	statistics.metricExecErrCnt = ExecutionErrorCounter.WithLabelValues(namespcae, changefeedID, s)
 	return statistics
@@ -52,6 +53,8 @@ type Statistics struct {
 	metricExecDDLHis prometheus.Observer
 	// Histogram for DML batch size.
 	metricExecBatchHis prometheus.Observer
+	// Counter for total bytes of DML.
+	metricTotalWriteBytesCnt prometheus.Counter
 	// Histogram for Row size.
 	metricRowSizeHis prometheus.Observer
 	// Counter for sink error.
@@ -70,13 +73,14 @@ func (b *Statistics) ObserveRows(rows ...*model.RowChangedEvent) {
 }
 
 // RecordBatchExecution stats batch executors which return (batchRowCount, error).
-func (b *Statistics) RecordBatchExecution(executor func() (int, error)) error {
-	batchSize, err := executor()
+func (b *Statistics) RecordBatchExecution(executor func() (int, int64, error)) error {
+	batchSize, batchWriteBytes, err := executor()
 	if err != nil {
 		b.metricExecErrCnt.Inc()
 		return err
 	}
 	b.metricExecBatchHis.Observe(float64(batchSize))
+	b.metricTotalWriteBytesCnt.Add(float64(batchWriteBytes))
 	return nil
 }
 
diff --git a/cdc/sink/tablesink/table_sink.go b/cdc/sink/tablesink/table_sink.go
index 69f6f61db..e5e8cf19b 100644
--- a/cdc/sink/tablesink/table_sink.go
+++ b/cdc/sink/tablesink/table_sink.go
@@ -32,6 +32,11 @@ type TableSink interface {
 	// For example, calculating the current progress from the statistics of the table sink.
 	// This is a thread-safe method.
 	GetCheckpointTs() model.ResolvedTs
+	// GetLastSyncedTs returns the last synced ts of table sink.
+	// the last synced ts means the biggest commits of the events
+	// that have been flushed to the downstream.
+	// This is a thread-safe method.
+	GetLastSyncedTs() model.Ts
 	// Close closes the table sink.
 	// After it returns, no more events will be sent out from this capture.
 	Close()
diff --git a/cdc/sink/tablesink/table_sink_impl.go b/cdc/sink/tablesink/table_sink_impl.go
index 5ee1b869d..76b65cc77 100644
--- a/cdc/sink/tablesink/table_sink_impl.go
+++ b/cdc/sink/tablesink/table_sink_impl.go
@@ -15,6 +15,7 @@ package tablesink
 
 import (
 	"sort"
+	"sync"
 
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/cdc/model"
@@ -31,6 +32,21 @@ var (
 	_ TableSink = (*EventTableSink[*model.SingleTableTxn, *dmlsink.TxnEventAppender])(nil)
 )
 
+// LastSyncedTsRecord is used to record the last synced ts of table sink with lock
+// lastSyncedTs means the biggest commits of the events
+// that have been flushed to the downstream.
+type LastSyncedTsRecord struct {
+	sync.Mutex
+	lastSyncedTs model.Ts
+}
+
+// getLastSyncedTs get value from LastSyncedTsRecord
+func (r *LastSyncedTsRecord) getLastSyncedTs() model.Ts {
+	r.Lock()
+	defer r.Unlock()
+	return r.lastSyncedTs
+}
+
 // EventTableSink is a table sink that can write events.
 type EventTableSink[E dmlsink.TableEvent, P dmlsink.Appender[E]] struct {
 	changefeedID model.ChangeFeedID
@@ -46,6 +62,8 @@ type EventTableSink[E dmlsink.TableEvent, P dmlsink.Appender[E]] struct {
 	eventBuffer []E
 	state       state.TableSinkState
 
+	lastSyncedTs LastSyncedTsRecord
+
 	// For dataflow metrics.
 	metricsTableSinkTotalRows prometheus.Counter
 }
@@ -69,6 +87,7 @@ func New[E dmlsink.TableEvent, P dmlsink.Appender[E]](
 		eventAppender:             appender,
 		eventBuffer:               make([]E, 0, 1024),
 		state:                     state.TableSinkSinking,
+		lastSyncedTs:              LastSyncedTsRecord{lastSyncedTs: startTs},
 		metricsTableSinkTotalRows: totalRowsCounter,
 	}
 }
@@ -114,9 +133,24 @@ func (e *EventTableSink[E, P]) UpdateResolvedTs(resolvedTs model.ResolvedTs) err
 			return SinkInternalError{err}
 		}
 		// We have to record the event ID for the callback.
+		postEventFlushFunc := e.progressTracker.addEvent()
+		evCommitTs := ev.GetCommitTs()
 		ce := &dmlsink.CallbackableEvent[E]{
-			Event:     ev,
-			Callback:  e.progressTracker.addEvent(),
+			Event: ev,
+			Callback: func() {
+				// Due to multi workers will call this callback concurrently,
+				// we need to add lock to protect lastSyncedTs
+				// we need make a performance test for it
+				{
+					e.lastSyncedTs.Lock()
+					defer e.lastSyncedTs.Unlock()
+
+					if e.lastSyncedTs.lastSyncedTs < evCommitTs {
+						e.lastSyncedTs.lastSyncedTs = evCommitTs
+					}
+				}
+				postEventFlushFunc()
+			},
 			SinkState: &e.state,
 		}
 		resolvedCallbackableEvents = append(resolvedCallbackableEvents, ce)
@@ -140,6 +174,13 @@ func (e *EventTableSink[E, P]) GetCheckpointTs() model.ResolvedTs {
 	return e.progressTracker.advance()
 }
 
+// GetLastSyncedTs returns the last synced ts of table sink.
+// lastSyncedTs means the biggest commits of all the events
+// that have been flushed to the downstream.
+func (e *EventTableSink[E, P]) GetLastSyncedTs() model.Ts {
+	return e.lastSyncedTs.getLastSyncedTs()
+}
+
 // Close closes the table sink.
 // After it returns, no more events will be sent out from this capture.
 func (e *EventTableSink[E, P]) Close() {
diff --git a/cmd/kafka-consumer/main.go b/cmd/kafka-consumer/main.go
index 29da7cf23..797e4609c 100644
--- a/cmd/kafka-consumer/main.go
+++ b/cmd/kafka-consumer/main.go
@@ -433,7 +433,7 @@ func NewConsumer(ctx context.Context) (*Consumer, error) {
 	// rule, make sure decoded `RowChangedEvent` contains information
 	// identical to the CDC side.
 	if replicaConfig != nil {
-		eventRouter, err := dispatcher.NewEventRouter(replicaConfig, kafkaTopic)
+		eventRouter, err := dispatcher.NewEventRouter(replicaConfig, kafkaTopic, protocol)
 		if err != nil {
 			return nil, errors.Trace(err)
 		}
diff --git a/go.mod b/go.mod
index e6f988a71..603a386c3 100644
--- a/go.mod
+++ b/go.mod
@@ -327,3 +327,9 @@ replace github.com/benbjohnson/clock v1.3.0 => github.com/benbjohnson/clock v1.1
 replace go.opencensus.io => go.opencensus.io v0.23.1-0.20220331163232-052120675fac
 
 replace github.com/go-ldap/ldap/v3 => github.com/YangKeao/ldap/v3 v3.4.5-0.20230421065457-369a3bab1117
+
+// TODO: `sourcegraph.com/sourcegraph/appdash` has been archived, and the original host has been removed.
+// Please remove these dependencies.
+replace sourcegraph.com/sourcegraph/appdash => github.com/sourcegraph/appdash v0.0.0-20190731080439-ebfcffb1b5c0
+
+replace sourcegraph.com/sourcegraph/appdash-data => github.com/sourcegraph/appdash-data v0.0.0-20151005221446-73f23eafcf67
diff --git a/go.sum b/go.sum
index 38c7b314c..33f7ceac0 100644
--- a/go.sum
+++ b/go.sum
@@ -1059,6 +1059,10 @@ github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9
 github.com/soheilhy/cmux v0.1.4/go.mod h1:IM3LyeVVIOuxMH7sFAkER9+bJ4dT7Ms6E4xg4kGIyLM=
 github.com/soheilhy/cmux v0.1.5 h1:jjzc5WVemNEDTLwv9tlmemhC73tI08BNOIGwBOo10Js=
 github.com/soheilhy/cmux v0.1.5/go.mod h1:T7TcVDs9LWfQgPlPsdngu6I6QIoyIFZDDC6sNE1GqG0=
+github.com/sourcegraph/appdash v0.0.0-20190731080439-ebfcffb1b5c0 h1:IJ3DuWHPTJrsqtIqjfdmPTELdTFGefvrOa2eTeRBleQ=
+github.com/sourcegraph/appdash v0.0.0-20190731080439-ebfcffb1b5c0/go.mod h1:V952P4GGl1v/MMynLwxVdWEbSZJx+n0oOO3ljnez+WU=
+github.com/sourcegraph/appdash-data v0.0.0-20151005221446-73f23eafcf67 h1:8ZnTA26bBOoPkAbbitKPgNlpw0Bwt7ZlpYgZWHWJR/w=
+github.com/sourcegraph/appdash-data v0.0.0-20151005221446-73f23eafcf67/go.mod h1:tNZjgbYncKL5HxvDULAr/mWDmFz4B7H8yrXEDlnoIiw=
 github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
 github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ=
 github.com/spf13/afero v1.2.2/go.mod h1:9ZxEEn6pIJ8Rxe320qSDBk6AsU0r9pR7Q4OcevTdifk=
@@ -1971,9 +1975,5 @@ sigs.k8s.io/structured-merge-diff/v4 v4.2.3/go.mod h1:qjx8mGObPmV2aSZepjQjbmb2ih
 sigs.k8s.io/yaml v1.2.0/go.mod h1:yfXDCHCao9+ENCvLSE62v9VSji2MKu5jeNfTrofGhJc=
 sigs.k8s.io/yaml v1.3.0 h1:a2VclLzOGrwOHDiV8EfBGhvjHvP46CtW5j6POvhYGGo=
 sigs.k8s.io/yaml v1.3.0/go.mod h1:GeOyir5tyXNByN85N/dRIT9es5UQNerPYEKK56eTBm8=
-sourcegraph.com/sourcegraph/appdash v0.0.0-20190731080439-ebfcffb1b5c0 h1:ucqkfpjg9WzSUubAO62csmucvxl4/JeW3F4I4909XkM=
-sourcegraph.com/sourcegraph/appdash v0.0.0-20190731080439-ebfcffb1b5c0/go.mod h1:hI742Nqp5OhwiqlzhgfbWU4mW4yO10fP+LoT9WOswdU=
-sourcegraph.com/sourcegraph/appdash-data v0.0.0-20151005221446-73f23eafcf67 h1:e1sMhtVq9AfcEy8AXNb8eSg6gbzfdpYhoNqnPJa+GzI=
-sourcegraph.com/sourcegraph/appdash-data v0.0.0-20151005221446-73f23eafcf67/go.mod h1:L5q+DGLGOQFpo1snNEkLOJT2d1YTW66rWNzatr3He1k=
 upper.io/db.v3 v3.7.1+incompatible h1:GiK/NmDUClH3LrZd54qj5OQsz8brGFv652QXyRXtg2U=
 upper.io/db.v3 v3.7.1+incompatible/go.mod h1:FgTdD24eBjJAbPKsQSiHUNgXjOR4Lub3u1UMHSIh82Y=
diff --git a/metrics/alertmanager/ticdc.rules.yml b/metrics/alertmanager/ticdc.rules.yml
index 3c80ccda1..acf51f932 100644
--- a/metrics/alertmanager/ticdc.rules.yml
+++ b/metrics/alertmanager/ticdc.rules.yml
@@ -51,6 +51,18 @@ groups:
       value: '{{ $value }}'
       summary: cdc owner checkpoint delay more than 10 minutes
 
+  - alert: cdc_resolvedts_high_delay
+    expr: ticdc_owner_resolved_ts_lag > 300
+    for: 1m
+    labels:
+      env: ENV_LABELS_ENV
+      level: critical
+      expr: ticdc_owner_resolved_ts_lag > 300
+    annotations:
+      description: 'cluster: ENV_LABELS_ENV, instance: {{ $labels.instance }}, values: {{ $value }}'
+      value: '{{ $value }}'
+      summary: cdc owner resolved ts delay more than 5 minutes
+
   - alert: ticdc_sink_execution_error
     expr: changes(ticdc_sink_execution_error[1m]) > 0
     for: 1m
diff --git a/pkg/causality/conflict_detector.go b/pkg/causality/conflict_detector.go
index 641dde8b8..f1f7495e7 100644
--- a/pkg/causality/conflict_detector.go
+++ b/pkg/causality/conflict_detector.go
@@ -44,8 +44,8 @@ type ConflictDetector[Worker worker[Txn], Txn txnEvent] struct {
 }
 
 type txnFinishedEvent struct {
-	node         *internal.Node
-	conflictKeys []uint64
+	node           *internal.Node
+	sortedKeysHash []uint64
 }
 
 // NewConflictDetector creates a new ConflictDetector.
@@ -73,20 +73,29 @@ func NewConflictDetector[Worker worker[Txn], Txn txnEvent](
 
 // Add pushes a transaction to the ConflictDetector.
 //
-// NOTE: if multiple threads access this concurrently, Txn.ConflictKeys must be sorted.
+// NOTE: if multiple threads access this concurrently,
+// Txn.GenSortedDedupKeysHash must be sorted by the slot index.
 func (d *ConflictDetector[Worker, Txn]) Add(txn Txn) {
-	conflictKeys := txn.ConflictKeys(d.numSlots)
+	sortedKeysHash := txn.GenSortedDedupKeysHash(d.numSlots)
 	node := internal.NewNode()
 	node.OnResolved = func(workerID int64) {
-		unlock := func() {
+		// This callback is called after the transaction is executed.
+		postTxnExecuted := func() {
+			// After this transaction is executed, we can remove the node from the graph,
+			// and resolve related dependencies for these transacitons which depend on this
+			// executed transaction.
 			node.Remove()
-			d.garbageNodes.In() <- txnFinishedEvent{node, conflictKeys}
+
+			// Send this node to garbageNodes to GC it from the slots if this node is still
+			// occupied related slots.
+			d.garbageNodes.In() <- txnFinishedEvent{node, sortedKeysHash}
 		}
-		d.sendToWorker(txn, unlock, workerID)
+		// Send this txn to related worker as soon as all dependencies are resolved.
+		d.sendToWorker(txn, postTxnExecuted, workerID)
 	}
 	node.RandWorkerID = func() int64 { return d.nextWorkerID.Add(1) % int64(len(d.workers)) }
 	node.OnNotified = func(callback func()) { d.notifiedNodes.In() <- callback }
-	d.slots.Add(node, conflictKeys)
+	d.slots.Add(node, sortedKeysHash)
 }
 
 // Close closes the ConflictDetector.
@@ -110,18 +119,18 @@ func (d *ConflictDetector[Worker, Txn]) runBackgroundTasks() {
 			}
 		case event := <-d.garbageNodes.Out():
 			if event.node != nil {
-				d.slots.Free(event.node, event.conflictKeys)
+				d.slots.Free(event.node, event.sortedKeysHash)
 			}
 		}
 	}
 }
 
 // sendToWorker should not call txn.Callback if it returns an error.
-func (d *ConflictDetector[Worker, Txn]) sendToWorker(txn Txn, unlock func(), workerID int64) {
+func (d *ConflictDetector[Worker, Txn]) sendToWorker(txn Txn, postTxnExecuted func(), workerID int64) {
 	if workerID < 0 {
 		panic("must assign with a valid workerID")
 	}
 	txn.OnConflictResolved()
 	worker := d.workers[workerID]
-	worker.Add(txn, unlock)
+	worker.Add(txn, postTxnExecuted)
 }
diff --git a/pkg/causality/internal/node.go b/pkg/causality/internal/node.go
index c9c7a3a29..cfcd5fc2f 100644
--- a/pkg/causality/internal/node.go
+++ b/pkg/causality/internal/node.go
@@ -48,7 +48,7 @@ type Node struct {
 	// Immutable fields.
 	id int64
 
-	// Set the callback that the node is resolved.
+	// Called when all dependencies are resolved.
 	OnResolved func(id workerID)
 	// Set the id generator to get a random ID.
 	RandWorkerID func() workerID
@@ -56,10 +56,10 @@ type Node struct {
 	OnNotified func(callback func())
 
 	// Following fields are used for notifying a node's dependers lock-free.
-	totalDependees    int32
-	resolvedDependees int32
-	removedDependees  int32
-	resolvedList      []int64
+	totalDependencies    int32
+	resolvedDependencies int32
+	removedDependencies  int32
+	resolvedList         []int64
 
 	// Following fields are protected by `mu`.
 	mu sync.Mutex
@@ -87,9 +87,9 @@ func NewNode() (ret *Node) {
 		ret.id = genNextNodeID()
 		ret.OnResolved = nil
 		ret.RandWorkerID = nil
-		ret.totalDependees = 0
-		ret.resolvedDependees = 0
-		ret.removedDependees = 0
+		ret.totalDependencies = 0
+		ret.resolvedDependencies = 0
+		ret.removedDependencies = 0
 		ret.resolvedList = nil
 		ret.assignedTo = unassigned
 		ret.removed = false
@@ -105,36 +105,43 @@ func (n *Node) NodeID() int64 {
 }
 
 // DependOn implements interface internal.SlotNode.
-func (n *Node) DependOn(unresolvedDeps map[int64]*Node, resolvedDeps int) {
-	resolvedDependees, removedDependees := int32(0), int32(0)
+func (n *Node) DependOn(dependencyNodes map[int64]*Node, noDependencyKeyCnt int) {
+	resolvedDependencies, removedDependencies := int32(0), int32(0)
 
 	depend := func(target *Node) {
 		if target == nil {
 			// For a given Node, every dependency corresponds to a target.
 			// If target is nil it means the dependency doesn't conflict
 			// with any other nodes. However it's still necessary to track
-			// it because Node.tryResolve needs to know it.
-			resolvedDependees = stdatomic.AddInt32(&n.resolvedDependees, 1)
-			stdatomic.StoreInt64(&n.resolvedList[resolvedDependees-1], assignedToAny)
-			removedDependees = stdatomic.AddInt32(&n.removedDependees, 1)
+			// it because Node.tryResolve needs to counting the number of
+			// resolved dependencies.
+			resolvedDependencies = stdatomic.AddInt32(&n.resolvedDependencies, 1)
+			stdatomic.StoreInt64(&n.resolvedList[resolvedDependencies-1], assignedToAny)
+			removedDependencies = stdatomic.AddInt32(&n.removedDependencies, 1)
 			return
 		}
 
 		if target.id == n.id {
 			panic("you cannot depend on yourself")
 		}
-		// Lock target and insert `n` into target.dependers.
+
+		// The target node might be removed or modified in other places, for example
+		// after its corresponding transaction has been executed.
 		target.mu.Lock()
 		defer target.mu.Unlock()
 
 		if target.assignedTo != unassigned {
 			// The target has already been assigned to a worker.
-			resolvedDependees = stdatomic.AddInt32(&n.resolvedDependees, 1)
-			stdatomic.StoreInt64(&n.resolvedList[resolvedDependees-1], target.assignedTo)
+			// In this case, record the worker ID in `resolvedList`, and this node
+			// probably can be sent to the same worker and executed sequentially.
+			resolvedDependencies = stdatomic.AddInt32(&n.resolvedDependencies, 1)
+			stdatomic.StoreInt64(&n.resolvedList[resolvedDependencies-1], target.assignedTo)
 		}
+
+		// Add the node to the target's dependers if the target has not been removed.
 		if target.removed {
 			// The target has already been removed.
-			removedDependees = stdatomic.AddInt32(&n.removedDependees, 1)
+			removedDependencies = stdatomic.AddInt32(&n.removedDependencies, 1)
 		} else if _, exist := target.getOrCreateDependers().ReplaceOrInsert(n); exist {
 			// Should never depend on a target redundantly.
 			panic("should never exist")
@@ -143,23 +150,24 @@ func (n *Node) DependOn(unresolvedDeps map[int64]*Node, resolvedDeps int) {
 
 	// Re-allocate ID in `DependOn` instead of creating the node, because the node can be
 	// pending in slots after it's created.
+	// ?: why gen new ID here?
 	n.id = genNextNodeID()
 
-	// `totalDependees` and `resolvedList` must be initialized before depending on any targets.
-	n.totalDependees = int32(len(unresolvedDeps) + resolvedDeps)
-	n.resolvedList = make([]int64, 0, n.totalDependees)
-	for i := 0; i < int(n.totalDependees); i++ {
+	// `totalDependcies` and `resolvedList` must be initialized before depending on any targets.
+	n.totalDependencies = int32(len(dependencyNodes) + noDependencyKeyCnt)
+	n.resolvedList = make([]int64, 0, n.totalDependencies)
+	for i := 0; i < int(n.totalDependencies); i++ {
 		n.resolvedList = append(n.resolvedList, unassigned)
 	}
 
-	for _, target := range unresolvedDeps {
-		depend(target)
+	for _, node := range dependencyNodes {
+		depend(node)
 	}
-	for i := 0; i < resolvedDeps; i++ {
+	for i := 0; i < noDependencyKeyCnt; i++ {
 		depend(nil)
 	}
 
-	n.maybeResolve(resolvedDependees, removedDependees)
+	n.maybeResolve(resolvedDependencies, removedDependencies)
 }
 
 // Remove implements interface internal.SlotNode.
@@ -171,8 +179,8 @@ func (n *Node) Remove() {
 	if n.dependers != nil {
 		// `mu` must be holded during accessing dependers.
 		n.dependers.Ascend(func(node *Node) bool {
-			removedDependees := stdatomic.AddInt32(&node.removedDependees, 1)
-			node.maybeResolve(0, removedDependees)
+			removedDependencies := stdatomic.AddInt32(&node.removedDependencies, 1)
+			node.maybeResolve(0, removedDependencies)
 			return true
 		})
 		n.dependers.Clear(true)
@@ -217,9 +225,9 @@ func (n *Node) assignTo(workerID int64) bool {
 	if n.dependers != nil {
 		// `mu` must be holded during accessing dependers.
 		n.dependers.Ascend(func(node *Node) bool {
-			resolvedDependees := stdatomic.AddInt32(&node.resolvedDependees, 1)
-			stdatomic.StoreInt64(&node.resolvedList[resolvedDependees-1], n.assignedTo)
-			node.maybeResolve(resolvedDependees, 0)
+			resolvedDependencies := stdatomic.AddInt32(&node.resolvedDependencies, 1)
+			stdatomic.StoreInt64(&node.resolvedList[resolvedDependencies-1], n.assignedTo)
+			node.maybeResolve(resolvedDependencies, 0)
 			return true
 		})
 	}
@@ -227,14 +235,16 @@ func (n *Node) assignTo(workerID int64) bool {
 	return true
 }
 
-func (n *Node) maybeResolve(resolvedDependees, removedDependees int32) {
-	if workerNum, ok := n.tryResolve(resolvedDependees, removedDependees); ok {
+func (n *Node) maybeResolve(resolvedDependencies, removedDependencies int32) {
+	if workerNum, ok := n.tryResolve(resolvedDependencies, removedDependencies); ok {
 		if workerNum < 0 {
 			panic("Node.tryResolve must return a valid worker ID")
 		}
 		if n.OnNotified != nil {
+			// Notify the conflict detector background worker to assign the node to the worker asynchronously.
 			n.OnNotified(func() { n.assignTo(workerNum) })
 		} else {
+			// Assign the node to the worker directly.
 			n.assignTo(workerNum)
 		}
 	}
@@ -244,39 +254,50 @@ func (n *Node) maybeResolve(resolvedDependees, removedDependees int32) {
 // Returns (_, false) if there is a conflict,
 // returns (rand, true) if there is no conflict,
 // returns (N, true) if only worker N can be used.
-func (n *Node) tryResolve(resolvedDependees, removedDependees int32) (int64, bool) {
-	assignedTo, resolved := n.doResolve(resolvedDependees, removedDependees)
+func (n *Node) tryResolve(resolvedDependencies, removedDependencies int32) (int64, bool) {
+	assignedTo, resolved := n.doResolve(resolvedDependencies, removedDependencies)
 	if resolved && assignedTo == assignedToAny {
 		assignedTo = n.RandWorkerID()
 	}
 	return assignedTo, resolved
 }
 
-func (n *Node) doResolve(resolvedDependees, removedDependees int32) (int64, bool) {
-	if n.totalDependees == 0 {
+func (n *Node) doResolve(resolvedDependencies, removedDependencies int32) (int64, bool) {
+	if n.totalDependencies == 0 {
 		// No conflicts, can select any workers.
 		return assignedToAny, true
 	}
 
-	if resolvedDependees == n.totalDependees {
+	if resolvedDependencies == n.totalDependencies {
 		firstDep := stdatomic.LoadInt64(&n.resolvedList[0])
 		hasDiffDep := false
-		for i := 1; i < int(n.totalDependees); i++ {
+		for i := 1; i < int(n.totalDependencies); i++ {
 			curr := stdatomic.LoadInt64(&n.resolvedList[i])
+			// // Todo: simplify assign to logic, only resolve dependencies nodes after
+			// // corresponding transactions are executed.
+			// //
+			// // In DependOn, depend(nil) set resolvedList[i] to assignedToAny
+			// // for these no dependecy keys.
+			// if curr == assignedToAny {
+			// 	continue
+			// }
 			if firstDep != curr {
 				hasDiffDep = true
 				break
 			}
 		}
 		if !hasDiffDep {
-			// If all dependees are assigned to one same worker, we can assign
-			// this node to the same worker directly.
+			// If all dependency nodes are assigned to the same worker, we can assign
+			// this node to the same worker directly, and they will execute sequentially.
+			// On the other hand, if dependency nodes are assigned to different workers,
+			// This node has to wait all dependency txn executed and all depencecy nodes
+			// are removed.
 			return firstDep, true
 		}
 	}
 
-	// All dependees are removed, so assign the node to any worker is fine.
-	if removedDependees == n.totalDependees {
+	// All dependcies are removed, so assign the node to any worker is fine.
+	if removedDependencies == n.totalDependencies {
 		return assignedToAny, true
 	}
 
diff --git a/pkg/causality/internal/slots.go b/pkg/causality/internal/slots.go
index cc70a0c41..d6d131e02 100644
--- a/pkg/causality/internal/slots.go
+++ b/pkg/causality/internal/slots.go
@@ -28,7 +28,7 @@ type SlotNode[T any] interface {
 	// NodeID tells the node's ID.
 	NodeID() int64
 	// Construct a dependency on `others`.
-	DependOn(unresolvedDeps map[int64]T, resolvedDeps int)
+	DependOn(dependencyNodes map[int64]T, noDependencyKeyCnt int)
 	// Remove the node itself and notify all dependers.
 	Remove()
 	// Free the node itself and remove it from the graph.
@@ -56,32 +56,43 @@ func NewSlots[E SlotNode[E]](numSlots uint64) *Slots[E] {
 }
 
 // Add adds an elem to the slots and calls DependOn for elem.
-func (s *Slots[E]) Add(elem E, keys []uint64) {
-	unresolvedDeps := make(map[int64]E, len(keys))
-	resolvedDeps := 0
+func (s *Slots[E]) Add(elem E, hashes []uint64) {
+	dependencyNodes := make(map[int64]E, len(hashes))
+	noDependecyCnt := 0
 
 	var lastSlot uint64 = math.MaxUint64
-	for _, key := range keys {
-		slotIdx := getSlot(key, s.numSlots)
+	for _, hash := range hashes {
+		// lock the slot that the node belongs to.
+		slotIdx := getSlot(hash, s.numSlots)
 		if lastSlot != slotIdx {
 			s.slots[slotIdx].mu.Lock()
 			lastSlot = slotIdx
 		}
-		if tail, ok := s.slots[slotIdx].nodes[key]; ok {
-			prevID := tail.NodeID()
-			unresolvedDeps[prevID] = tail
+
+		// If there is a node occpuied the same hash slot, we may have conflict with it.
+		// Add the conflict node to the dependencyNodes.
+		if prevNode, ok := s.slots[slotIdx].nodes[hash]; ok {
+			prevID := prevNode.NodeID()
+			// If there are multiple hashes conflicts with the same node, we only need to
+			// depend on the node once.
+			dependencyNodes[prevID] = prevNode
 		} else {
-			resolvedDeps += 1
+			noDependecyCnt += 1
 		}
-		s.slots[slotIdx].nodes[key] = elem
+		// Add this node to the slot, make sure new coming nodes with the same hash should
+		// depend on this node.
+		s.slots[slotIdx].nodes[hash] = elem
 	}
-	elem.DependOn(unresolvedDeps, resolvedDeps)
+
+	// Construct the dependency graph based on collected `dependencyNodes` and with corresponding
+	// slots locked.
+	elem.DependOn(dependencyNodes, noDependecyCnt)
 
 	// Lock those slots one by one and then unlock them one by one, so that
 	// we can avoid 2 transactions get executed interleaved.
 	lastSlot = math.MaxUint64
-	for _, key := range keys {
-		slotIdx := getSlot(key, s.numSlots)
+	for _, hash := range hashes {
+		slotIdx := getSlot(hash, s.numSlots)
 		if lastSlot != slotIdx {
 			s.slots[slotIdx].mu.Unlock()
 			lastSlot = slotIdx
@@ -90,18 +101,21 @@ func (s *Slots[E]) Add(elem E, keys []uint64) {
 }
 
 // Free removes an element from the Slots.
-func (s *Slots[E]) Free(elem E, keys []uint64) {
-	for _, key := range keys {
-		slotIdx := getSlot(key, s.numSlots)
+func (s *Slots[E]) Free(elem E, hashes []uint64) {
+	for _, hash := range hashes {
+		slotIdx := getSlot(hash, s.numSlots)
 		s.slots[slotIdx].mu.Lock()
-		if tail, ok := s.slots[slotIdx].nodes[key]; ok && tail.NodeID() == elem.NodeID() {
-			delete(s.slots[slotIdx].nodes, key)
+		// Remove the node from the slot.
+		// If the node is not in the slot, it means the node has been replaced by new node with the same hash,
+		// in this case we don't need to remove it from the slot.
+		if tail, ok := s.slots[slotIdx].nodes[hash]; ok && tail.NodeID() == elem.NodeID() {
+			delete(s.slots[slotIdx].nodes, hash)
 		}
 		s.slots[slotIdx].mu.Unlock()
 	}
 	elem.Free()
 }
 
-func getSlot(key, numSlots uint64) uint64 {
-	return key % numSlots
+func getSlot(hash, numSlots uint64) uint64 {
+	return hash % numSlots
 }
diff --git a/pkg/causality/worker.go b/pkg/causality/worker.go
index fb2234b26..738ba8891 100644
--- a/pkg/causality/worker.go
+++ b/pkg/causality/worker.go
@@ -13,19 +13,15 @@
 
 package causality
 
-type (
-	conflictKey = uint64
-)
-
 type txnEvent interface {
 	// OnConflictResolved is called when the event leaves ConflictDetector.
 	OnConflictResolved()
 
-	// Keys are in range [0, math.MaxUint64) and must be deduped.
+	// Hashes are in range [0, math.MaxUint64) and must be deduped.
 	//
 	// NOTE: if the conflict detector is accessed by multiple threads concurrently,
-	// ConflictKeys must also be sorted based on `key % numSlots`.
-	ConflictKeys(numSlots uint64) []conflictKey
+	// GenSortedDedupKeysHash must also be sorted based on `key % numSlots`.
+	GenSortedDedupKeysHash(numSlots uint64) []uint64
 }
 
 type worker[Txn txnEvent] interface {
diff --git a/pkg/config/replica_config.go b/pkg/config/replica_config.go
index 8be46c978..d354d4498 100644
--- a/pkg/config/replica_config.go
+++ b/pkg/config/replica_config.go
@@ -101,6 +101,7 @@ var defaultReplicaConfig = &ReplicaConfig{
 	},
 	ChangefeedErrorStuckDuration: util.AddressOf(time.Minute * 30),
 	SQLMode:                      defaultSQLMode,
+	SyncedStatus:                 &SyncedStatusConfig{SyncedCheckInterval: 5 * 60, CheckpointInterval: 15},
 }
 
 // GetDefaultReplicaConfig returns the default replica config.
@@ -145,6 +146,7 @@ type replicaConfig struct {
 	Integrity                    *integrity.Config          `toml:"integrity" json:"integrity"`
 	ChangefeedErrorStuckDuration *time.Duration             `toml:"changefeed-error-stuck-duration" json:"changefeed-error-stuck-duration,omitempty"`
 	SQLMode                      string                     `toml:"sql-mode" json:"sql-mode"`
+	SyncedStatus                 *SyncedStatusConfig        `toml:"synced-status" json:"synced-status,omitempty"`
 }
 
 // Marshal returns the json marshal format of a ReplicationConfig
diff --git a/pkg/config/synced_status_config.go b/pkg/config/synced_status_config.go
new file mode 100644
index 000000000..144851f83
--- /dev/null
+++ b/pkg/config/synced_status_config.go
@@ -0,0 +1,23 @@
+// Copyright 2020 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package config
+
+// SyncedStatusConfig represents synced check interval config for a changefeed
+type SyncedStatusConfig struct {
+	// The minimum interval between the latest synced ts and now required to reach synced state
+	SyncedCheckInterval int64 `toml:"synced-check-interval" json:"synced-check-interval"`
+	// The maximum interval between latest checkpoint ts and now or
+	// between latest sink's checkpoint ts and puller's checkpoint ts required to reach synced state
+	CheckpointInterval int64 `toml:"checkpoint-interval" json:"checkpoint-interval"`
+}
diff --git a/pkg/filter/filter.go b/pkg/filter/filter.go
index 269d704bb..61aedc508 100644
--- a/pkg/filter/filter.go
+++ b/pkg/filter/filter.go
@@ -62,12 +62,16 @@ var allowDDLList = []timodel.ActionType{
 // Filter are safe for concurrent use.
 // TODO: find a better way to abstract this interface.
 type Filter interface {
-	// ShouldIgnoreDMLEvent returns true and nil if the DML event should be ignored.
+	// ShouldIgnoreDMLEvent returns true if the DML event should be ignored.
 	ShouldIgnoreDMLEvent(dml *model.RowChangedEvent, rawRow model.RowChangedDatums, tableInfo *model.TableInfo) (bool, error)
+	// ShouldIgnoreDDLEvent returns true if the DDL event should be ignored.
+	// If a ddl is ignored, it will be applied to cdc's schema storage,
+	// but will not be sent to downstream.
+	ShouldIgnoreDDLEvent(ddl *model.DDLEvent) (bool, error)
 	// ShouldDiscardDDL returns true if this DDL should be discarded.
 	// If a ddl is discarded, it will neither be applied to cdc's schema storage
 	// nor sent to downstream.
-	ShouldDiscardDDL(startTs uint64, ddlType timodel.ActionType, schema, table, query string) (bool, error)
+	ShouldDiscardDDL(ddlType timodel.ActionType, schema, table string) bool
 	// ShouldIgnoreTable returns true if the table should be ignored.
 	ShouldIgnoreTable(schema, table string) bool
 	// ShouldIgnoreSchema returns true if the schema should be ignored.
@@ -144,31 +148,41 @@ func (f *filter) ShouldIgnoreDMLEvent(
 	return f.dmlExprFilter.shouldSkipDML(dml, rawRow, ti)
 }
 
-// ShouldDiscardDDL returns true if this DDL should be discarded.
-// If a ddl is discarded, it will not be applied to cdc's schema storage
-// and sent to downstream.
-func (f *filter) ShouldDiscardDDL(startTs uint64, ddlType timodel.ActionType, schema, table, query string) (discard bool, err error) {
-	discard = !isAllowedDDL(ddlType)
-	if discard {
-		return
-	}
-
-	discard = f.shouldIgnoreStartTs(startTs)
-	if discard {
-		return
+// ShouldDiscardDDL checks if a DDL should be discarded by conditions below:
+// 0. By allow list.
+// 1. By schema name.
+// 2. By table name.
+func (f *filter) ShouldDiscardDDL(ddlType timodel.ActionType, schema, table string) bool {
+	if !isAllowedDDL(ddlType) {
+		return true
 	}
 
 	if IsSchemaDDL(ddlType) {
-		discard = !f.tableFilter.MatchSchema(schema)
-	} else {
-		discard = f.ShouldIgnoreTable(schema, table)
+		return f.ShouldIgnoreSchema(schema)
 	}
+	return f.ShouldIgnoreTable(schema, table)
+}
 
-	if discard {
-		return
+// ShouldIgnoreDDLEvent checks if a DDL event should be ignore by conditions below:
+// 0. By startTs.
+// 1. By ddl type.
+// 2. By ddl query.
+//
+// If a ddl is ignored, it will be applied to cdc's schema storage,
+// but will not be sent to downstream.
+// Note that a ignored ddl is different from a discarded ddl. For example, suppose
+// we have a changefeed-test with the following config:
+//   - table filter: rules = ['test.*']
+//   - event-filters: matcher = ["test.worker"] ignore-event = ["create table"]
+//
+// Then, for the following DDLs:
+//  1. `CREATE TABLE test.worker` will be ignored, but the table will be replicated by changefeed-test.
+//  2. `CREATE TABLE other.worker` will be discarded, and the table will not be replicated by changefeed-test.
+func (f *filter) ShouldIgnoreDDLEvent(ddl *model.DDLEvent) (bool, error) {
+	if f.shouldIgnoreStartTs(ddl.StartTs) {
+		return true, nil
 	}
-
-	return f.sqlEventFilter.shouldSkipDDL(ddlType, schema, table, query)
+	return f.sqlEventFilter.shouldSkipDDL(ddl)
 }
 
 // ShouldIgnoreTable returns true if the specified table should be ignored by this changefeed.
diff --git a/pkg/filter/sql_event_filter.go b/pkg/filter/sql_event_filter.go
index bfc4c17d8..6e00bd35b 100644
--- a/pkg/filter/sql_event_filter.go
+++ b/pkg/filter/sql_event_filter.go
@@ -21,7 +21,6 @@ import (
 	"github.com/pingcap/log"
 	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb/parser"
-	timodel "github.com/pingcap/tidb/parser/model"
 	"github.com/pingcap/tidb/parser/mysql"
 	tfilter "github.com/pingcap/tidb/util/table-filter"
 	"github.com/pingcap/tiflow/cdc/model"
@@ -156,14 +155,14 @@ func (f *sqlEventFilter) getRules(schema, table string) []*sqlEventRule {
 }
 
 // skipDDLEvent skips ddl event by its type and query.
-func (f *sqlEventFilter) shouldSkipDDL(
-	ddlType timodel.ActionType, schema, table, query string,
-) (bool, error) {
+func (f *sqlEventFilter) shouldSkipDDL(ddl *model.DDLEvent) (bool, error) {
+	schema := ddl.TableInfo.TableName.Schema
+	table := ddl.TableInfo.TableName.Table
 	log.Info("sql event filter handle ddl event",
-		zap.Any("ddlType", ddlType), zap.String("schema", schema),
-		zap.String("table", table), zap.String("query", query))
+		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, query, ddlType)
+	evenType, err := ddlToEventType(f.ddlParser, ddl.Query, ddl.Type)
 	f.pLock.Unlock()
 
 	if err != nil {
@@ -171,8 +170,8 @@ func (f *sqlEventFilter) shouldSkipDDL(
 	}
 	if evenType == bf.NullEvent {
 		log.Warn("sql event filter unsupported ddl type, do nothing",
-			zap.String("type", ddlType.String()),
-			zap.String("query", query))
+			zap.String("type", ddl.Type.String()),
+			zap.String("query", ddl.Query))
 		return false, nil
 	}
 
@@ -181,7 +180,7 @@ func (f *sqlEventFilter) shouldSkipDDL(
 		action, err := rule.bf.Filter(
 			binlogFilterSchemaPlaceholder,
 			binlogFilterTablePlaceholder,
-			evenType, query)
+			evenType, ddl.Query)
 		if err != nil {
 			return false, errors.Trace(err)
 		}


Test patch:
diff --git a/cdc/api/v1/api_test.go b/cdc/api/v1/api_test.go
index fd56c7f7a..e2a51d1ef 100644
--- a/cdc/api/v1/api_test.go
+++ b/cdc/api/v1/api_test.go
@@ -99,6 +99,13 @@ func (p *mockStatusProvider) GetCaptures(ctx context.Context) ([]*model.CaptureI
 	return args.Get(0).([]*model.CaptureInfo), args.Error(1)
 }
 
+func (p *mockStatusProvider) GetChangeFeedSyncedStatus(ctx context.Context,
+	changefeedID model.ChangeFeedID,
+) (*model.ChangeFeedSyncedStatusForAPI, error) {
+	args := p.Called(ctx)
+	return args.Get(0).(*model.ChangeFeedSyncedStatusForAPI), args.Error(1)
+}
+
 func (p *mockStatusProvider) IsHealthy(ctx context.Context) (bool, error) {
 	args := p.Called(ctx)
 	return args.Get(0).(bool), args.Error(1)
diff --git a/cdc/api/v2/api_test.go b/cdc/api/v2/api_test.go
index 29f572851..de6f274ab 100644
--- a/cdc/api/v2/api_test.go
+++ b/cdc/api/v2/api_test.go
@@ -48,8 +48,8 @@ func (m *mockPDClient) UpdateServiceGCSafePoint(ctx context.Context,
 }
 
 // GetTS of mockPDClient returns a mock tso
-func (m *mockPDClient) GetTS(ctx context.Context) (int64, int64, error) {
-	return m.logicTime, m.timestamp, nil
+func (c *mockPDClient) GetTS(ctx context.Context) (int64, int64, error) {
+	return c.timestamp, c.logicTime, nil
 }
 
 // GetClusterID of mockPDClient returns a mock ClusterID
@@ -62,13 +62,14 @@ func (c *mockPDClient) Close() {}
 
 type mockStatusProvider struct {
 	owner.StatusProvider
-	changefeedStatus   *model.ChangeFeedStatusForAPI
-	changefeedInfo     *model.ChangeFeedInfo
-	processors         []*model.ProcInfoSnap
-	taskStatus         map[model.CaptureID]*model.TaskStatus
-	changefeedInfos    map[model.ChangeFeedID]*model.ChangeFeedInfo
-	changefeedStatuses map[model.ChangeFeedID]*model.ChangeFeedStatusForAPI
-	err                error
+	changefeedStatus       *model.ChangeFeedStatusForAPI
+	changefeedInfo         *model.ChangeFeedInfo
+	processors             []*model.ProcInfoSnap
+	taskStatus             map[model.CaptureID]*model.TaskStatus
+	changefeedInfos        map[model.ChangeFeedID]*model.ChangeFeedInfo
+	changefeedStatuses     map[model.ChangeFeedID]*model.ChangeFeedStatusForAPI
+	changeFeedSyncedStatus *model.ChangeFeedSyncedStatusForAPI
+	err                    error
 }
 
 // GetChangeFeedStatus returns a changefeeds' runtime status.
@@ -119,3 +120,11 @@ func (m *mockStatusProvider) GetAllChangeFeedStatuses(_ context.Context) (
 ) {
 	return m.changefeedStatuses, m.err
 }
+
+// GetChangeFeedSyncedStatus returns a mock changefeed status.
+func (m *mockStatusProvider) GetChangeFeedSyncedStatus(_ context.Context, changefeedID model.ChangeFeedID) (
+	*model.ChangeFeedSyncedStatusForAPI,
+	error,
+) {
+	return m.changeFeedSyncedStatus, m.err
+}
diff --git a/cdc/api/v2/changefeed_test.go b/cdc/api/v2/changefeed_test.go
index 5b7dcfb03..2b75af1e0 100644
--- a/cdc/api/v2/changefeed_test.go
+++ b/cdc/api/v2/changefeed_test.go
@@ -699,7 +699,9 @@ func TestResumeChangefeed(t *testing.T) {
 	pdClient := &mockPDClient{}
 	etcdClient := mock_etcd.NewMockCDCEtcdClient(gomock.NewController(t))
 	mockUpManager := upstream.NewManager4Test(pdClient)
-	statusProvider := &mockStatusProvider{}
+	statusProvider := &mockStatusProvider{
+		changefeedStatus: &model.ChangeFeedStatusForAPI{},
+	}
 
 	etcdClient.EXPECT().
 		GetEnsureGCServiceID(gomock.Any()).
@@ -951,6 +953,213 @@ func TestPauseChangefeed(t *testing.T) {
 	require.Equal(t, "{}", w.Body.String())
 }
 
+func TestChangefeedSynced(t *testing.T) {
+	syncedInfo := testCase{url: "/api/v2/changefeeds/%s/synced?namespace=abc", method: "GET"}
+	helpers := NewMockAPIV2Helpers(gomock.NewController(t))
+	cp := mock_capture.NewMockCapture(gomock.NewController(t))
+	owner := mock_owner.NewMockOwner(gomock.NewController(t))
+	apiV2 := NewOpenAPIV2ForTest(cp, helpers)
+	router := newRouter(apiV2)
+
+	statusProvider := &mockStatusProvider{}
+
+	cp.EXPECT().StatusProvider().Return(statusProvider).AnyTimes()
+	cp.EXPECT().IsReady().Return(true).AnyTimes()
+	cp.EXPECT().GetOwner().Return(owner, nil).AnyTimes()
+	cp.EXPECT().IsOwner().Return(true).AnyTimes()
+
+	pdClient := &mockPDClient{}
+	mockUpManager := upstream.NewManager4Test(pdClient)
+	cp.EXPECT().GetUpstreamManager().Return(mockUpManager, nil).AnyTimes()
+
+	{
+		// case 1: invalid changefeed id
+		w := httptest.NewRecorder()
+		invalidID := "@^Invalid"
+		req, _ := http.NewRequestWithContext(context.Background(),
+			syncedInfo.method, fmt.Sprintf(syncedInfo.url, invalidID), nil)
+		router.ServeHTTP(w, req)
+		respErr := model.HTTPError{}
+		err := json.NewDecoder(w.Body).Decode(&respErr)
+		require.Nil(t, err)
+		require.Contains(t, respErr.Code, "ErrAPIInvalidParam")
+	}
+
+	{
+		// case 2: not existed changefeed id
+		validID := changeFeedID.ID
+		statusProvider.err = cerrors.ErrChangeFeedNotExists.GenWithStackByArgs(validID)
+		w := httptest.NewRecorder()
+		req, _ := http.NewRequestWithContext(context.Background(), syncedInfo.method,
+			fmt.Sprintf(syncedInfo.url, validID), nil)
+		router.ServeHTTP(w, req)
+		respErr := model.HTTPError{}
+		err := json.NewDecoder(w.Body).Decode(&respErr)
+		require.Nil(t, err)
+		require.Contains(t, respErr.Code, "ErrChangeFeedNotExists")
+		require.Equal(t, http.StatusBadRequest, w.Code)
+	}
+
+	validID := changeFeedID.ID
+	cfInfo := &model.ChangeFeedInfo{
+		ID: validID,
+	}
+	statusProvider.err = nil
+	statusProvider.changefeedInfo = cfInfo
+	{
+		helpers.EXPECT().getPDClient(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, cerrors.ErrAPIGetPDClientFailed).Times(1)
+		// case3: pd is offline，resolvedTs - checkpointTs > 15s
+		statusProvider.changeFeedSyncedStatus = &model.ChangeFeedSyncedStatusForAPI{
+			CheckpointTs:     1701153217279 << 18,
+			LastSyncedTs:     1701153217279 << 18,
+			PullerResolvedTs: 1701153247279 << 18,
+		}
+		w := httptest.NewRecorder()
+		req, _ := http.NewRequestWithContext(
+			context.Background(),
+			syncedInfo.method,
+			fmt.Sprintf(syncedInfo.url, validID),
+			nil,
+		)
+		router.ServeHTTP(w, req)
+		require.Equal(t, http.StatusOK, w.Code)
+		resp := SyncedStatus{}
+		err := json.NewDecoder(w.Body).Decode(&resp)
+		require.Nil(t, err)
+		require.Equal(t, false, resp.Synced)
+		require.Equal(t, "[CDC:ErrAPIGetPDClientFailed]failed to get PDClient to connect PD, "+
+			"please recheck. Besides the data is not finish syncing", resp.Info)
+	}
+
+	{
+		helpers.EXPECT().getPDClient(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, cerrors.ErrAPIGetPDClientFailed).Times(1)
+		// case4: pd is offline，resolvedTs - checkpointTs < 15s
+		statusProvider.changeFeedSyncedStatus = &model.ChangeFeedSyncedStatusForAPI{
+			CheckpointTs:     1701153217279 << 18,
+			LastSyncedTs:     1701153217279 << 18,
+			PullerResolvedTs: 1701153217479 << 18,
+		}
+		w := httptest.NewRecorder()
+		req, _ := http.NewRequestWithContext(
+			context.Background(),
+			syncedInfo.method,
+			fmt.Sprintf(syncedInfo.url, validID),
+			nil,
+		)
+		router.ServeHTTP(w, req)
+		require.Equal(t, http.StatusOK, w.Code)
+		resp := SyncedStatus{}
+		err := json.NewDecoder(w.Body).Decode(&resp)
+		require.Nil(t, err)
+		require.Equal(t, false, resp.Synced)
+		require.Equal(t, "[CDC:ErrAPIGetPDClientFailed]failed to get PDClient to connect PD, please recheck. "+
+			"You should check the pd status first. If pd status is normal, means we don't finish sync data. "+
+			"If pd is offline, please check whether we satisfy the condition that "+
+			"the time difference from lastSyncedTs to the current time from the time zone of pd is greater than 300 secs. "+
+			"If it's satisfied, means the data syncing is totally finished", resp.Info)
+	}
+
+	helpers.EXPECT().getPDClient(gomock.Any(), gomock.Any(), gomock.Any()).Return(pdClient, nil).AnyTimes()
+	pdClient.logicTime = 1000
+	pdClient.timestamp = 1701153217279
+	{
+		// case5: pdTs - lastSyncedTs > 5min, pdTs - checkpointTs < 15s
+		statusProvider.changeFeedSyncedStatus = &model.ChangeFeedSyncedStatusForAPI{
+			CheckpointTs:     1701153217209 << 18,
+			LastSyncedTs:     1701152217279 << 18,
+			PullerResolvedTs: 1701153217229 << 18,
+		}
+		w := httptest.NewRecorder()
+		req, _ := http.NewRequestWithContext(
+			context.Background(),
+			syncedInfo.method,
+			fmt.Sprintf(syncedInfo.url, validID),
+			nil,
+		)
+		router.ServeHTTP(w, req)
+		require.Equal(t, http.StatusOK, w.Code)
+		resp := SyncedStatus{}
+		err := json.NewDecoder(w.Body).Decode(&resp)
+		require.Nil(t, err)
+		require.Equal(t, true, resp.Synced)
+		require.Equal(t, "Data syncing is finished", resp.Info)
+	}
+
+	{
+		// case6: pdTs - lastSyncedTs > 5min, pdTs - checkpointTs > 15s, resolvedTs - checkpointTs < 15s
+		statusProvider.changeFeedSyncedStatus = &model.ChangeFeedSyncedStatusForAPI{
+			CheckpointTs:     1701153201279 << 18,
+			LastSyncedTs:     1701152217279 << 18,
+			PullerResolvedTs: 1701153201379 << 18,
+		}
+		w := httptest.NewRecorder()
+		req, _ := http.NewRequestWithContext(
+			context.Background(),
+			syncedInfo.method,
+			fmt.Sprintf(syncedInfo.url, validID),
+			nil,
+		)
+		router.ServeHTTP(w, req)
+		require.Equal(t, http.StatusOK, w.Code)
+		resp := SyncedStatus{}
+		err := json.NewDecoder(w.Body).Decode(&resp)
+		require.Nil(t, err)
+		require.Equal(t, false, resp.Synced)
+		require.Equal(t, "Please check whether PD is online and TiKV Regions are all available. "+
+			"If PD is offline or some TiKV regions are not available, it means that the data syncing process is complete. "+
+			"To check whether TiKV regions are all available, "+
+			"you can view 'TiKV-Details' > 'Resolved-Ts' > 'Max Leader Resolved TS gap' on Grafana. "+
+			"If the gap is large, such as a few minutes, it means that some regions in TiKV are unavailable. "+
+			"Otherwise, if the gap is small and PD is online, it means the data syncing is incomplete, so please wait", resp.Info)
+	}
+
+	{
+		// case7: pdTs - lastSyncedTs > 5min, pdTs - checkpointTs > 15s, resolvedTs - checkpointTs > 15s
+		statusProvider.changeFeedSyncedStatus = &model.ChangeFeedSyncedStatusForAPI{
+			CheckpointTs:     1701153201279 << 18,
+			LastSyncedTs:     1701152207279 << 18,
+			PullerResolvedTs: 1701153218279 << 18,
+		}
+		w := httptest.NewRecorder()
+		req, _ := http.NewRequestWithContext(
+			context.Background(),
+			syncedInfo.method,
+			fmt.Sprintf(syncedInfo.url, validID),
+			nil,
+		)
+		router.ServeHTTP(w, req)
+		require.Equal(t, http.StatusOK, w.Code)
+		resp := SyncedStatus{}
+		err := json.NewDecoder(w.Body).Decode(&resp)
+		require.Nil(t, err)
+		require.Equal(t, false, resp.Synced)
+		require.Equal(t, "The data syncing is not finished, please wait", resp.Info)
+	}
+
+	{
+		// case8: pdTs - lastSyncedTs < 5min
+		statusProvider.changeFeedSyncedStatus = &model.ChangeFeedSyncedStatusForAPI{
+			CheckpointTs:     1701153217279 << 18,
+			LastSyncedTs:     1701153213279 << 18,
+			PullerResolvedTs: 1701153217279 << 18,
+		}
+		w := httptest.NewRecorder()
+		req, _ := http.NewRequestWithContext(
+			context.Background(),
+			syncedInfo.method,
+			fmt.Sprintf(syncedInfo.url, validID),
+			nil,
+		)
+		router.ServeHTTP(w, req)
+		require.Equal(t, http.StatusOK, w.Code)
+		resp := SyncedStatus{}
+		err := json.NewDecoder(w.Body).Decode(&resp)
+		require.Nil(t, err)
+		require.Equal(t, false, resp.Synced)
+		require.Equal(t, "The data syncing is not finished, please wait", resp.Info)
+	}
+}
+
 func TestHasRunningImport(t *testing.T) {
 	integration.BeforeTestExternal(t)
 	testEtcdCluster := integration.NewClusterV3(
diff --git a/cdc/api/v2/model_test.go b/cdc/api/v2/model_test.go
index ed08876ab..b6558f24d 100644
--- a/cdc/api/v2/model_test.go
+++ b/cdc/api/v2/model_test.go
@@ -85,7 +85,8 @@ var defaultAPIConfig = &ReplicaConfig{
 	},
 	ChangefeedErrorStuckDuration: &JSONDuration{*config.
 		GetDefaultReplicaConfig().ChangefeedErrorStuckDuration},
-	SQLMode: config.GetDefaultReplicaConfig().SQLMode,
+	SQLMode:      config.GetDefaultReplicaConfig().SQLMode,
+	SyncedStatus: (*SyncedStatusConfig)(config.GetDefaultReplicaConfig().SyncedStatus),
 }
 
 func TestDefaultReplicaConfig(t *testing.T) {
diff --git a/cdc/entry/schema/snapshot_test.go b/cdc/entry/schema/snapshot_test.go
index d8248784f..68106599e 100644
--- a/cdc/entry/schema/snapshot_test.go
+++ b/cdc/entry/schema/snapshot_test.go
@@ -219,19 +219,19 @@ func TestUpdatePartition(t *testing.T) {
 	oldTb = newTbInfo(1, "DB_1", 11)
 	oldTb.Partition = nil
 	require.Nil(t, snap.inner.createTable(oldTb, 110))
-	require.Error(t, snap.inner.updatePartition(newTbInfo(1, "DB_1", 11), 120))
+	require.Error(t, snap.inner.updatePartition(newTbInfo(1, "DB_1", 11), false, 120))
 
 	// updatePartition fails if the new table is not partitioned.
 	require.Nil(t, snap.inner.dropTable(11, 130))
 	require.Nil(t, snap.inner.createTable(newTbInfo(1, "DB_1", 11), 140))
 	newTb = newTbInfo(1, "DB_1", 11)
 	newTb.Partition = nil
-	require.Error(t, snap.inner.updatePartition(newTb, 150))
+	require.Error(t, snap.inner.updatePartition(newTb, false, 150))
 	snap1 = snap.Copy()
 
 	newTb = newTbInfo(1, "DB_1", 11)
 	newTb.Partition.Definitions[0] = timodel.PartitionDefinition{ID: 11 + 65536*2}
-	require.Nil(t, snap.inner.updatePartition(newTb, 160))
+	require.Nil(t, snap.inner.updatePartition(newTb, false, 160))
 	snap2 = snap.Copy()
 
 	info, _ = snap1.PhysicalTableByID(11)
@@ -253,6 +253,31 @@ func TestUpdatePartition(t *testing.T) {
 	require.True(t, snap2.IsIneligibleTableID(11+65536*2))
 }
 
+func TestTruncateTablePartition(t *testing.T) {
+	var oldTb, newTb *model.TableInfo
+
+	snap := NewEmptySnapshot(false)
+	require.Nil(t, snap.inner.createSchema(newDBInfo(1), 100))
+
+	// updatePartition fails if the old table is not partitioned.
+	oldTb = newTbInfo(1, "DB_1", 11)
+	oldTb.Partition = nil
+	require.Nil(t, snap.inner.createTable(oldTb, 110))
+	require.Error(t, snap.inner.updatePartition(newTbInfo(1, "DB_1", 11), false, 120))
+
+	// updatePartition fails if the new table is not partitioned.
+	require.Nil(t, snap.inner.dropTable(11, 130))
+	require.Nil(t, snap.inner.createTable(newTbInfo(1, "DB_1", 11), 140))
+	newTb = newTbInfo(1, "DB_1", 11)
+	newTb.Partition = nil
+	require.Error(t, snap.inner.updatePartition(newTb, false, 150))
+
+	newTb = newTbInfo(1, "DB_1", 11)
+	newTb.Partition.Definitions[0] = timodel.PartitionDefinition{ID: 11 + 65536*2}
+	require.Nil(t, snap.inner.updatePartition(newTb, true, 160))
+	require.True(t, snap.IsTruncateTableID(11+65536))
+}
+
 func TestExchangePartition(t *testing.T) {
 	var targetTb, sourceTb *model.TableInfo
 
diff --git a/cdc/entry/schema_test_helper.go b/cdc/entry/schema_test_helper.go
index e4f639e09..f438ae312 100644
--- a/cdc/entry/schema_test_helper.go
+++ b/cdc/entry/schema_test_helper.go
@@ -11,9 +11,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//go:build intest
-// +build intest
-
 package entry
 
 import (
diff --git a/cdc/owner/changefeed_test.go b/cdc/owner/changefeed_test.go
index 5d1254184..73b09a0aa 100644
--- a/cdc/owner/changefeed_test.go
+++ b/cdc/owner/changefeed_test.go
@@ -164,9 +164,14 @@ func (m *mockScheduler) Tick(
 	currentTables []model.TableID,
 	captures map[model.CaptureID]*model.CaptureInfo,
 	barrier *schedulepb.BarrierWithMinTs,
-) (newCheckpointTs, newResolvedTs model.Ts, err error) {
+) (watermark schedulepb.Watermark, err error) {
 	m.currentTables = currentTables
-	return barrier.MinTableBarrierTs, barrier.GlobalBarrierTs, nil
+	return schedulepb.Watermark{
+		CheckpointTs:     barrier.MinTableBarrierTs,
+		ResolvedTs:       barrier.GlobalBarrierTs,
+		LastSyncedTs:     scheduler.CheckpointCannotProceed,
+		PullerResolvedTs: scheduler.CheckpointCannotProceed,
+	}, nil
 }
 
 // MoveTable is used to trigger manual table moves.
diff --git a/cdc/owner/ddl_manager_test.go b/cdc/owner/ddl_manager_test.go
index 3255608be..520b6e0f0 100644
--- a/cdc/owner/ddl_manager_test.go
+++ b/cdc/owner/ddl_manager_test.go
@@ -44,6 +44,7 @@ func createDDLManagerForTest(t *testing.T) *ddlManager {
 		startTs,
 		checkpointTs,
 		ddlSink,
+		f,
 		ddlPuller,
 		schema,
 		redo.NewDisabledDDLManager(),
diff --git a/cdc/processor/sinkmanager/manager_test.go b/cdc/processor/sinkmanager/manager_test.go
index 00eeebb37..0a1bc375f 100644
--- a/cdc/processor/sinkmanager/manager_test.go
+++ b/cdc/processor/sinkmanager/manager_test.go
@@ -96,6 +96,13 @@ func addTableAndAddEventsToSortEngine(
 				CRTs:   4,
 			},
 		},
+		{
+			CRTs: 6,
+			RawKV: &model.RawKVEntry{
+				OpType: model.OpTypeResolved,
+				CRTs:   6,
+			},
+		},
 	}
 	for _, event := range events {
 		engine.Add(span, event)
@@ -178,6 +185,9 @@ func TestRemoveTable(t *testing.T) {
 
 func TestGenerateTableSinkTaskWithBarrierTs(t *testing.T) {
 	t.Parallel()
+	// This test is unstable, so skip it.
+	// issue: https://github.com/pingcap/tiflow/issues/10523
+	t.Skip("Note(dongmen): Skip this case since it is unstable.")
 
 	ctx, cancel := context.WithCancel(context.Background())
 	changefeedInfo := getChangefeedInfo()
@@ -198,10 +208,16 @@ func TestGenerateTableSinkTaskWithBarrierTs(t *testing.T) {
 	require.NoError(t, err)
 
 	require.Eventually(t, func() bool {
-		tableSink, ok := manager.tableSinks.Load(span)
-		require.True(t, ok)
-		checkpointTS := tableSink.(*tableSinkWrapper).getCheckpointTs()
-		return checkpointTS.ResolvedMark() == 4
+		s := manager.GetTableStats(span)
+		return s.CheckpointTs == 4 && s.LastSyncedTs == 4
+	}, 5*time.Second, 10*time.Millisecond)
+
+	manager.UpdateBarrierTs(6, nil)
+	manager.UpdateReceivedSorterResolvedTs(span, 6)
+	manager.schemaStorage.AdvanceResolvedTs(6)
+	require.Eventually(t, func() bool {
+		s := manager.GetTableStats(span)
+		return s.CheckpointTs == 6 && s.LastSyncedTs == 4
 	}, 5*time.Second, 10*time.Millisecond)
 }
 
@@ -229,10 +245,8 @@ func TestGenerateTableSinkTaskWithResolvedTs(t *testing.T) {
 	require.NoError(t, err)
 
 	require.Eventually(t, func() bool {
-		tableSink, ok := manager.tableSinks.Load(span)
-		require.True(t, ok)
-		checkpointTS := tableSink.(*tableSinkWrapper).getCheckpointTs()
-		return checkpointTS.ResolvedMark() == 3
+		s := manager.GetTableStats(span)
+		return s.CheckpointTs == 3 && s.LastSyncedTs == 3
 	}, 5*time.Second, 10*time.Millisecond)
 }
 
@@ -260,7 +274,7 @@ func TestGetTableStatsToReleaseMemQuota(t *testing.T) {
 
 	require.Eventually(t, func() bool {
 		s := manager.GetTableStats(span)
-		return manager.sinkMemQuota.GetUsedBytes() == 0 && s.CheckpointTs == 4
+		return manager.sinkMemQuota.GetUsedBytes() == 0 && s.CheckpointTs == 4 && s.LastSyncedTs == 4
 	}, 5*time.Second, 10*time.Millisecond)
 }
 
diff --git a/cdc/puller/ddl_puller_test.go b/cdc/puller/ddl_puller_test.go
index 05af4fe0b..e74b5ab17 100644
--- a/cdc/puller/ddl_puller_test.go
+++ b/cdc/puller/ddl_puller_test.go
@@ -437,7 +437,7 @@ func TestHandleJob(t *testing.T) {
 		job = helper.DDL2Job("alter table test1.t1 add column c1 int")
 		skip, err = ddlJobPullerImpl.handleJob(job)
 		require.NoError(t, err)
-		require.True(t, skip)
+		require.False(t, skip)
 
 		job = helper.DDL2Job("create table test1.testStartTs(id int)")
 		skip, err = ddlJobPullerImpl.handleJob(job)
@@ -448,7 +448,7 @@ func TestHandleJob(t *testing.T) {
 		job.StartTS = 1
 		skip, err = ddlJobPullerImpl.handleJob(job)
 		require.NoError(t, err)
-		require.True(t, skip)
+		require.False(t, skip)
 
 		job = helper.DDL2Job("create table test1.t2(id int)")
 		skip, err = ddlJobPullerImpl.handleJob(job)
diff --git a/cdc/puller/frontier/frontier_test.go b/cdc/puller/frontier/frontier_test.go
index 2ca327160..cb2e1cd11 100644
--- a/cdc/puller/frontier/frontier_test.go
+++ b/cdc/puller/frontier/frontier_test.go
@@ -15,6 +15,7 @@ package frontier
 
 import (
 	"bytes"
+	"fmt"
 	"math"
 	"math/rand"
 	"sort"
@@ -435,3 +436,21 @@ func TestFrontierEntries(t *testing.T) {
 	require.Equal(t, []byte("a"), []byte(slowestRange.StartKey))
 	require.Equal(t, []byte("b"), []byte(slowestRange.EndKey))
 }
+
+func TestMergeSpitWithDifferentRegionID(t *testing.T) {
+	frontier := NewFrontier(100, tablepb.Span{StartKey: []byte("a"), EndKey: []byte("c")})
+	frontier.Forward(1, tablepb.Span{StartKey: []byte("a"), EndKey: []byte("b")}, 1222)
+	frontier.Forward(2, tablepb.Span{StartKey: []byte("b"), EndKey: []byte("c")}, 102)
+	frontier.Forward(4, tablepb.Span{StartKey: []byte("b"), EndKey: []byte("c")}, 103)
+	frontier.Forward(1, tablepb.Span{StartKey: []byte("a"), EndKey: []byte("c")}, 104)
+	frontier.Forward(1, tablepb.Span{StartKey: []byte("a"), EndKey: []byte("b")}, 1223)
+	frontier.Forward(3, tablepb.Span{StartKey: []byte("b"), EndKey: []byte("c")}, 105)
+	frontier.Forward(2, tablepb.Span{StartKey: []byte("b"), EndKey: []byte("c")}, 107)
+	frontier.(*spanFrontier).spanList.Entries(func(node *skipListNode) bool {
+		fmt.Printf("%d:[%s: %s) %d\n", node.regionID,
+			string(node.Key()),
+			string(node.End()), node.value.key)
+		return true
+	})
+	require.Equal(t, uint64(107), frontier.Frontier())
+}
diff --git a/cdc/scheduler/internal/v3/coordinator_test.go b/cdc/scheduler/internal/v3/coordinator_test.go
index 8d691b4e0..8caacc30a 100644
--- a/cdc/scheduler/internal/v3/coordinator_test.go
+++ b/cdc/scheduler/internal/v3/coordinator_test.go
@@ -251,7 +251,7 @@ func TestCoordinatorHeartbeat(t *testing.T) {
 	ctx := context.Background()
 	currentTables := []model.TableID{1, 2, 3}
 	aliveCaptures := map[model.CaptureID]*model.CaptureInfo{"a": {}, "b": {}}
-	_, _, err := coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(0))
+	_, err := coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(0))
 	require.Nil(t, err)
 	msgs := trans.SendBuffer
 	require.Len(t, msgs, 2)
@@ -283,7 +283,7 @@ func TestCoordinatorHeartbeat(t *testing.T) {
 		},
 	})
 	trans.SendBuffer = []*schedulepb.Message{}
-	_, _, err = coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(0))
+	_, err = coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(0))
 	require.Nil(t, err)
 	require.True(t, coord.captureM.CheckAllCaptureInitialized())
 	msgs = trans.SendBuffer
@@ -324,7 +324,7 @@ func TestCoordinatorAddCapture(t *testing.T) {
 	ctx := context.Background()
 	currentTables := []model.TableID{1, 2, 3}
 	aliveCaptures := map[model.CaptureID]*model.CaptureInfo{"a": {}, "b": {}}
-	_, _, err = coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(0))
+	_, err = coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(0))
 	require.Nil(t, err)
 	msgs = trans.SendBuffer
 	require.Len(t, msgs, 1)
@@ -340,7 +340,7 @@ func TestCoordinatorAddCapture(t *testing.T) {
 		HeartbeatResponse: &schedulepb.HeartbeatResponse{},
 	})
 	trans.SendBuffer = []*schedulepb.Message{}
-	_, _, err = coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(0))
+	_, err = coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(0))
 	require.Nil(t, err)
 	msgs = trans.SendBuffer
 	require.Len(t, msgs, 1)
@@ -381,7 +381,7 @@ func TestCoordinatorRemoveCapture(t *testing.T) {
 	ctx := context.Background()
 	currentTables := []model.TableID{1, 2, 3}
 	aliveCaptures := map[model.CaptureID]*model.CaptureInfo{"a": {}, "b": {}}
-	_, _, err = coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(0))
+	_, err = coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(0))
 	require.Nil(t, err)
 	msgs = trans.SendBuffer
 	require.Len(t, msgs, 1)
@@ -456,7 +456,7 @@ func TestCoordinatorAdvanceCheckpoint(t *testing.T) {
 	ctx := context.Background()
 	currentTables := []model.TableID{1, 2}
 	aliveCaptures := map[model.CaptureID]*model.CaptureInfo{"a": {}, "b": {}}
-	_, _, err := coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(0))
+	_, err := coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(0))
 	require.Nil(t, err)
 
 	// Initialize captures.
@@ -482,24 +482,40 @@ func TestCoordinatorAdvanceCheckpoint(t *testing.T) {
 					Span:  spanz.TableIDToComparableSpan(1),
 					State: tablepb.TableStateReplicating,
 					Checkpoint: tablepb.Checkpoint{
-						CheckpointTs: 2, ResolvedTs: 4,
+						CheckpointTs: 2, ResolvedTs: 4, LastSyncedTs: 3,
+					},
+					Stats: tablepb.Stats{
+						StageCheckpoints: map[string]tablepb.Checkpoint{
+							"puller-egress": {
+								ResolvedTs: model.Ts(5),
+							},
+						},
 					},
 				},
 				{
 					Span:  spanz.TableIDToComparableSpan(2),
 					State: tablepb.TableStateReplicating,
 					Checkpoint: tablepb.Checkpoint{
-						CheckpointTs: 2, ResolvedTs: 4,
+						CheckpointTs: 2, ResolvedTs: 4, LastSyncedTs: 4,
+					},
+					Stats: tablepb.Stats{
+						StageCheckpoints: map[string]tablepb.Checkpoint{
+							"puller-egress": {
+								ResolvedTs: model.Ts(6),
+							},
+						},
 					},
 				},
 			},
 		},
 	})
-	cts, rts, err := coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(5))
+	watermark, err := coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(5))
 	require.Nil(t, err)
 	require.True(t, coord.captureM.CheckAllCaptureInitialized())
-	require.EqualValues(t, 2, cts)
-	require.EqualValues(t, 4, rts)
+	require.EqualValues(t, 2, watermark.CheckpointTs)
+	require.EqualValues(t, 4, watermark.ResolvedTs)
+	require.EqualValues(t, 4, watermark.LastSyncedTs)
+	require.EqualValues(t, 5, watermark.PullerResolvedTs)
 
 	// Checkpoint should be advanced even if there is an uninitialized capture.
 	aliveCaptures["c"] = &model.CaptureInfo{}
@@ -517,24 +533,40 @@ func TestCoordinatorAdvanceCheckpoint(t *testing.T) {
 					Span:  spanz.TableIDToComparableSpan(1),
 					State: tablepb.TableStateReplicating,
 					Checkpoint: tablepb.Checkpoint{
-						CheckpointTs: 3, ResolvedTs: 5,
+						CheckpointTs: 3, ResolvedTs: 5, LastSyncedTs: 4,
+					},
+					Stats: tablepb.Stats{
+						StageCheckpoints: map[string]tablepb.Checkpoint{
+							"puller-egress": {
+								ResolvedTs: model.Ts(7),
+							},
+						},
 					},
 				},
 				{
 					Span:  spanz.TableIDToComparableSpan(2),
 					State: tablepb.TableStateReplicating,
 					Checkpoint: tablepb.Checkpoint{
-						CheckpointTs: 4, ResolvedTs: 5,
+						CheckpointTs: 4, ResolvedTs: 5, LastSyncedTs: 6,
+					},
+					Stats: tablepb.Stats{
+						StageCheckpoints: map[string]tablepb.Checkpoint{
+							"puller-egress": {
+								ResolvedTs: model.Ts(7),
+							},
+						},
 					},
 				},
 			},
 		},
 	})
-	cts, rts, err = coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(5))
+	watermark, err = coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(5))
 	require.Nil(t, err)
 	require.False(t, coord.captureM.CheckAllCaptureInitialized())
-	require.EqualValues(t, 3, cts)
-	require.EqualValues(t, 5, rts)
+	require.EqualValues(t, 3, watermark.CheckpointTs)
+	require.EqualValues(t, 5, watermark.ResolvedTs)
+	require.EqualValues(t, 6, watermark.LastSyncedTs)
+	require.EqualValues(t, 7, watermark.PullerResolvedTs)
 }
 
 func TestCoordinatorDropMsgIfChangefeedEpochMismatch(t *testing.T) {
diff --git a/cdc/scheduler/internal/v3/replication/replication_manager_test.go b/cdc/scheduler/internal/v3/replication/replication_manager_test.go
index 6d8cb9f3b..98335ca23 100644
--- a/cdc/scheduler/internal/v3/replication/replication_manager_test.go
+++ b/cdc/scheduler/internal/v3/replication/replication_manager_test.go
@@ -15,6 +15,7 @@ package replication
 
 import (
 	"context"
+	"math"
 	"testing"
 	"time"
 
@@ -621,6 +622,14 @@ func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 				Checkpoint: tablepb.Checkpoint{
 					CheckpointTs: model.Ts(10),
 					ResolvedTs:   model.Ts(20),
+					LastSyncedTs: model.Ts(15),
+				},
+				Stats: tablepb.Stats{
+					StageCheckpoints: map[string]tablepb.Checkpoint{
+						"puller-egress": {
+							ResolvedTs: model.Ts(30),
+						},
+					},
 				},
 			},
 		}, model.ChangeFeedID{})
@@ -636,6 +645,14 @@ func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 				Checkpoint: tablepb.Checkpoint{
 					CheckpointTs: model.Ts(15),
 					ResolvedTs:   model.Ts(30),
+					LastSyncedTs: model.Ts(20),
+				},
+				Stats: tablepb.Stats{
+					StageCheckpoints: map[string]tablepb.Checkpoint{
+						"puller-egress": {
+							ResolvedTs: model.Ts(40),
+						},
+					},
 				},
 			},
 		}, model.ChangeFeedID{})
@@ -647,21 +664,30 @@ func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 	// no tables are replicating, resolvedTs should be advanced to globalBarrierTs and checkpoint
 	// should be advanced to minTableBarrierTs.
 	currentTables := &TableRanges{}
-	checkpoint, resolved := r.AdvanceCheckpoint(currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(5), redoMetaManager)
-	require.Equal(t, model.Ts(5), checkpoint)
-	require.Equal(t, model.Ts(5), resolved)
+	watermark := r.AdvanceCheckpoint(
+		currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(5), redoMetaManager)
+	require.Equal(t, model.Ts(5), watermark.CheckpointTs)
+	require.Equal(t, model.Ts(5), watermark.ResolvedTs)
+	require.Equal(t, model.Ts(0), watermark.LastSyncedTs)
+	require.Equal(t, model.Ts(math.MaxUint64), watermark.PullerResolvedTs)
 
 	// all tables are replicating
 	currentTables.UpdateTables([]model.TableID{1, 2})
-	checkpoint, resolved = r.AdvanceCheckpoint(currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
-	require.Equal(t, model.Ts(10), checkpoint)
-	require.Equal(t, model.Ts(20), resolved)
+	watermark = r.AdvanceCheckpoint(
+		currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
+	require.Equal(t, model.Ts(10), watermark.CheckpointTs)
+	require.Equal(t, model.Ts(20), watermark.ResolvedTs)
+	require.Equal(t, model.Ts(20), watermark.LastSyncedTs)
+	require.Equal(t, model.Ts(30), watermark.PullerResolvedTs)
 
 	// some table not exist yet.
 	currentTables.UpdateTables([]model.TableID{1, 2, 3})
-	checkpoint, resolved = r.AdvanceCheckpoint(currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
-	require.Equal(t, checkpointCannotProceed, checkpoint)
-	require.Equal(t, checkpointCannotProceed, resolved)
+	watermark = r.AdvanceCheckpoint(
+		currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
+	require.Equal(t, checkpointCannotProceed, watermark.CheckpointTs)
+	require.Equal(t, checkpointCannotProceed, watermark.ResolvedTs)
+	require.Equal(t, checkpointCannotProceed, watermark.LastSyncedTs)
+	require.Equal(t, checkpointCannotProceed, watermark.PullerResolvedTs)
 
 	span3 := spanz.TableIDToComparableSpan(3)
 	rs, err = NewReplicationSet(span3, model.Ts(5),
@@ -672,6 +698,14 @@ func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 				Checkpoint: tablepb.Checkpoint{
 					CheckpointTs: model.Ts(5),
 					ResolvedTs:   model.Ts(40),
+					LastSyncedTs: model.Ts(30),
+				},
+				Stats: tablepb.Stats{
+					StageCheckpoints: map[string]tablepb.Checkpoint{
+						"puller-egress": {
+							ResolvedTs: model.Ts(50),
+						},
+					},
 				},
 			},
 			"2": {
@@ -680,14 +714,25 @@ func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 				Checkpoint: tablepb.Checkpoint{
 					CheckpointTs: model.Ts(5),
 					ResolvedTs:   model.Ts(40),
+					LastSyncedTs: model.Ts(32),
+				},
+				Stats: tablepb.Stats{
+					StageCheckpoints: map[string]tablepb.Checkpoint{
+						"puller-egress": {
+							ResolvedTs: model.Ts(50),
+						},
+					},
 				},
 			},
 		}, model.ChangeFeedID{})
 	require.NoError(t, err)
 	r.spans.ReplaceOrInsert(span3, rs)
-	checkpoint, resolved = r.AdvanceCheckpoint(currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
-	require.Equal(t, model.Ts(5), checkpoint)
-	require.Equal(t, model.Ts(20), resolved)
+	watermark = r.AdvanceCheckpoint(
+		currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
+	require.Equal(t, model.Ts(5), watermark.CheckpointTs)
+	require.Equal(t, model.Ts(20), watermark.ResolvedTs)
+	require.Equal(t, model.Ts(32), watermark.LastSyncedTs)
+	require.Equal(t, model.Ts(30), watermark.PullerResolvedTs)
 
 	currentTables.UpdateTables([]model.TableID{1, 2, 3, 4})
 	span4 := spanz.TableIDToComparableSpan(4)
@@ -699,14 +744,25 @@ func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 				Checkpoint: tablepb.Checkpoint{
 					CheckpointTs: model.Ts(3),
 					ResolvedTs:   model.Ts(10),
+					LastSyncedTs: model.Ts(5),
+				},
+				Stats: tablepb.Stats{
+					StageCheckpoints: map[string]tablepb.Checkpoint{
+						"puller-egress": {
+							ResolvedTs: model.Ts(12),
+						},
+					},
 				},
 			},
 		}, model.ChangeFeedID{})
 	require.NoError(t, err)
 	r.spans.ReplaceOrInsert(span4, rs)
-	checkpoint, resolved = r.AdvanceCheckpoint(currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
-	require.Equal(t, model.Ts(3), checkpoint)
-	require.Equal(t, model.Ts(10), resolved)
+	watermark = r.AdvanceCheckpoint(
+		currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
+	require.Equal(t, model.Ts(3), watermark.CheckpointTs)
+	require.Equal(t, model.Ts(10), watermark.ResolvedTs)
+	require.Equal(t, model.Ts(32), watermark.LastSyncedTs)
+	require.Equal(t, model.Ts(12), watermark.PullerResolvedTs)
 
 	// Split table 5 into 2 spans.
 	currentTables.UpdateTables([]model.TableID{1, 2, 3, 4, 5})
@@ -723,28 +779,45 @@ func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 					Checkpoint: tablepb.Checkpoint{
 						CheckpointTs: model.Ts(3),
 						ResolvedTs:   model.Ts(10),
+						LastSyncedTs: model.Ts(8),
+					},
+					Stats: tablepb.Stats{
+						StageCheckpoints: map[string]tablepb.Checkpoint{
+							"puller-egress": {
+								ResolvedTs: model.Ts(11),
+							},
+						},
 					},
 				},
 			}, model.ChangeFeedID{})
 		require.NoError(t, err)
 		r.spans.ReplaceOrInsert(span, rs)
 	}
-	checkpoint, resolved = r.AdvanceCheckpoint(currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
-	require.Equal(t, model.Ts(3), checkpoint)
-	require.Equal(t, model.Ts(10), resolved)
+	watermark = r.AdvanceCheckpoint(
+		currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
+	require.Equal(t, model.Ts(3), watermark.CheckpointTs)
+	require.Equal(t, model.Ts(10), watermark.ResolvedTs)
+	require.Equal(t, model.Ts(32), watermark.LastSyncedTs)
+	require.Equal(t, model.Ts(11), watermark.PullerResolvedTs)
 
 	// The start span is missing
 	rs5_1, _ := r.spans.Delete(span5_1)
-	checkpoint, resolved = r.AdvanceCheckpoint(currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
-	require.Equal(t, checkpointCannotProceed, checkpoint)
-	require.Equal(t, checkpointCannotProceed, resolved)
+	watermark = r.AdvanceCheckpoint(
+		currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
+	require.Equal(t, checkpointCannotProceed, watermark.CheckpointTs)
+	require.Equal(t, checkpointCannotProceed, watermark.ResolvedTs)
+	require.Equal(t, checkpointCannotProceed, watermark.LastSyncedTs)
+	require.Equal(t, checkpointCannotProceed, watermark.PullerResolvedTs)
 
 	// The end span is missing
 	r.spans.ReplaceOrInsert(span5_1, rs5_1)
 	r.spans.Delete(span5_2)
-	checkpoint, resolved = r.AdvanceCheckpoint(currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
-	require.Equal(t, checkpointCannotProceed, checkpoint)
-	require.Equal(t, checkpointCannotProceed, resolved)
+	watermark = r.AdvanceCheckpoint(
+		currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
+	require.Equal(t, checkpointCannotProceed, watermark.CheckpointTs)
+	require.Equal(t, checkpointCannotProceed, watermark.ResolvedTs)
+	require.Equal(t, checkpointCannotProceed, watermark.LastSyncedTs)
+	require.Equal(t, checkpointCannotProceed, watermark.PullerResolvedTs)
 
 	// redo is enabled
 	currentTables.UpdateTables([]model.TableID{4})
@@ -757,6 +830,14 @@ func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 				Checkpoint: tablepb.Checkpoint{
 					CheckpointTs: model.Ts(10),
 					ResolvedTs:   model.Ts(15),
+					LastSyncedTs: model.Ts(12),
+				},
+				Stats: tablepb.Stats{
+					StageCheckpoints: map[string]tablepb.Checkpoint{
+						"puller-egress": {
+							ResolvedTs: model.Ts(16),
+						},
+					},
 				},
 			},
 		}, model.ChangeFeedID{})
@@ -765,9 +846,12 @@ func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 	barrier := schedulepb.NewBarrierWithMinTs(30)
 	redoMetaManager.enable = true
 	redoMetaManager.resolvedTs = 9
-	checkpoint, resolved = r.AdvanceCheckpoint(currentTables, time.Now(), barrier, redoMetaManager)
-	require.Equal(t, model.Ts(9), resolved)
-	require.Equal(t, model.Ts(9), checkpoint)
+	watermark = r.AdvanceCheckpoint(
+		currentTables, time.Now(), barrier, redoMetaManager)
+	require.Equal(t, model.Ts(9), watermark.ResolvedTs)
+	require.Equal(t, model.Ts(9), watermark.CheckpointTs)
+	require.Equal(t, model.Ts(12), watermark.LastSyncedTs)
+	require.Equal(t, model.Ts(16), watermark.PullerResolvedTs)
 	require.Equal(t, model.Ts(9), barrier.GetGlobalBarrierTs())
 }
 
@@ -783,6 +867,14 @@ func TestReplicationManagerAdvanceCheckpointWithRedoEnabled(t *testing.T) {
 				Checkpoint: tablepb.Checkpoint{
 					CheckpointTs: model.Ts(10),
 					ResolvedTs:   model.Ts(20),
+					LastSyncedTs: model.Ts(12),
+				},
+				Stats: tablepb.Stats{
+					StageCheckpoints: map[string]tablepb.Checkpoint{
+						"puller-egress": {
+							ResolvedTs: model.Ts(26),
+						},
+					},
 				},
 			},
 		}, model.ChangeFeedID{})
@@ -798,6 +890,14 @@ func TestReplicationManagerAdvanceCheckpointWithRedoEnabled(t *testing.T) {
 				Checkpoint: tablepb.Checkpoint{
 					CheckpointTs: model.Ts(15),
 					ResolvedTs:   model.Ts(30),
+					LastSyncedTs: model.Ts(18),
+				},
+				Stats: tablepb.Stats{
+					StageCheckpoints: map[string]tablepb.Checkpoint{
+						"puller-egress": {
+							ResolvedTs: model.Ts(39),
+						},
+					},
 				},
 			},
 		}, model.ChangeFeedID{})
@@ -810,9 +910,13 @@ func TestReplicationManagerAdvanceCheckpointWithRedoEnabled(t *testing.T) {
 	currentTables := &TableRanges{}
 	currentTables.UpdateTables([]model.TableID{1, 2, 3})
 	barrier := schedulepb.NewBarrierWithMinTs(30)
-	checkpoint, resolved := r.AdvanceCheckpoint(currentTables, time.Now(), barrier, redoMetaManager)
-	require.Equal(t, checkpointCannotProceed, checkpoint)
-	require.Equal(t, checkpointCannotProceed, resolved)
+	watermark := r.AdvanceCheckpoint(
+		currentTables,
+		time.Now(), barrier, redoMetaManager)
+	require.Equal(t, checkpointCannotProceed, watermark.CheckpointTs)
+	require.Equal(t, checkpointCannotProceed, watermark.ResolvedTs)
+	require.Equal(t, checkpointCannotProceed, watermark.LastSyncedTs)
+	require.Equal(t, checkpointCannotProceed, watermark.PullerResolvedTs)
 	require.Equal(t, uint64(25), barrier.Barrier.GetGlobalBarrierTs())
 }
 
diff --git a/cdc/sink/dmlsink/mq/dispatcher/event_router_test.go b/cdc/sink/dmlsink/mq/dispatcher/event_router_test.go
index f06934138..5a63c17ca 100644
--- a/cdc/sink/dmlsink/mq/dispatcher/event_router_test.go
+++ b/cdc/sink/dmlsink/mq/dispatcher/event_router_test.go
@@ -24,78 +24,112 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestEventRouter(t *testing.T) {
-	t.Parallel()
-
-	d, err := NewEventRouter(config.GetDefaultReplicaConfig(), "test")
-	require.Nil(t, err)
-	require.Equal(t, "test", d.GetDefaultTopic())
-	topicDispatcher, partitionDispatcher := d.matchDispatcher("test", "test")
-	require.IsType(t, &topic.StaticTopicDispatcher{}, topicDispatcher)
-	require.IsType(t, &partition.DefaultDispatcher{}, partitionDispatcher)
-
-	d, err = NewEventRouter(&config.ReplicaConfig{
+func newReplicaConfig4DispatcherTest() *config.ReplicaConfig {
+	return &config.ReplicaConfig{
 		Sink: &config.SinkConfig{
 			DispatchRules: []*config.DispatchRule{
+				// rule-0
 				{
 					Matcher:       []string{"test_default1.*"},
 					PartitionRule: "default",
 				},
+				// rule-1
 				{
 					Matcher:       []string{"test_default2.*"},
 					PartitionRule: "unknown-dispatcher",
 				},
+				// rule-2
 				{
 					Matcher:       []string{"test_table.*"},
 					PartitionRule: "table",
 					TopicRule:     "hello_{schema}_world",
 				},
+				// rule-3
 				{
 					Matcher:       []string{"test_index_value.*"},
 					PartitionRule: "index-value",
 					TopicRule:     "{schema}_world",
 				},
+				// rule-4
 				{
 					Matcher:       []string{"test.*"},
 					PartitionRule: "rowid",
 					TopicRule:     "hello_{schema}",
 				},
+				// rule-5
 				{
 					Matcher:       []string{"*.*", "!*.test"},
 					PartitionRule: "ts",
 					TopicRule:     "{schema}_{table}",
 				},
+				// rule-6: hard code the topic
+				{
+					Matcher:       []string{"hard_code_schema.*"},
+					PartitionRule: "default",
+					TopicRule:     "hard_code_topic",
+				},
 			},
 		},
-	}, "")
-	require.Nil(t, err)
-	topicDispatcher, partitionDispatcher = d.matchDispatcher("test", "table1")
-	require.IsType(t, &topic.DynamicTopicDispatcher{}, topicDispatcher)
-	require.IsType(t, &partition.IndexValueDispatcher{}, partitionDispatcher)
+	}
+}
 
-	topicDispatcher, partitionDispatcher = d.matchDispatcher("sbs", "table2")
-	require.IsType(t, &topic.DynamicTopicDispatcher{}, topicDispatcher)
-	require.IsType(t, &partition.TsDispatcher{}, partitionDispatcher)
+func TestEventRouter(t *testing.T) {
+	t.Parallel()
+
+	replicaConfig := config.GetDefaultReplicaConfig()
+	d, err := NewEventRouter(replicaConfig, "test", config.ProtocolCanalJSON)
+	require.NoError(t, err)
+	require.Equal(t, "test", d.GetDefaultTopic())
 
+	topicDispatcher, partitionDispatcher := d.matchDispatcher("test", "test")
+	require.IsType(t, &topic.StaticTopicDispatcher{}, topicDispatcher)
+	require.IsType(t, &partition.DefaultDispatcher{}, partitionDispatcher)
+
+	actual := topicDispatcher.Substitute("test", "test")
+	require.Equal(t, d.defaultTopic, actual)
+
+	replicaConfig = newReplicaConfig4DispatcherTest()
+	d, err = NewEventRouter(replicaConfig, "", config.ProtocolCanalJSON)
+	require.NoError(t, err)
+
+	// no matched, use the default
 	topicDispatcher, partitionDispatcher = d.matchDispatcher("sbs", "test")
 	require.IsType(t, &topic.StaticTopicDispatcher{}, topicDispatcher)
 	require.IsType(t, &partition.DefaultDispatcher{}, partitionDispatcher)
 
+	// match rule-0
 	topicDispatcher, partitionDispatcher = d.matchDispatcher("test_default1", "test")
 	require.IsType(t, &topic.StaticTopicDispatcher{}, topicDispatcher)
 	require.IsType(t, &partition.DefaultDispatcher{}, partitionDispatcher)
 
+	// match rule-1
 	topicDispatcher, partitionDispatcher = d.matchDispatcher("test_default2", "test")
 	require.IsType(t, &topic.StaticTopicDispatcher{}, topicDispatcher)
 	require.IsType(t, &partition.DefaultDispatcher{}, partitionDispatcher)
 
+	// match rule-2
 	topicDispatcher, partitionDispatcher = d.matchDispatcher("test_table", "test")
 	require.IsType(t, &topic.DynamicTopicDispatcher{}, topicDispatcher)
 	require.IsType(t, &partition.TableDispatcher{}, partitionDispatcher)
 
+	// match rule-4
 	topicDispatcher, partitionDispatcher = d.matchDispatcher("test_index_value", "test")
 	require.IsType(t, &topic.DynamicTopicDispatcher{}, topicDispatcher)
 	require.IsType(t, &partition.IndexValueDispatcher{}, partitionDispatcher)
+
+	// match rule-4
+	topicDispatcher, partitionDispatcher = d.matchDispatcher("test", "table1")
+	require.IsType(t, &topic.DynamicTopicDispatcher{}, topicDispatcher)
+	require.IsType(t, &partition.IndexValueDispatcher{}, partitionDispatcher)
+
+	topicDispatcher, partitionDispatcher = d.matchDispatcher("sbs", "table2")
+	require.IsType(t, &topic.DynamicTopicDispatcher{}, topicDispatcher)
+	require.IsType(t, &partition.TsDispatcher{}, partitionDispatcher)
+
+	// match rule-6
+	topicDispatcher, partitionDispatcher = d.matchDispatcher("hard_code_schema", "test")
+	require.IsType(t, &topic.StaticTopicDispatcher{}, topicDispatcher)
+	require.IsType(t, &partition.DefaultDispatcher{}, partitionDispatcher)
 }
 
 func TestGetActiveTopics(t *testing.T) {
@@ -134,7 +168,7 @@ func TestGetActiveTopics(t *testing.T) {
 				},
 			},
 		},
-	}, "test")
+	}, "test", config.ProtocolCanalJSON)
 	require.Nil(t, err)
 	names := []model.TableName{
 		{Schema: "test_default1", Table: "table"},
@@ -184,7 +218,7 @@ func TestGetTopicForRowChange(t *testing.T) {
 				},
 			},
 		},
-	}, "test")
+	}, "test", config.ProtocolCanalJSON)
 	require.Nil(t, err)
 
 	topicName := d.GetTopicForRowChange(&model.RowChangedEvent{
@@ -245,7 +279,7 @@ func TestGetPartitionForRowChange(t *testing.T) {
 				},
 			},
 		},
-	}, "test")
+	}, "test", config.ProtocolCanalJSON)
 	require.Nil(t, err)
 
 	p := d.GetPartitionForRowChange(&model.RowChangedEvent{
@@ -313,7 +347,7 @@ func TestGetDLLDispatchRuleByProtocol(t *testing.T) {
 				},
 			},
 		},
-	}, "test")
+	}, "test", config.ProtocolCanalJSON)
 	require.Nil(t, err)
 
 	tests := []struct {
@@ -374,7 +408,7 @@ func TestGetTopicForDDL(t *testing.T) {
 				},
 			},
 		},
-	}, "test")
+	}, "test", config.ProtocolCanalJSON)
 	require.Nil(t, err)
 
 	tests := []struct {
diff --git a/cdc/sink/dmlsink/mq/dispatcher/topic/dispatcher_test.go b/cdc/sink/dmlsink/mq/dispatcher/topic/dispatcher_test.go
index 73974317f..c387ac0e8 100644
--- a/cdc/sink/dmlsink/mq/dispatcher/topic/dispatcher_test.go
+++ b/cdc/sink/dmlsink/mq/dispatcher/topic/dispatcher_test.go
@@ -29,7 +29,8 @@ func TestDynamicTopicDispatcherForSchema(t *testing.T) {
 
 	topicExpr := Expression("hello_{schema}_world")
 	err := topicExpr.Validate()
-	require.Nil(t, err)
+	require.NoError(t, err)
+
 	testCase := []struct {
 		schema      string
 		table       string
@@ -68,7 +69,8 @@ func TestDynamicTopicDispatcherForTable(t *testing.T) {
 
 	topicExpr := Expression("{schema}_{table}")
 	err := topicExpr.Validate()
-	require.Nil(t, err)
+	require.NoError(t, err)
+
 	testCases := []struct {
 		schema        string
 		table         string
diff --git a/cdc/sink/dmlsink/mq/dispatcher/topic/expression_test.go b/cdc/sink/dmlsink/mq/dispatcher/topic/expression_test.go
index c02bdc7ff..070ae718d 100644
--- a/cdc/sink/dmlsink/mq/dispatcher/topic/expression_test.go
+++ b/cdc/sink/dmlsink/mq/dispatcher/topic/expression_test.go
@@ -227,6 +227,30 @@ func TestSubstituteTopicExpression(t *testing.T) {
 	}
 }
 
+func TestSchemaOptional(t *testing.T) {
+	expression := "prefix_{table}"
+	topicExpr := Expression(expression)
+	err := topicExpr.Validate()
+	require.NoError(t, err)
+
+	schemaName := "test"
+	tableName := "table1"
+	topicName := topicExpr.Substitute(schemaName, tableName)
+	require.Equal(t, topicName, "prefix_table1")
+}
+
+func TestTableOptional(t *testing.T) {
+	expression := "prefix_{schema}"
+	topicExpr := Expression(expression)
+	err := topicExpr.Validate()
+	require.NoError(t, err)
+
+	schemaName := "test"
+	tableName := "abc"
+	topicName := topicExpr.Substitute(schemaName, tableName)
+	require.Equal(t, topicName, "prefix_test")
+}
+
 // cmd: go test -run='^$' -bench '^(BenchmarkSubstitute)$' github.com/pingcap/tiflow/cdc/sink/dispatcher/topic
 // goos: linux
 // goarch: amd64
diff --git a/cdc/sink/dmlsink/txn/event_test.go b/cdc/sink/dmlsink/txn/event_test.go
index c0dfb0049..c5899757f 100644
--- a/cdc/sink/dmlsink/txn/event_test.go
+++ b/cdc/sink/dmlsink/txn/event_test.go
@@ -203,3 +203,30 @@ func TestGenKeys(t *testing.T) {
 		require.Equal(t, tc.expected, keys)
 	}
 }
+
+func TestSortAndDedupHash(t *testing.T) {
+	// If a transaction contains multiple rows, these rows may generate the same hash
+	// in some rare cases. We should dedup these hashes to avoid unnecessary self cyclic
+	// dependency in the causality dependency graph.
+	t.Parallel()
+	testCases := []struct {
+		hashes   []uint64
+		expected []uint64
+	}{{
+		// No duplicate hashes
+		hashes:   []uint64{1, 2, 3, 4, 5},
+		expected: []uint64{1, 2, 3, 4, 5},
+	}, {
+		// Duplicate hashes
+		hashes:   []uint64{1, 2, 3, 4, 5, 1, 2, 3, 4, 5},
+		expected: []uint64{1, 2, 3, 4, 5},
+	}, {
+		// Has hash value larger than slots count, should sort by `hash % numSlots` first.
+		hashes:   []uint64{4, 9, 9, 3},
+		expected: []uint64{9, 3, 4},
+	}}
+
+	for _, tc := range testCases {
+		require.Equal(t, tc.expected, sortAndDedupHashes(tc.hashes, 8))
+	}
+}
diff --git a/cdc/sink/tablesink/table_sink_impl_test.go b/cdc/sink/tablesink/table_sink_impl_test.go
index 9381a7c81..0c8ce77b3 100644
--- a/cdc/sink/tablesink/table_sink_impl_test.go
+++ b/cdc/sink/tablesink/table_sink_impl_test.go
@@ -247,6 +247,7 @@ func TestGetCheckpointTs(t *testing.T) {
 
 	tb.AppendRowChangedEvents(getTestRows()...)
 	require.Equal(t, model.NewResolvedTs(0), tb.GetCheckpointTs(), "checkpointTs should be 0")
+	require.Equal(t, tb.lastSyncedTs.getLastSyncedTs(), uint64(0), "lastSyncedTs should be not updated")
 
 	// One event will be flushed.
 	err := tb.UpdateResolvedTs(model.NewResolvedTs(101))
@@ -254,11 +255,13 @@ func TestGetCheckpointTs(t *testing.T) {
 	require.Equal(t, model.NewResolvedTs(0), tb.GetCheckpointTs(), "checkpointTs should be 0")
 	sink.acknowledge(101)
 	require.Equal(t, model.NewResolvedTs(101), tb.GetCheckpointTs(), "checkpointTs should be 101")
+	require.Equal(t, tb.lastSyncedTs.getLastSyncedTs(), uint64(101), "lastSyncedTs should be the same as the flushed event")
 
 	// Flush all events.
 	err = tb.UpdateResolvedTs(model.NewResolvedTs(105))
 	require.Nil(t, err)
 	require.Equal(t, model.NewResolvedTs(101), tb.GetCheckpointTs(), "checkpointTs should be 101")
+	require.Equal(t, tb.lastSyncedTs.getLastSyncedTs(), uint64(101), "lastSyncedTs should be not updated")
 
 	// Only acknowledge some events.
 	sink.acknowledge(102)
@@ -268,10 +271,12 @@ func TestGetCheckpointTs(t *testing.T) {
 		tb.GetCheckpointTs(),
 		"checkpointTs should still be 101",
 	)
+	require.Equal(t, tb.lastSyncedTs.getLastSyncedTs(), uint64(102), "lastSyncedTs should be updated")
 
 	// Ack all events.
 	sink.acknowledge(105)
 	require.Equal(t, model.NewResolvedTs(105), tb.GetCheckpointTs(), "checkpointTs should be 105")
+	require.Equal(t, tb.lastSyncedTs.getLastSyncedTs(), uint64(105), "lastSyncedTs should be updated")
 }
 
 func TestClose(t *testing.T) {
@@ -404,4 +409,5 @@ func TestCheckpointTsFrozenWhenStopping(t *testing.T) {
 	currentTs := tb.GetCheckpointTs()
 	sink.acknowledge(105)
 	require.Equal(t, currentTs, tb.GetCheckpointTs(), "checkpointTs should not be updated")
+	require.Equal(t, tb.lastSyncedTs.getLastSyncedTs(), uint64(105), "lastSyncedTs should not change")
 }
diff --git a/pkg/causality/tests/integration_test.go b/pkg/causality/tests/integration_test.go
index 9ed509531..35de5533d 100644
--- a/pkg/causality/tests/integration_test.go
+++ b/pkg/causality/tests/integration_test.go
@@ -40,7 +40,7 @@ func TestConflictBasics(t *testing.T) {
 		numWorkers, numSlots, newUniformGenerator(workingSetSize, batchSize, numSlots),
 	).WithExecFunc(
 		func(txn *txnForTest) error {
-			for _, key := range txn.ConflictKeys(numSlots) {
+			for _, key := range txn.GenSortedDedupKeysHash(numSlots) {
 				// Access a position in the array without synchronization,
 				// so that if causality check is buggy, the Go race detection would fail.
 				conflictArray[key]++
diff --git a/pkg/causality/tests/worker.go b/pkg/causality/tests/worker.go
index 901cc24f5..f78f406ba 100644
--- a/pkg/causality/tests/worker.go
+++ b/pkg/causality/tests/worker.go
@@ -27,7 +27,7 @@ type txnForTest struct {
 
 func (t *txnForTest) OnConflictResolved() {}
 
-func (t *txnForTest) ConflictKeys(numSlots uint64) []uint64 {
+func (t *txnForTest) GenSortedDedupKeysHash(numSlots uint64) []uint64 {
 	return t.keys
 }
 
diff --git a/pkg/config/config_test_data.go b/pkg/config/config_test_data.go
index 846d523ac..1499cf5e7 100644
--- a/pkg/config/config_test_data.go
+++ b/pkg/config/config_test_data.go
@@ -80,7 +80,11 @@ const (
     "corruption-handle-level": "warn"
  },
   "changefeed-error-stuck-duration": 1800000000000,
-  "sql-mode":"ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION"
+  "sql-mode":"ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION",
+  "synced-status": {
+    "synced-check-interval": 300,
+    "checkpoint-interval": 15
+  }
 }`
 
 	testCfgTestServerConfigMarshal = `{
@@ -309,7 +313,11 @@ const (
     "corruption-handle-level": "warn"
   },
   "changefeed-error-stuck-duration": 1800000000000,
-  "sql-mode":"ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION"
+  "sql-mode":"ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION",
+  "synced-status": {
+    "synced-check-interval": 300,
+    "checkpoint-interval": 15
+  }
 }`
 
 	testCfgTestReplicaConfigMarshal2 = `{
@@ -449,6 +457,10 @@ const (
     "corruption-handle-level": "warn"
   },
   "changefeed-error-stuck-duration": 1800000000000,
-  "sql-mode":"ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION"
+  "sql-mode":"ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION",
+  "synced-status": {
+    "synced-check-interval": 300,
+    "checkpoint-interval": 15
+  }
 }`
 )
diff --git a/pkg/filter/filter_test.go b/pkg/filter/filter_test.go
index 33cd51d69..9d59ec5dd 100644
--- a/pkg/filter/filter_test.go
+++ b/pkg/filter/filter_test.go
@@ -152,8 +152,9 @@ func TestShouldIgnoreDMLEvent(t *testing.T) {
 	}
 }
 
-func TestShouldDiscardDDL(t *testing.T) {
+func TestShouldIgnoreDDL(t *testing.T) {
 	t.Parallel()
+
 	testCases := []struct {
 		cases []struct {
 			startTs uint64
@@ -166,137 +167,199 @@ func TestShouldDiscardDDL(t *testing.T) {
 		rules        []string
 		ignoredTs    []uint64
 		eventFilters []*config.EventFilterRule
-	}{{
-		// Ignore by table name cases.
-		cases: []struct {
-			startTs uint64
-			schema  string
-			table   string
-			query   string
-			ddlType timodel.ActionType
-			ignore  bool
-		}{
-			{1, "sns", "", "create database test", timodel.ActionCreateSchema, false},
-			{1, "sns", "", "drop database test", timodel.ActionDropSchema, false},
-			{
-				1, "sns", "", "ALTER DATABASE dbname CHARACTER SET utf8 COLLATE utf8_general_ci",
-				timodel.ActionModifySchemaCharsetAndCollate, false,
+	}{
+		{ // cases ignore by startTs
+			cases: []struct {
+				startTs uint64
+				schema  string
+				table   string
+				query   string
+				ddlType timodel.ActionType
+				ignore  bool
+			}{
+				{1, "ts", "", "create database test", timodel.ActionCreateSchema, true},
+				{2, "ts", "student", "drop database test2", timodel.ActionDropSchema, true},
+				{
+					3, "ts", "teacher", "ALTER DATABASE dbname CHARACTER SET utf8 COLLATE utf8_general_ci",
+					timodel.ActionModifySchemaCharsetAndCollate, true,
+				},
+				{4, "ts", "man", "create table test.t1(a int primary key)", timodel.ActionCreateTable, false},
+				{5, "ts", "fruit", "create table test.t1(a int primary key)", timodel.ActionCreateTable, false},
+				{6, "ts", "insect", "create table test.t1(a int primary key)", timodel.ActionCreateTable, false},
 			},
-			{1, "ecom", "", "create database test", timodel.ActionCreateSchema, false},
-			{1, "ecom", "aa", "create table test.t1(a int primary key)", timodel.ActionCreateTable, false},
-			{1, "ecom", "", "create database test", timodel.ActionCreateSchema, false},
-			{1, "test", "", "create database test", timodel.ActionCreateSchema, true},
+			rules:     []string{"*.*"},
+			ignoredTs: []uint64{1, 2, 3},
 		},
-		rules: []string{"sns.*", "ecom.*", "!sns.log", "!ecom.test"},
-		// Ignore by schema name cases.
-	}, {
-		cases: []struct {
-			startTs uint64
-			schema  string
-			table   string
-			query   string
-			ddlType timodel.ActionType
-			ignore  bool
-		}{
-			{1, "schema", "C1", "create database test", timodel.ActionCreateSchema, false},
-			{1, "test", "", "drop database test1", timodel.ActionDropSchema, true},
-			{
-				1, "dbname", "", "ALTER DATABASE dbname CHARACTER SET utf8 COLLATE utf8_general_ci",
-				timodel.ActionModifySchemaCharsetAndCollate, true,
+		{ // cases ignore by ddl type.
+			cases: []struct {
+				startTs uint64
+				schema  string
+				table   string
+				query   string
+				ddlType timodel.ActionType
+				ignore  bool
+			}{
+				{1, "event", "", "drop table t1", timodel.ActionDropTable, true},
+				{1, "event", "January", "drop index i on t1", timodel.ActionDropIndex, true},
+				{1, "event", "February", "drop index x2 on t2", timodel.ActionDropIndex, true},
+				{1, "event", "March", "create table t2(age int)", timodel.ActionCreateTable, false},
+				{1, "event", "April", "create table t2(age int)", timodel.ActionCreateTable, false},
+				{1, "event", "May", "create table t2(age int)", timodel.ActionCreateTable, false},
 			},
-			{1, "test", "aa", "create table test.t1(a int primary key)", timodel.ActionCreateTable, true},
-			{1, "schema", "C1", "create table test.t1(a int primary key)", timodel.ActionCreateTable, false},
-			{1, "schema", "", "create table test.t1(a int primary key)", timodel.ActionCreateTable, true},
-		},
-		rules: []string{"schema.C1"},
-	}, { // cases ignore by startTs
-		cases: []struct {
-			startTs uint64
-			schema  string
-			table   string
-			query   string
-			ddlType timodel.ActionType
-			ignore  bool
-		}{
-			{1, "ts", "", "create database test", timodel.ActionCreateSchema, true},
-			{2, "ts", "student", "drop database test2", timodel.ActionDropSchema, true},
-			{
-				3, "ts", "teacher", "ALTER DATABASE dbname CHARACTER SET utf8 COLLATE utf8_general_ci",
-				timodel.ActionModifySchemaCharsetAndCollate, true,
+			rules: []string{"*.*"},
+			eventFilters: []*config.EventFilterRule{
+				{
+					Matcher: []string{"event.*"},
+					IgnoreEvent: []bf.EventType{
+						bf.AlterTable, bf.DropTable,
+					},
+				},
 			},
-			{4, "ts", "man", "create table test.t1(a int primary key)", timodel.ActionCreateTable, false},
-			{5, "ts", "fruit", "create table test.t1(a int primary key)", timodel.ActionCreateTable, false},
-			{6, "ts", "insect", "create table test.t1(a int primary key)", timodel.ActionCreateTable, false},
+			ignoredTs: []uint64{},
 		},
-		rules:     []string{"*.*"},
-		ignoredTs: []uint64{1, 2, 3},
-	}, { // cases ignore by ddl type.
-		cases: []struct {
-			startTs uint64
-			schema  string
-			table   string
-			query   string
-			ddlType timodel.ActionType
-			ignore  bool
-		}{
-			{1, "event", "", "drop table t1", timodel.ActionDropTable, true},
-			{1, "event", "January", "drop index i on t1", timodel.ActionDropIndex, true},
-			{1, "event", "February", "drop index x2 on t2", timodel.ActionDropIndex, true},
-			{1, "event", "March", "create table t2(age int)", timodel.ActionCreateTable, false},
-			{1, "event", "April", "create table t2(age int)", timodel.ActionCreateTable, false},
-			{1, "event", "May", "create table t2(age int)", timodel.ActionCreateTable, false},
-		},
-		rules: []string{"*.*"},
-		eventFilters: []*config.EventFilterRule{
-			{
-				Matcher: []string{"event.*"},
-				IgnoreEvent: []bf.EventType{
-					bf.AlterTable, bf.DropTable,
+		{ // cases ignore by ddl query
+			cases: []struct {
+				startTs uint64
+				schema  string
+				table   string
+				query   string
+				ddlType timodel.ActionType
+				ignore  bool
+			}{
+				{1, "sql_pattern", "t1", "CREATE DATABASE sql_pattern", timodel.ActionCreateSchema, false},
+				{1, "sql_pattern", "t1", "DROP DATABASE sql_pattern", timodel.ActionDropSchema, true},
+				{
+					1, "sql_pattern", "t1",
+					"ALTER DATABASE `test_db` CHARACTER SET 'utf8' COLLATE 'utf8_general_ci'",
+					timodel.ActionModifySchemaCharsetAndCollate,
+					false,
 				},
+				{1, "sql_pattern", "t1", "CREATE TABLE t1(id int primary key)", timodel.ActionCreateTable, false},
+				{1, "sql_pattern", "t1", "DROP TABLE t1", timodel.ActionDropTable, true},
+				{1, "sql_pattern", "t1", "CREATE VIEW test.v AS SELECT * FROM t", timodel.ActionCreateView, true},
 			},
+			rules: []string{"*.*"},
+			eventFilters: []*config.EventFilterRule{
+				{
+					Matcher:   []string{"sql_pattern.*"},
+					IgnoreSQL: []string{"^DROP TABLE", "^CREATE VIEW", "^DROP DATABASE"},
+				},
+			},
+			ignoredTs: []uint64{},
 		},
-	}, { // cases ignore by ddl query
-		cases: []struct {
-			startTs uint64
+	}
+
+	for _, ftc := range testCases {
+		filter, err := NewFilter(&config.ReplicaConfig{
+			Filter: &config.FilterConfig{
+				Rules:            ftc.rules,
+				EventFilters:     ftc.eventFilters,
+				IgnoreTxnStartTs: ftc.ignoredTs,
+			},
+		}, "")
+		require.Nil(t, err)
+		for _, tc := range ftc.cases {
+			ddl := &model.DDLEvent{
+				StartTs: tc.startTs,
+				TableInfo: &model.TableInfo{
+					TableName: model.TableName{
+						Schema: tc.schema,
+						Table:  tc.table,
+					},
+				},
+				Query: tc.query,
+				Type:  tc.ddlType,
+			}
+			ignore, err := filter.ShouldIgnoreDDLEvent(ddl)
+			require.NoError(t, err)
+			require.Equal(t, tc.ignore, ignore, "%#v", tc)
+		}
+	}
+}
+
+func TestShouldDiscardDDL(t *testing.T) {
+	t.Parallel()
+	testCases := []struct {
+		cases []struct {
 			schema  string
 			table   string
 			query   string
 			ddlType timodel.ActionType
 			ignore  bool
-		}{
-			{1, "sql_pattern", "t1", "CREATE DATABASE sql_pattern", timodel.ActionCreateSchema, false},
-			{1, "sql_pattern", "t1", "DROP DATABASE sql_pattern", timodel.ActionDropSchema, true},
-			{
-				1, "sql_pattern", "t1",
-				"ALTER DATABASE `test_db` CHARACTER SET 'utf8' COLLATE 'utf8_general_ci'",
-				timodel.ActionModifySchemaCharsetAndCollate,
-				false,
+		}
+		rules        []string
+		eventFilters []*config.EventFilterRule
+	}{
+		{
+			// Discard by not allowed DDL type cases.
+			cases: []struct {
+				schema  string
+				table   string
+				query   string
+				ddlType timodel.ActionType
+				ignore  bool
+			}{
+				{"sns", "", "create database test", timodel.ActionCreateSchema, false},
+				{"sns", "", "drop database test", timodel.ActionDropSchema, false},
+				{"test", "", "create database test", timodel.ActionCreateSequence, true},
 			},
-			{1, "sql_pattern", "t1", "CREATE TABLE t1(id int primary key)", timodel.ActionCreateTable, false},
-			{1, "sql_pattern", "t1", "DROP TABLE t1", timodel.ActionDropTable, true},
-			{1, "sql_pattern", "t1", "CREATE VIEW test.v AS SELECT * FROM t", timodel.ActionCreateView, true},
+			rules: []string{"*.*"},
 		},
-		rules: []string{"*.*"},
-		eventFilters: []*config.EventFilterRule{
-			{
-				Matcher:   []string{"sql_pattern.*"},
-				IgnoreSQL: []string{"^DROP TABLE", "^CREATE VIEW", "^DROP DATABASE"},
+		{
+			// Discard by table name cases.
+			cases: []struct {
+				schema  string
+				table   string
+				query   string
+				ddlType timodel.ActionType
+				ignore  bool
+			}{
+				{"sns", "", "create database test", timodel.ActionCreateSchema, false},
+				{"sns", "", "drop database test", timodel.ActionDropSchema, false},
+				{
+					"sns", "", "ALTER DATABASE dbname CHARACTER SET utf8 COLLATE utf8_general_ci",
+					timodel.ActionModifySchemaCharsetAndCollate, false,
+				},
+				{"ecom", "", "create database test", timodel.ActionCreateSchema, false},
+				{"ecom", "aa", "create table test.t1(a int primary key)", timodel.ActionCreateTable, false},
+				{"ecom", "", "create database test", timodel.ActionCreateSchema, false},
+				{"test", "", "create database test", timodel.ActionCreateSchema, true},
 			},
+			rules: []string{"sns.*", "ecom.*", "!sns.log", "!ecom.test"},
 		},
-	}}
+		{
+			// Discard by schema name cases.
+			cases: []struct {
+				schema  string
+				table   string
+				query   string
+				ddlType timodel.ActionType
+				ignore  bool
+			}{
+				{"schema", "C1", "create database test", timodel.ActionCreateSchema, false},
+				{"test", "", "drop database test1", timodel.ActionDropSchema, true},
+				{
+					"dbname", "", "ALTER DATABASE dbname CHARACTER SET utf8 COLLATE utf8_general_ci",
+					timodel.ActionModifySchemaCharsetAndCollate, true,
+				},
+				{"test", "aa", "create table test.t1(a int primary key)", timodel.ActionCreateTable, true},
+				{"schema", "C1", "create table test.t1(a int primary key)", timodel.ActionCreateTable, false},
+				{"schema", "", "create table test.t1(a int primary key)", timodel.ActionCreateTable, true},
+			},
+			rules: []string{"schema.C1"},
+		},
+	}
 
 	for _, ftc := range testCases {
 		filter, err := NewFilter(&config.ReplicaConfig{
 			Filter: &config.FilterConfig{
-				Rules:            ftc.rules,
-				IgnoreTxnStartTs: ftc.ignoredTs,
-				EventFilters:     ftc.eventFilters,
+				Rules:        ftc.rules,
+				EventFilters: ftc.eventFilters,
 			},
 		}, "")
 		require.Nil(t, err)
 		for _, tc := range ftc.cases {
-			ignore, err := filter.ShouldDiscardDDL(tc.startTs, tc.ddlType, tc.schema, tc.table, tc.query)
-			require.Nil(t, err, "%#v", tc)
+			ignore := filter.ShouldDiscardDDL(tc.ddlType, tc.schema, tc.table)
 			require.Equal(t, tc.ignore, ignore, "%#v", tc)
 		}
 	}
diff --git a/pkg/filter/sql_event_filter_test.go b/pkg/filter/sql_event_filter_test.go
index 9e9707d07..9b27fa2b9 100644
--- a/pkg/filter/sql_event_filter_test.go
+++ b/pkg/filter/sql_event_filter_test.go
@@ -183,8 +183,7 @@ func TestShouldSkipDDL(t *testing.T) {
 				},
 				Query: c.query,
 			}
-			skip, err := f.shouldSkipDDL(ddl.Type,
-				ddl.TableInfo.TableName.Schema, ddl.TableInfo.TableName.Table, ddl.Query)
+			skip, err := f.shouldSkipDDL(ddl)
 			require.NoError(t, err)
 			require.Equal(t, c.skip, skip, "case: %+v", c)
 		}
diff --git a/pkg/orchestrator/reactor_state_test.go b/pkg/orchestrator/reactor_state_test.go
index 64998c6d1..f4285c3a1 100644
--- a/pkg/orchestrator/reactor_state_test.go
+++ b/pkg/orchestrator/reactor_state_test.go
@@ -130,7 +130,8 @@ func TestChangefeedStateUpdate(t *testing.T) {
 						Integrity:  config.GetDefaultReplicaConfig().Integrity,
 						ChangefeedErrorStuckDuration: config.
 							GetDefaultReplicaConfig().ChangefeedErrorStuckDuration,
-						SQLMode: config.GetDefaultReplicaConfig().SQLMode,
+						SQLMode:      config.GetDefaultReplicaConfig().SQLMode,
+						SyncedStatus: config.GetDefaultReplicaConfig().SyncedStatus,
 					},
 				},
 				Status: &model.ChangeFeedStatus{CheckpointTs: 421980719742451713},
@@ -186,7 +187,8 @@ func TestChangefeedStateUpdate(t *testing.T) {
 						Integrity:  config.GetDefaultReplicaConfig().Integrity,
 						ChangefeedErrorStuckDuration: config.
 							GetDefaultReplicaConfig().ChangefeedErrorStuckDuration,
-						SQLMode: config.GetDefaultReplicaConfig().SQLMode,
+						SQLMode:      config.GetDefaultReplicaConfig().SQLMode,
+						SyncedStatus: config.GetDefaultReplicaConfig().SyncedStatus,
 					},
 				},
 				Status: &model.ChangeFeedStatus{CheckpointTs: 421980719742451713},
@@ -247,7 +249,8 @@ func TestChangefeedStateUpdate(t *testing.T) {
 						Integrity:  config.GetDefaultReplicaConfig().Integrity,
 						ChangefeedErrorStuckDuration: config.
 							GetDefaultReplicaConfig().ChangefeedErrorStuckDuration,
-						SQLMode: config.GetDefaultReplicaConfig().SQLMode,
+						SQLMode:      config.GetDefaultReplicaConfig().SQLMode,
+						SyncedStatus: config.GetDefaultReplicaConfig().SyncedStatus,
 					},
 				},
 				Status: &model.ChangeFeedStatus{CheckpointTs: 421980719742451713},
@@ -340,6 +343,7 @@ func TestPatchInfo(t *testing.T) {
 			Integrity:                    defaultConfig.Integrity,
 			ChangefeedErrorStuckDuration: defaultConfig.ChangefeedErrorStuckDuration,
 			SQLMode:                      defaultConfig.SQLMode,
+			SyncedStatus:                 defaultConfig.SyncedStatus,
 		},
 	})
 	state.PatchInfo(func(info *model.ChangeFeedInfo) (*model.ChangeFeedInfo, bool, error) {
@@ -360,6 +364,7 @@ func TestPatchInfo(t *testing.T) {
 			Integrity:                    defaultConfig.Integrity,
 			ChangefeedErrorStuckDuration: defaultConfig.ChangefeedErrorStuckDuration,
 			SQLMode:                      defaultConfig.SQLMode,
+			SyncedStatus:                 defaultConfig.SyncedStatus,
 		},
 	})
 	state.PatchInfo(func(info *model.ChangeFeedInfo) (*model.ChangeFeedInfo, bool, error) {
diff --git a/tests/integration_tests/event_filter/conf/cf.toml b/tests/integration_tests/event_filter/conf/cf.toml
index 2ea6083c5..d3a36188b 100644
--- a/tests/integration_tests/event_filter/conf/cf.toml
+++ b/tests/integration_tests/event_filter/conf/cf.toml
@@ -6,3 +6,10 @@ matcher = ["event_filter.t1"]
 ignore-event = ["drop table", "delete"]
 ignore-insert-value-expr = "id = 2 or city = 'tokyo'"
 
+[[filter.event-filters]]
+matcher = ["event_filter.t_truncate"]
+ignore-event = ["truncate table"]
+
+[[filter.event-filters]]
+matcher = ["event_filter.t_alter"]
+ignore-event = ["alter table"]
diff --git a/tests/integration_tests/event_filter/conf/diff_config.toml b/tests/integration_tests/event_filter/conf/diff_config.toml
index 75593ed89..fff11e2ee 100644
--- a/tests/integration_tests/event_filter/conf/diff_config.toml
+++ b/tests/integration_tests/event_filter/conf/diff_config.toml
@@ -13,7 +13,7 @@ source-instances = ["tidb0"]
 
 target-instance = "mysql1"
 
-target-check-tables = ["event_filter.t2"]
+target-check-tables = ["event_filter.t_normal", "event_filter.t_truncate", "event_filter.t_alter"]
 
 [data-sources]
 [data-sources.tidb0]
diff --git a/tests/integration_tests/event_filter/data/test.sql b/tests/integration_tests/event_filter/data/test.sql
index a3ff2e479..6af0efed2 100644
--- a/tests/integration_tests/event_filter/data/test.sql
+++ b/tests/integration_tests/event_filter/data/test.sql
@@ -38,8 +38,8 @@ WHERE id = 4;
 /* ignore by event type*/
 DROP TABLE t1;
 
-/* all event of t2 will be replicated to downstream */
-CREATE TABLE t2 (
+/* all event of t_normal will be replicated to downstream */
+CREATE TABLE t_normal (
                     id INT,
                     name varchar(128),
                     country char(32),
@@ -48,16 +48,45 @@ CREATE TABLE t2 (
                     gender char(32),
                     PRIMARY KEY (id)
 );
-INSERT INTO t2
+INSERT INTO t_normal
 VALUES (1, 'guagua', "china", "chengdu", 1, "female");
 
-INSERT INTO t2
+INSERT INTO t_normal
 VALUES (2, 'huahua', "china", "chengdu", 2, "female");
 
-INSERT INTO t2
+INSERT INTO t_normal
 VALUES (3, 'xigua', "japan", "tokyo", 2, "male");
 
-INSERT INTO t2
+INSERT INTO t_normal
 VALUES (4, 'yuko', "japan", "nagoya", 33, "female");
 
-create table finish_mark(id int primary key);
+CREATE TABLE t_truncate (
+                    id INT,
+                    name varchar(128),
+                    PRIMARY KEY (id)
+);
+CREATE TABLE t_alter
+(
+    id          INT AUTO_INCREMENT,
+    t_boolean   BOOLEAN,
+    t_bigint    DECIMAL(38, 19),
+    t_double    DOUBLE,
+    t_decimal   DECIMAL(38, 19),
+    t_bit       BIT(64),
+    t_date      DATE,
+    t_datetime  DATETIME,
+    t_timestamp TIMESTAMP NULL,
+    t_time      TIME,
+    t_year      YEAR,
+    t_char      CHAR,
+    t_varchar   VARCHAR(10),
+    t_blob      BLOB,
+    t_text      TEXT,
+    t_enum      ENUM ('enum1', 'enum2', 'enum3'),
+    t_set       SET ('a', 'b', 'c'),
+    t_json      JSON,
+    PRIMARY KEY (id)
+) ENGINE = InnoDB
+  DEFAULT CHARSET = utf8
+  COLLATE = utf8_bin;
+  
\ No newline at end of file
diff --git a/tests/integration_tests/event_filter/data/test_alter.sql b/tests/integration_tests/event_filter/data/test_alter.sql
new file mode 100644
index 000000000..0e6fbabd2
--- /dev/null
+++ b/tests/integration_tests/event_filter/data/test_alter.sql
@@ -0,0 +1,49 @@
+USE `event_filter`;
+
+/* test  */
+ALTER TABLE t_alter MODIFY t_bigint BIGINT;
+
+-- make sure `nullable` can be handled by the mounter and mq encoding protocol
+INSERT INTO t_alter() VALUES ();
+
+INSERT INTO t_alter( t_boolean, t_bigint, t_double, t_decimal, t_bit
+                           , t_date, t_datetime, t_timestamp, t_time, t_year
+                           , t_char, t_varchar, t_blob, t_text, t_enum
+                           , t_set, t_json)
+VALUES ( true, 9223372036854775807, 123.123, 123456789012.123456789012, b'1000001'
+       , '1000-01-01', '9999-12-31 23:59:59', '19731230153000', '23:59:59', 1970
+       , '测', '测试', 'blob', '测试text', 'enum2'
+       , 'a,b', NULL);
+
+INSERT INTO t_alter( t_boolean, t_bigint, t_double, t_decimal, t_bit
+                           , t_date, t_datetime, t_timestamp, t_time, t_year
+                           , t_char, t_varchar, t_blob, t_text, t_enum
+                           , t_set, t_json)
+VALUES ( false, 666, 123.777, 123456789012.123456789012, b'1000001'
+       , '1000-01-01', '9999-12-31 23:59:59', '19731230153000', '23:59:59', 1970
+       , '测', '测试', 'blob', '测试text11', 'enum3'
+       , 'a,b', NULL);
+
+UPDATE t_alter
+SET t_bigint = 555
+WHERE id = 1;
+
+INSERT INTO t_alter( t_boolean, t_bigint, t_double, t_decimal, t_bit
+                           , t_date, t_datetime, t_timestamp, t_time, t_year
+                           , t_char, t_varchar, t_blob, t_text, t_enum
+                           , t_set, t_json)
+VALUES ( true, 9223372036875807, 153.123, 123456669012.123456789012, b'1010001'
+       , '2000-01-01', '9999-12-31 23:59:59', '19731230153000', '23:59:59', 1970
+       , '测', '测试', 'blob', '测试text', 'enum1'
+       , 'a,b', '{
+    "key1": "value1",
+    "key2": "value2"
+  }');
+
+UPDATE t_alter
+SET t_bigint = 888,
+    t_json   = '{
+      "key0": "value0",
+      "key2": "value2"
+    }'
+WHERE id = 2;
diff --git a/tests/integration_tests/event_filter/data/test_truncate.sql b/tests/integration_tests/event_filter/data/test_truncate.sql
new file mode 100644
index 000000000..170d94c9c
--- /dev/null
+++ b/tests/integration_tests/event_filter/data/test_truncate.sql
@@ -0,0 +1,15 @@
+USE `event_filter`;
+
+/* test new physical table is replicated */
+TRUNCATE TABLE t_truncate;
+INSERT INTO t_truncate
+VALUES (1, 'guagua');
+
+INSERT INTO t_truncate
+VALUES (2, 'huahua');
+
+INSERT INTO t_truncate
+VALUES (3, 'xigua');
+
+INSERT INTO t_truncate
+VALUES (4, 'yuko');
diff --git a/tests/integration_tests/event_filter/run.sh b/tests/integration_tests/event_filter/run.sh
index b996adc4d..7f52b2076 100644
--- a/tests/integration_tests/event_filter/run.sh
+++ b/tests/integration_tests/event_filter/run.sh
@@ -36,8 +36,9 @@ function run() {
 	# make suer table t1 is deleted in upstream and exists in downstream
 	check_table_not_exists "event_filter.t1" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 	check_table_exists "event_filter.t1" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
-	check_table_exists "event_filter.t2" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
-	check_table_exists "event_filter.finish_mark" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	check_table_exists "event_filter.t_normal" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	check_table_exists "event_filter.t_truncate" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	check_table_exists "event_filter.t_alter" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 
 	# check those rows that are not filtered are synced to downstream
 	run_sql "select count(1) from event_filter.t1;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
@@ -51,7 +52,14 @@ function run() {
 	run_sql "select count(5) from event_filter.t1 where id=4;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 	check_contains "count(5): 1"
 
-	# check table t2 is replicated
+	run_sql "TRUNCATE TABLE event_filter.t_truncate;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	run_sql_file $CUR/data/test_truncate.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	run_sql "ALTER TABLE event_filter.t_alter MODIFY t_bigint BIGINT;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	run_sql_file $CUR/data/test_alter.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	run_sql "create table event_filter.finish_mark(id int primary key);" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	check_table_exists "event_filter.finish_mark" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+
+	# check table t_normal is replicated
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
 
 	cleanup_process $CDC_BINARY
diff --git a/tests/integration_tests/multi_topics/conf/changefeed.toml b/tests/integration_tests/multi_topics/conf/changefeed.toml
index 0f72c584c..2d5a3c83c 100644
--- a/tests/integration_tests/multi_topics/conf/changefeed.toml
+++ b/tests/integration_tests/multi_topics/conf/changefeed.toml
@@ -1,4 +1,5 @@
 [sink]
 dispatchers = [
+    { matcher = ['workload.*'], topic = "workload"},
     { matcher = ['test.*'], topic = "{schema}_{table}" },
 ]
diff --git a/tests/integration_tests/multi_topics/data/step1.sql b/tests/integration_tests/multi_topics/data/step1.sql
index fa18f2208..a33dbc81c 100644
--- a/tests/integration_tests/multi_topics/data/step1.sql
+++ b/tests/integration_tests/multi_topics/data/step1.sql
@@ -41,4 +41,50 @@ create table table3
 );
 
 insert into table3(c_tinyint, c_smallint, c_mediumint, c_int, c_bigint)
-values (1, 2, 3, 4, 5);
\ No newline at end of file
+values (1, 2, 3, 4, 5);
+
+create database workload;
+use workload;
+
+create table table1
+(
+    id          int auto_increment,
+    c_tinyint   tinyint null,
+    c_smallint  smallint null,
+    c_mediumint mediumint null,
+    c_int       int null,
+    c_bigint    bigint null,
+    constraint pk
+        primary key (id)
+);
+insert into table1(c_tinyint, c_smallint, c_mediumint, c_int, c_bigint)
+values (1, 2, 3, 4, 5);
+
+create table table2
+(
+    id          int auto_increment,
+    c_tinyint   tinyint null,
+    c_smallint  smallint null,
+    c_mediumint mediumint null,
+    c_int       int null,
+    c_bigint    bigint null,
+    constraint pk
+        primary key (id)
+);
+insert into table2(c_tinyint, c_smallint, c_mediumint, c_int, c_bigint)
+values (1, 2, 3, 4, 5);
+
+create table table3
+(
+    id          int auto_increment,
+    c_tinyint   tinyint null,
+    c_smallint  smallint null,
+    c_mediumint mediumint null,
+    c_int       int null,
+    c_bigint    bigint null,
+    constraint pk
+        primary key (id)
+);
+
+insert into table3(c_tinyint, c_smallint, c_mediumint, c_int, c_bigint)
+values (1, 2, 3, 4, 5);
diff --git a/tests/integration_tests/multi_topics/run.sh b/tests/integration_tests/multi_topics/run.sh
index fa38b2f10..99f8acf64 100644
--- a/tests/integration_tests/multi_topics/run.sh
+++ b/tests/integration_tests/multi_topics/run.sh
@@ -36,6 +36,8 @@ function run() {
 		run_kafka_consumer $WORK_DIR "kafka://127.0.0.1:9092/test_table${i}?protocol=canal-json&version=${KAFKA_VERSION}&enable-tidb-extension=true" "" ${i}
 	done
 
+	run_kafka_consumer $WORK_DIR "kafka://127.0.0.1:9092/workload?protocol=canal-json&enable-tidb-extension=true" ""
+
 	# sync_diff can't check non-exist table, so we check expected tables are created in downstream first
 	for i in $(seq 1 3); do
 		check_table_exists test.table${i} ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 300
diff --git a/tests/integration_tests/run_group.sh b/tests/integration_tests/run_group.sh
index 1e5c4392b..a93653d8e 100755
--- a/tests/integration_tests/run_group.sh
+++ b/tests/integration_tests/run_group.sh
@@ -40,7 +40,7 @@ groups=(
 	["G06"]='sink_retry changefeed_error ddl_sequence resourcecontrol'
 	["G07"]='kv_client_stream_reconnect cdc split_region'
 	["G08"]='server_config_compatibility processor_err_chan changefeed_reconstruct multi_capture'
-	["G09"]='gc_safepoint changefeed_pause_resume cli savepoint'
+	["G09"]='gc_safepoint changefeed_pause_resume cli savepoint synced_status'
 	["G10"]='default_value simple cdc_server_tips event_filter'
 	["G11"]='resolve_lock move_table autorandom generate_column'
 	["G12"]='many_pk_or_uk capture_session_done_during_task ddl_attributes'
diff --git a/tests/integration_tests/synced_status/conf/changefeed-redo.toml b/tests/integration_tests/synced_status/conf/changefeed-redo.toml
new file mode 100644
index 000000000..8f67ecb9d
--- /dev/null
+++ b/tests/integration_tests/synced_status/conf/changefeed-redo.toml
@@ -0,0 +1,7 @@
+[synced-status]
+synced-check-interval = 120 
+checkpoint-interval = 20
+
+[consistent]
+level="eventual"
+storage = "file:///tmp/tidb_cdc_test/synced_status/redo"
\ No newline at end of file
diff --git a/tests/integration_tests/synced_status/conf/changefeed.toml b/tests/integration_tests/synced_status/conf/changefeed.toml
new file mode 100644
index 000000000..0d8aa19fb
--- /dev/null
+++ b/tests/integration_tests/synced_status/conf/changefeed.toml
@@ -0,0 +1,3 @@
+[synced-status]
+synced-check-interval = 120 
+checkpoint-interval = 20
\ No newline at end of file
diff --git a/tests/integration_tests/synced_status/run.sh b/tests/integration_tests/synced_status/run.sh
new file mode 100644
index 000000000..ffb7d6892
--- /dev/null
+++ b/tests/integration_tests/synced_status/run.sh
@@ -0,0 +1,308 @@
+#!/bin/bash
+
+# [DISCRIPTION]:
+#   This test is related to
+#   It will test the sync status request of cdc server in the following scenarios:
+#   1. The sync status request of cdc server when the upstream cluster is available
+#      1.1 pdNow - lastSyncedTs > threshold, pdNow - checkpointTs < threshold
+#      1.2 pdNow - lastSyncedTs < threshold
+#      1.3 pdNow - lastSyncedTs > threshold, pdNow - checkpointTs < threshold, resolvedTs - checkpointTs > threshold
+#   2. The sync status request of cdc server when the upstream pd is unavailable
+#      2.1 resolvedTs - checkpointTs < threshold
+#   3. The sync status request of cdc server when the upstream tikv is unavailable
+#      3.1 pdNow - lastSyncedTs > threshold, pdNow - checkpointTs > threshold, resolvedTs - checkpointTs < threshold
+#      3.2 pdNow - lastSyncedTs < threshold
+#   4. The sync status request of cdc server when the downstream tidb is available
+#      4.1 pdNow - lastSyncedTs > threshold, pdNow - checkpointTs < threshold
+#      4.2 pdNow - lastSyncedTs < threshold
+# [STEP]:
+#   1. Create changefeed with synced-time-config = xx
+#   2. insert data to upstream cluster, and do the related actions for each scenarios
+#   3. do the query of synced status of cdc server
+#   4. check the info and status of query
+
+set -xeu
+
+CUR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+source $CUR/../_utils/test_prepare
+WORK_DIR=$OUT_DIR/$TEST_NAME
+CDC_BINARY=cdc.test
+SINK_TYPE=$1
+
+CDC_COUNT=3
+DB_COUNT=4
+
+function kill_pd() {
+	info=$(ps aux | grep pd-server | grep $WORK_DIR) || true
+	$(ps aux | grep pd-server | grep $WORK_DIR | awk '{print $2}' | xargs kill -9 &>/dev/null) || true
+}
+
+function kill_tikv() {
+	info=$(ps aux | grep tikv-server | grep $WORK_DIR) || true
+	$(ps aux | grep tikv-server | grep $WORK_DIR | awk '{print $2}' | xargs kill -9 &>/dev/null) || true
+}
+
+function kill_tidb() {
+	info=$(ps aux | grep tidb-server | grep $WORK_DIR) || true
+	$(ps aux | grep tidb-server | grep $WORK_DIR | awk '{print $2}' | xargs kill -9 &>/dev/null) || true
+}
+
+function run_normal_case_and_unavailable_pd() {
+	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
+
+	start_tidb_cluster --workdir $WORK_DIR
+
+	cd $WORK_DIR
+
+	start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1})
+	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
+
+	config_path=$1
+
+	SINK_URI="mysql://root@127.0.0.1:3306/?max-txn-row=1"
+	run_cdc_cli changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" --changefeed-id="test-1" --config="$CUR/$config_path"
+
+	# case 1: test in available cluster
+	synced_status=$(curl -X GET http://127.0.0.1:8300/api/v2/changefeeds/test-1/synced)
+
+	status=$(echo $synced_status | jq '.synced')
+	sink_checkpoint_ts=$(echo $synced_status | jq -r '.sink_checkpoint_ts')
+	puller_resolved_ts=$(echo $synced_status | jq -r '.puller_resolved_ts')
+	last_synced_ts=$(echo $synced_status | jq -r '.last_synced_ts')
+	if [ $status != true ]; then
+		echo "synced status isn't correct"
+		exit 1
+	fi
+	# the timestamp for puller_resolved_ts is 0 when do data insert
+	if [ "$puller_resolved_ts" != "1970-01-01 08:00:00.000" ]; then
+		echo "puller_resolved_ts is not 1970-01-01 08:00:00.000"
+		exit 1
+	fi
+	# the timestamp for last_synced_ts is 0 when do data insert
+	if [ "$last_synced_ts" != "1970-01-01 08:00:00.000" ]; then
+		echo "last_synced_ts is not 1970-01-01 08:00:00.000"
+		exit 1
+	fi
+
+	# compare sink_checkpoint_ts with current time
+	current=$(date +"%Y-%m-%d %H:%M:%S")
+	echo "sink_checkpoint_ts is "$sink_checkpoint_ts
+	checkpoint_timestamp=$(date -d "$sink_checkpoint_ts" +%s)
+	current_timestamp=$(date -d "$current" +%s)
+	if [ $(($current_timestamp - $checkpoint_timestamp)) -gt 300 ]; then # give a soft check
+		echo "sink_checkpoint_ts is not correct"
+		exit 1
+	fi
+
+	run_sql "USE TEST;Create table t1(a int primary key, b int);insert into t1 values(1,2);insert into t1 values(2,3);"
+	check_table_exists "test.t1" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+
+	sleep 5 # wait data insert
+	synced_status=$(curl -X GET http://127.0.0.1:8300/api/v2/changefeeds/test-1/synced)
+	status=$(echo $synced_status | jq '.synced')
+	if [ $status != false ]; then
+		echo "synced status isn't correct"
+		exit 1
+	fi
+	info=$(echo $synced_status | jq -r '.info')
+	if [ "$info" != "The data syncing is not finished, please wait" ]; then
+		echo "synced status info is not correct"
+		exit 1
+	fi
+
+	sleep 130 # wait enough time for pass synced-check-interval
+	synced_status=$(curl -X GET http://127.0.0.1:8300/api/v2/changefeeds/test-1/synced)
+	status=$(echo $synced_status | jq '.synced')
+	if [ $status != true ]; then
+		echo "synced status isn't correct"
+		exit 1
+	fi
+
+	#==========
+	# case 2: test with unavailable pd, query will not get the available response
+	kill_pd
+
+	sleep 20
+
+	synced_status=$(curl -X GET http://127.0.0.1:8300/api/v2/changefeeds/test-1/synced)
+	error_code=$(echo $synced_status | jq -r '.error_code')
+	cleanup_process $CDC_BINARY
+	stop_tidb_cluster
+}
+
+function run_case_with_unavailable_tikv() {
+	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
+
+	start_tidb_cluster --workdir $WORK_DIR
+
+	cd $WORK_DIR
+
+	start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1})
+	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
+
+	config_path=$1
+
+	SINK_URI="mysql://root@127.0.0.1:3306/?max-txn-row=1"
+	run_cdc_cli changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" --changefeed-id="test-1" --config="$CUR/$config_path"
+
+	# case 3: test in unavailable tikv cluster
+	run_sql "USE TEST;Create table t1(a int primary key, b int);insert into t1 values(1,2);insert into t1 values(2,3);"
+	check_table_exists "test.t1" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+
+	sleep 5 # make data inserted into downstream
+	kill_tikv
+
+	# test the case when pdNow - lastSyncedTs < threshold
+	synced_status=$(curl -X GET http://127.0.0.1:8300/api/v2/changefeeds/test-1/synced)
+	status=$(echo $synced_status | jq '.synced')
+	if [ $status != false ]; then
+		echo "synced status isn't correct"
+		exit 1
+	fi
+	info=$(echo $synced_status | jq -r '.info')
+	target_message="The data syncing is not finished, please wait"
+
+	if [ "$info" != "$target_message" ]; then
+		echo "synced status info is not correct"
+		exit 1
+	fi
+
+	sleep 130 # wait enough time for pass synced-check-interval
+	# test the case when pdNow - lastSyncedTs > threshold
+	synced_status=$(curl -X GET http://127.0.0.1:8300/api/v2/changefeeds/test-1/synced)
+	status=$(echo $synced_status | jq '.synced')
+	if [ $status != false ]; then
+		echo "synced status isn't correct"
+		exit 1
+	fi
+	info=$(echo $synced_status | jq -r '.info')
+	target_message="Please check whether PD is online and TiKV Regions are all available. \
+If PD is offline or some TiKV regions are not available, it means that the data syncing process is complete. \
+To check whether TiKV regions are all available, \
+you can view 'TiKV-Details' > 'Resolved-Ts' > 'Max Leader Resolved TS gap' on Grafana. \
+If the gap is large, such as a few minutes, it means that some regions in TiKV are unavailable. \
+Otherwise, if the gap is small and PD is online, it means the data syncing is incomplete, so please wait"
+
+	if [ "$info" != "$target_message" ]; then
+		echo "synced status info is not correct"
+		exit 1
+	fi
+
+	cleanup_process $CDC_BINARY
+	stop_tidb_cluster
+}
+
+function run_case_with_unavailable_tidb() {
+	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
+
+	start_tidb_cluster --workdir $WORK_DIR
+
+	cd $WORK_DIR
+
+	start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1})
+	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
+
+	config_path=$1
+
+	SINK_URI="mysql://root@127.0.0.1:3306/?max-txn-row=1"
+	run_cdc_cli changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" --changefeed-id="test-1" --config="$CUR/$config_path"
+
+	# case 3: test in unavailable tikv cluster
+	run_sql "USE TEST;Create table t1(a int primary key, b int);insert into t1 values(1,2);insert into t1 values(2,3);"
+	check_table_exists "test.t1" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+
+	sleep 5 # make data inserted into downstream
+	kill_tidb
+
+	# test the case when pdNow - lastSyncedTs < threshold
+	synced_status=$(curl -X GET http://127.0.0.1:8300/api/v2/changefeeds/test-1/synced)
+	status=$(echo $synced_status | jq '.synced')
+	if [ $status != false ]; then
+		echo "synced status isn't correct"
+		exit 1
+	fi
+	info=$(echo $synced_status | jq -r '.info')
+	target_message="The data syncing is not finished, please wait"
+
+	if [ "$info" != "$target_message" ]; then
+		echo "synced status info is not correct"
+		exit 1
+	fi
+
+	sleep 130 # wait enough time for pass synced-check-interval
+	# test the case when pdNow - lastSyncedTs > threshold
+	synced_status=$(curl -X GET http://127.0.0.1:8300/api/v2/changefeeds/test-1/synced)
+	status=$(echo $synced_status | jq '.synced')
+	if [ $status != true ]; then
+		echo "synced status isn't correct"
+		exit 1
+	fi
+	info=$(echo $synced_status | jq -r '.info')
+	target_message="Data syncing is finished"
+
+	if [ "$info" != "$target_message" ]; then
+		echo "synced status info is not correct"
+		exit 1
+	fi
+
+	cleanup_process $CDC_BINARY
+	stop_tidb_cluster
+}
+
+function run_case_with_failpoint() {
+	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
+
+	start_tidb_cluster --workdir $WORK_DIR
+
+	cd $WORK_DIR
+
+	# make failpoint to block checkpoint-ts
+	export GO_FAILPOINTS='github.com/pingcap/tiflow/cdc/owner/ChangefeedOwnerNotUpdateCheckpoint=return(true)'
+
+	start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1})
+	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
+
+	config_path=$1
+
+	SINK_URI="mysql://root@127.0.0.1:3306/?max-txn-row=1"
+	run_cdc_cli changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" --changefeed-id="test-1" --config="$CUR/$config_path"
+
+	sleep 20 # wait enough time for pass checkpoint-check-interval
+	synced_status=$(curl -X GET http://127.0.0.1:8300/api/v2/changefeeds/test-1/synced)
+	status=$(echo $synced_status | jq '.synced')
+	if [ $status != false ]; then
+		echo "synced status isn't correct"
+		exit 1
+	fi
+	info=$(echo $synced_status | jq -r '.info')
+	target_message="Please check whether PD is online and TiKV Regions are all available. \
+If PD is offline or some TiKV regions are not available, it means that the data syncing process is complete. \
+To check whether TiKV regions are all available, \
+you can view 'TiKV-Details' > 'Resolved-Ts' > 'Max Leader Resolved TS gap' on Grafana. \
+If the gap is large, such as a few minutes, it means that some regions in TiKV are unavailable. \
+Otherwise, if the gap is small and PD is online, it means the data syncing is incomplete, so please wait"
+	if [ "$info" != "$target_message" ]; then
+		echo "synced status info is not correct"
+		exit 1
+	fi
+
+	export GO_FAILPOINTS=''
+
+	cleanup_process $CDC_BINARY
+	stop_tidb_cluster
+}
+
+trap stop_tidb_cluster EXIT
+run_normal_case_and_unavailable_pd "conf/changefeed.toml"
+run_case_with_unavailable_tikv "conf/changefeed.toml"
+run_case_with_unavailable_tidb "conf/changefeed.toml"
+run_case_with_failpoint "conf/changefeed.toml"
+
+# enable redo
+run_normal_case_and_unavailable_pd "conf/changefeed-redo.toml"
+run_case_with_unavailable_tikv "conf/changefeed-redo.toml"
+run_case_with_unavailable_tidb "conf/changefeed-redo.toml"
+run_case_with_failpoint "conf/changefeed-redo.toml"
+
+check_logs $WORK_DIR
+echo "[$(date)] <<<<<< run test case $TEST_NAME success! >>>>>>"