Repository: pingcap/tiflow
Pull request #10506 (2024-01-18 07:32:08)

kv-client(cdc): introduce a more simple frontier
### What problem does this PR solve?

Issue Number: ref #10574 

### What is changed and how it works?

* Add a new server configuration `kv-client.advance-interval-in-ms`:
  a. by default it's `300`, which means calculate ResolvedTs every `300ms`;
  b. if it's set to 0, `cdc/puller.Frontier` will still be used to advance ResolvedTs.

This PR can reduce CPU and memory usage and make TiCDC more happy with the large single table case.

We can remove `cdc/puller.Frontier` and the new configuration after the change is stable.


### Check List <!--REMOVE the items that are not applicable-->

#### Tests <!-- At least one of them must be included. -->

 - Unit test
 - Manual test with a heavy sysbench insert load
    With the patch:
    ![图片](https://github.com/pingcap/tiflow/assets/8407317/fb57f173-314c-4745-84ec-44743ba1ace3)

    Without the patch:
    ![图片](https://github.com/pingcap/tiflow/assets/8407317/364cdc9d-74eb-44f4-84a7-7cb3581a7b42)

    The result shows the patch could increase lag about `100ms`. I think it's acceptable.


#### Questions <!-- Authors should answer these questions and reviewers should consider these questions. -->

##### Will it cause performance regression or break compatibility?

##### Do you need to update user documentation, design documentation or monitoring documentation?

### Release note <!-- bugfixes or new features need a release note -->

```release-note
None
```


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

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

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["hicqu"]} -->
ti-chi-bot[bot]: 
[FORMAT CHECKER NOTIFICATION]

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

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


<!--Checker Notification Identifier-->

ti-chi-bot[bot]: 
[FORMAT CHECKER NOTIFICATION]

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

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


<!--Checker Notification Identifier-->

ti-chi-bot[bot]: @hicqu: 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-cdc-integration-storage-test | 9f166de82f8b76c84505e5ab59cf9aab229f2826 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_storage_test/329/display/redirect) | true | `/test cdc-integration-storage-test`

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

<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 -->
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/pingcap/tiflow/pull/10506?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) Report
> Merging [#10506](https://app.codecov.io/gh/pingcap/tiflow/pull/10506?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) (9f166de) into [master](https://app.codecov.io/gh/pingcap/tiflow/commit/50d96a6508001f98703a0cecceb4edc56080e39b?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) (50d96a6) will **increase** coverage by `0.0326%`.
> Report is 3 commits behind head on master.
> The diff coverage is `90.6666%`.

<details><summary>Additional details and impacted files</summary>


| [Components](https://app.codecov.io/gh/pingcap/tiflow/pull/10506/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | Coverage Δ | |
|---|---|---|
| [cdc](https://app.codecov.io/gh/pingcap/tiflow/pull/10506/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `61.6768% <90.6666%> (+0.0501%)` | :arrow_up: |
| [dm](https://app.codecov.io/gh/pingcap/tiflow/pull/10506/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `51.2174% <ø> (+0.0101%)` | :arrow_up: |
| [engine](https://app.codecov.io/gh/pingcap/tiflow/pull/10506/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `63.3999% <ø> (+0.0070%)` | :arrow_up: |

| [Flag](https://app.codecov.io/gh/pingcap/tiflow/pull/10506/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | Coverage Δ | |
|---|---|---|
| [unit](https://app.codecov.io/gh/pingcap/tiflow/pull/10506/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `57.4232% <90.6666%> (+0.0326%)` | :arrow_up: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap#carryforward-flags-in-the-pull-request-comment) to find out more.

```diff
@@               Coverage Diff                @@
##             master     #10506        +/-   ##
================================================
+ Coverage   57.3905%   57.4232%   +0.0326%     
================================================
  Files           849        849                
  Lines        125864     125970       +106     
================================================
+ Hits          72234      72336       +102     
  Misses        48219      48219                
- Partials       5411       5415         +4     
```

</details>
ti-chi-bot[bot]: 
[FORMAT CHECKER NOTIFICATION]

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

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


<!--Checker Notification Identifier-->

ti-chi-bot[bot]: @hicqu: 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-cdc-integration-storage-test | a3f1fe4634a92d0675050129454cc74ecaf3cfec | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_storage_test/372/display/redirect) | true | `/test cdc-integration-storage-test`

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

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cdc-integration-storage-test | a3f1fe4634a92d0675050129454cc74ecaf3cfec | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_storage_test/372/display/redirect) | true | `/test cdc-integration-storage-test`
pull-cdc-integration-kafka-test | a3f1fe4634a92d0675050129454cc74ecaf3cfec | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/346/display/redirect) | true | `/test cdc-integration-kafka-test`

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

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cdc-integration-storage-test | a3f1fe4634a92d0675050129454cc74ecaf3cfec | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_storage_test/372/display/redirect) | true | `/test cdc-integration-storage-test`
pull-cdc-integration-kafka-test | a3f1fe4634a92d0675050129454cc74ecaf3cfec | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/346/display/redirect) | true | `/test cdc-integration-kafka-test`
pull-cdc-integration-mysql-test | a3f1fe4634a92d0675050129454cc74ecaf3cfec | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_test/353/display/redirect) | true | `/test cdc-integration-mysql-test`

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

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cdc-integration-kafka-test | a3f1fe4634a92d0675050129454cc74ecaf3cfec | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/346/display/redirect) | true | `/test cdc-integration-kafka-test`
pull-cdc-integration-storage-test | a3f1fe4634a92d0675050129454cc74ecaf3cfec | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_storage_test/409/display/redirect) | true | `/test cdc-integration-storage-test`

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

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cdc-integration-storage-test | a3f1fe4634a92d0675050129454cc74ecaf3cfec | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_storage_test/409/display/redirect) | true | `/test cdc-integration-storage-test`
pull-cdc-integration-pulsar-test | a3f1fe4634a92d0675050129454cc74ecaf3cfec | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_pulsar_test/108/display/redirect) | true | `/test cdc-integration-pulsar-test`
pull-cdc-integration-kafka-test | a3f1fe4634a92d0675050129454cc74ecaf3cfec | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/386/display/redirect) | true | `/test cdc-integration-kafka-test`

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

<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]: 
[FORMAT CHECKER NOTIFICATION]

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

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


<!--Checker Notification Identifier-->

ti-chi-bot[bot]: @hicqu: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cdc-integration-storage-test | a3f1fe4634a92d0675050129454cc74ecaf3cfec | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_storage_test/409/display/redirect) | true | `/test cdc-integration-storage-test`
pull-cdc-integration-pulsar-test | a3f1fe4634a92d0675050129454cc74ecaf3cfec | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_pulsar_test/108/display/redirect) | true | `/test cdc-integration-pulsar-test`
pull-cdc-integration-kafka-test | a3f1fe4634a92d0675050129454cc74ecaf3cfec | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/386/display/redirect) | true | `/test cdc-integration-kafka-test`
jenkins-ticdc/verify | 6120a12fe08c68353bed9b0bb5df4c7ecfe93c95 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/408/display/redirect) | true | `/test verify`

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

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cdc-integration-storage-test | a3f1fe4634a92d0675050129454cc74ecaf3cfec | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_storage_test/409/display/redirect) | true | `/test cdc-integration-storage-test`
pull-cdc-integration-pulsar-test | a3f1fe4634a92d0675050129454cc74ecaf3cfec | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_pulsar_test/108/display/redirect) | true | `/test cdc-integration-pulsar-test`
jenkins-ticdc/verify | 6120a12fe08c68353bed9b0bb5df4c7ecfe93c95 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/408/display/redirect) | true | `/test verify`
pull-cdc-integration-kafka-test | 6120a12fe08c68353bed9b0bb5df4c7ecfe93c95 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/406/display/redirect) | true | `/test cdc-integration-kafka-test`

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

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cdc-integration-storage-test | a3f1fe4634a92d0675050129454cc74ecaf3cfec | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_storage_test/409/display/redirect) | true | `/test cdc-integration-storage-test`
jenkins-ticdc/verify | 6120a12fe08c68353bed9b0bb5df4c7ecfe93c95 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/408/display/redirect) | true | `/test verify`
pull-cdc-integration-kafka-test | 6120a12fe08c68353bed9b0bb5df4c7ecfe93c95 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/406/display/redirect) | true | `/test cdc-integration-kafka-test`
pull-cdc-integration-pulsar-test | 6120a12fe08c68353bed9b0bb5df4c7ecfe93c95 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_pulsar_test/131/display/redirect) | true | `/test cdc-integration-pulsar-test`

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

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
jenkins-ticdc/verify | 6120a12fe08c68353bed9b0bb5df4c7ecfe93c95 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/408/display/redirect) | true | `/test verify`
pull-cdc-integration-kafka-test | 6120a12fe08c68353bed9b0bb5df4c7ecfe93c95 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/406/display/redirect) | true | `/test cdc-integration-kafka-test`
pull-cdc-integration-pulsar-test | 6120a12fe08c68353bed9b0bb5df4c7ecfe93c95 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_pulsar_test/131/display/redirect) | true | `/test cdc-integration-pulsar-test`
pull-cdc-integration-storage-test | 6120a12fe08c68353bed9b0bb5df4c7ecfe93c95 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_storage_test/431/display/redirect) | true | `/test cdc-integration-storage-test`

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

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
jenkins-ticdc/verify | 6120a12fe08c68353bed9b0bb5df4c7ecfe93c95 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/408/display/redirect) | true | `/test verify`
pull-cdc-integration-kafka-test | 6120a12fe08c68353bed9b0bb5df4c7ecfe93c95 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/406/display/redirect) | true | `/test cdc-integration-kafka-test`
pull-cdc-integration-pulsar-test | 6120a12fe08c68353bed9b0bb5df4c7ecfe93c95 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_pulsar_test/131/display/redirect) | true | `/test cdc-integration-pulsar-test`
pull-cdc-integration-storage-test | 6120a12fe08c68353bed9b0bb5df4c7ecfe93c95 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_storage_test/431/display/redirect) | true | `/test cdc-integration-storage-test`
pull-cdc-integration-mysql-test | 6120a12fe08c68353bed9b0bb5df4c7ecfe93c95 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_test/414/display/redirect) | true | `/test cdc-integration-mysql-test`

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

<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]: 
[FORMAT CHECKER NOTIFICATION]

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

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


<!--Checker Notification Identifier-->

ti-chi-bot[bot]: 
[FORMAT CHECKER NOTIFICATION]

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

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


<!--Checker Notification Identifier-->

ti-chi-bot[bot]: @hicqu: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cdc-integration-kafka-test | 6120a12fe08c68353bed9b0bb5df4c7ecfe93c95 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/406/display/redirect) | true | `/test cdc-integration-kafka-test`
pull-cdc-integration-pulsar-test | 6120a12fe08c68353bed9b0bb5df4c7ecfe93c95 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_pulsar_test/131/display/redirect) | true | `/test cdc-integration-pulsar-test`
pull-cdc-integration-storage-test | 6120a12fe08c68353bed9b0bb5df4c7ecfe93c95 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_storage_test/431/display/redirect) | true | `/test cdc-integration-storage-test`
pull-cdc-integration-mysql-test | 6120a12fe08c68353bed9b0bb5df4c7ecfe93c95 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_test/414/display/redirect) | true | `/test cdc-integration-mysql-test`
jenkins-ticdc/verify | 8b178fb137b4da23012cae7db3d9abf7bb309a59 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/415/display/redirect) | true | `/test verify`

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

<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]: @hicqu: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
jenkins-ticdc/verify | 9b959e2c7a05854c66e975ae1fcfc3c080b3ab10 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/420/display/redirect) | true | `/test verify`

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

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
jenkins-ticdc/verify | 9b959e2c7a05854c66e975ae1fcfc3c080b3ab10 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/420/display/redirect) | true | `/test verify`
pull-cdc-integration-storage-test | 9b959e2c7a05854c66e975ae1fcfc3c080b3ab10 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_storage_test/445/display/redirect) | true | `/test cdc-integration-storage-test`

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

<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]: 
[FORMAT CHECKER NOTIFICATION]

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

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


<!--Checker Notification Identifier-->

ti-chi-bot[bot]: 
[FORMAT CHECKER NOTIFICATION]

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

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


<!--Checker Notification Identifier-->

ti-chi-bot[bot]: @hicqu: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cdc-integration-storage-test | 9b959e2c7a05854c66e975ae1fcfc3c080b3ab10 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_storage_test/445/display/redirect) | true | `/test cdc-integration-storage-test`
jenkins-ticdc/verify | cff546f2467078518ce5151278c1ea4f78f374d0 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/626/display/redirect) | true | `/test verify`

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

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cdc-integration-storage-test | 9b959e2c7a05854c66e975ae1fcfc3c080b3ab10 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_storage_test/445/display/redirect) | true | `/test cdc-integration-storage-test`
jenkins-ticdc/verify | cff546f2467078518ce5151278c1ea4f78f374d0 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/626/display/redirect) | true | `/test verify`
pull-cdc-integration-pulsar-test | cff546f2467078518ce5151278c1ea4f78f374d0 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_pulsar_test/343/display/redirect) | true | `/test cdc-integration-pulsar-test`

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

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cdc-integration-storage-test | 9b959e2c7a05854c66e975ae1fcfc3c080b3ab10 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_storage_test/445/display/redirect) | true | `/test cdc-integration-storage-test`
jenkins-ticdc/verify | cff546f2467078518ce5151278c1ea4f78f374d0 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/626/display/redirect) | true | `/test verify`
pull-cdc-integration-pulsar-test | cff546f2467078518ce5151278c1ea4f78f374d0 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_pulsar_test/343/display/redirect) | true | `/test cdc-integration-pulsar-test`
pull-dm-integration-test | cff546f2467078518ce5151278c1ea4f78f374d0 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_integration_test/623/display/redirect) | true | `/test dm-integration-test`

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

<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]: 
[FORMAT CHECKER NOTIFICATION]

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

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


<!--Checker Notification Identifier-->

ti-chi-bot[bot]: @hicqu: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cdc-integration-pulsar-test | cff546f2467078518ce5151278c1ea4f78f374d0 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_pulsar_test/343/display/redirect) | true | `/test cdc-integration-pulsar-test`
pull-dm-integration-test | cff546f2467078518ce5151278c1ea4f78f374d0 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_integration_test/623/display/redirect) | true | `/test dm-integration-test`
jenkins-ticdc/verify | 8b7af23a41b72b6b732dd2f357aa9a27460d66f3 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/643/display/redirect) | true | `/test verify`

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

<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]: 
[FORMAT CHECKER NOTIFICATION]

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

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


<!--Checker Notification Identifier-->

ti-chi-bot[bot]: @hicqu: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cdc-integration-pulsar-test | cff546f2467078518ce5151278c1ea4f78f374d0 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_pulsar_test/343/display/redirect) | true | `/test cdc-integration-pulsar-test`
pull-dm-integration-test | cff546f2467078518ce5151278c1ea4f78f374d0 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_integration_test/623/display/redirect) | true | `/test dm-integration-test`
jenkins-ticdc/verify | 8bce4de6d27cc8f8f6c2027aed46e63cf7231f07 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/646/display/redirect) | true | `/test verify`

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

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cdc-integration-pulsar-test | cff546f2467078518ce5151278c1ea4f78f374d0 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_pulsar_test/343/display/redirect) | true | `/test cdc-integration-pulsar-test`
jenkins-ticdc/verify | 8bce4de6d27cc8f8f6c2027aed46e63cf7231f07 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/646/display/redirect) | true | `/test verify`
pull-dm-integration-test | 8bce4de6d27cc8f8f6c2027aed46e63cf7231f07 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_integration_test/643/display/redirect) | true | `/test dm-integration-test`

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

<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]: @hicqu: 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 | 8bce4de6d27cc8f8f6c2027aed46e63cf7231f07 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_integration_test/643/display/redirect) | true | `/test dm-integration-test`
pull-cdc-integration-storage-test | d90cb4d64add488136ac882cedd9561c99bf67f3 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_storage_test/684/display/redirect) | true | `/test cdc-integration-storage-test`

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

<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 -->
hicqu: /retest
ti-chi-bot[bot]: @hicqu: 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-compatibility-test | 0a04025c65b2c48f74d47d6c5bbbc76dca7dae9f | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_compatibility_test/633/display/redirect) | true | `/test dm-compatibility-test`

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

<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]: @hicqu: 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-compatibility-test | 0a04025c65b2c48f74d47d6c5bbbc76dca7dae9f | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_compatibility_test/633/display/redirect) | true | `/test dm-compatibility-test`
jenkins-ticdc/verify | 0a04025c65b2c48f74d47d6c5bbbc76dca7dae9f | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/669/display/redirect) | true | `/test verify`

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

<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]: @hicqu: 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-compatibility-test | 0a04025c65b2c48f74d47d6c5bbbc76dca7dae9f | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_compatibility_test/633/display/redirect) | true | `/test dm-compatibility-test`
jenkins-ticdc/verify | 0a04025c65b2c48f74d47d6c5bbbc76dca7dae9f | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/669/display/redirect) | true | `/test verify`
pull-cdc-integration-storage-test | 0a04025c65b2c48f74d47d6c5bbbc76dca7dae9f | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_storage_test/701/display/redirect) | true | `/test cdc-integration-storage-test`

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

<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]: @hicqu: 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-compatibility-test | 0a04025c65b2c48f74d47d6c5bbbc76dca7dae9f | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_compatibility_test/633/display/redirect) | true | `/test dm-compatibility-test`
jenkins-ticdc/verify | 0a04025c65b2c48f74d47d6c5bbbc76dca7dae9f | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/669/display/redirect) | true | `/test verify`
pull-cdc-integration-storage-test | 0a04025c65b2c48f74d47d6c5bbbc76dca7dae9f | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_storage_test/701/display/redirect) | true | `/test cdc-integration-storage-test`
pull-cdc-integration-kafka-test | 0a04025c65b2c48f74d47d6c5bbbc76dca7dae9f | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/661/display/redirect) | true | `/test cdc-integration-kafka-test`

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

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
jenkins-ticdc/verify | 0a04025c65b2c48f74d47d6c5bbbc76dca7dae9f | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/669/display/redirect) | true | `/test verify`
pull-cdc-integration-storage-test | 0a04025c65b2c48f74d47d6c5bbbc76dca7dae9f | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_storage_test/701/display/redirect) | true | `/test cdc-integration-storage-test`
pull-cdc-integration-kafka-test | 0a04025c65b2c48f74d47d6c5bbbc76dca7dae9f | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/661/display/redirect) | true | `/test cdc-integration-kafka-test`
pull-dm-compatibility-test | ad1c66d0b3de4f7f51bf3c9a2d2f68212136d3e8 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_compatibility_test/634/display/redirect) | true | `/test dm-compatibility-test`

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

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cdc-integration-storage-test | 0a04025c65b2c48f74d47d6c5bbbc76dca7dae9f | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_storage_test/701/display/redirect) | true | `/test cdc-integration-storage-test`
pull-dm-compatibility-test | ad1c66d0b3de4f7f51bf3c9a2d2f68212136d3e8 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_compatibility_test/634/display/redirect) | true | `/test dm-compatibility-test`
pull-cdc-integration-kafka-test | ad1c66d0b3de4f7f51bf3c9a2d2f68212136d3e8 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/662/display/redirect) | true | `/test cdc-integration-kafka-test`

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

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cdc-integration-storage-test | 0a04025c65b2c48f74d47d6c5bbbc76dca7dae9f | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_storage_test/701/display/redirect) | true | `/test cdc-integration-storage-test`
pull-cdc-integration-kafka-test | ad1c66d0b3de4f7f51bf3c9a2d2f68212136d3e8 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/662/display/redirect) | true | `/test cdc-integration-kafka-test`
pull-dm-compatibility-test | 65a351a63d7fe525c2631763051d2d462959e20e | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_compatibility_test/636/display/redirect) | true | `/test dm-compatibility-test`

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

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cdc-integration-storage-test | 0a04025c65b2c48f74d47d6c5bbbc76dca7dae9f | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_storage_test/701/display/redirect) | true | `/test cdc-integration-storage-test`
pull-dm-compatibility-test | 65a351a63d7fe525c2631763051d2d462959e20e | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_compatibility_test/636/display/redirect) | true | `/test dm-compatibility-test`
pull-cdc-integration-kafka-test | 65a351a63d7fe525c2631763051d2d462959e20e | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/664/display/redirect) | true | `/test cdc-integration-kafka-test`

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

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cdc-integration-storage-test | 0a04025c65b2c48f74d47d6c5bbbc76dca7dae9f | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_storage_test/701/display/redirect) | true | `/test cdc-integration-storage-test`
pull-dm-compatibility-test | 65a351a63d7fe525c2631763051d2d462959e20e | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_compatibility_test/636/display/redirect) | true | `/test dm-compatibility-test`
pull-cdc-integration-kafka-test | 65a351a63d7fe525c2631763051d2d462959e20e | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/664/display/redirect) | true | `/test cdc-integration-kafka-test`
pull-dm-integration-test | 65a351a63d7fe525c2631763051d2d462959e20e | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_integration_test/670/display/redirect) | true | `/test dm-integration-test`

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

<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]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/pingcap/tiflow/pull/10506#pullrequestreview-1858634967" title="Approved">asddongmen</a>*

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

The pull request process is described [here](https://book.prow.tidb.net/#/workflows/pr)

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

- ~~[OWNERS](https://github.com/pingcap/tiflow/blob/master/OWNERS)~~ [asddongmen]

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":[]} -->
ti-chi-bot[bot]: [LGTM Timeline notifier]
---
Timeline:

- `2024-02-02 08:17:02.106584935 +0000 UTC m=+1726263.670882624`: :ballot_box_with_check: agreed by [asddongmen](https://github.com/asddongmen).
ti-chi-bot[bot]: @hicqu: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cdc-integration-kafka-test | 65a351a63d7fe525c2631763051d2d462959e20e | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/664/display/redirect) | true | `/test cdc-integration-kafka-test`
pull-dm-integration-test | 65a351a63d7fe525c2631763051d2d462959e20e | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_integration_test/670/display/redirect) | true | `/test dm-integration-test`
pull-dm-compatibility-test | 65a351a63d7fe525c2631763051d2d462959e20e | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_compatibility_test/640/display/redirect) | true | `/test dm-compatibility-test`

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

<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]: @hicqu: 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-compatibility-test | 65a351a63d7fe525c2631763051d2d462959e20e | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_compatibility_test/640/display/redirect) | true | `/test dm-compatibility-test`
pull-cdc-integration-kafka-test | 65a351a63d7fe525c2631763051d2d462959e20e | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/668/display/redirect) | true | `/test cdc-integration-kafka-test`
pull-dm-integration-test | 65a351a63d7fe525c2631763051d2d462959e20e | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_integration_test/674/display/redirect) | true | `/test dm-integration-test`

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

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cdc-integration-kafka-test | 65a351a63d7fe525c2631763051d2d462959e20e | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/668/display/redirect) | true | `/test cdc-integration-kafka-test`
pull-dm-integration-test | 65a351a63d7fe525c2631763051d2d462959e20e | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_integration_test/674/display/redirect) | true | `/test dm-integration-test`
pull-dm-compatibility-test | 65a351a63d7fe525c2631763051d2d462959e20e | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_compatibility_test/641/display/redirect) | true | `/test dm-compatibility-test`

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

<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]: @hicqu: 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 | 65a351a63d7fe525c2631763051d2d462959e20e | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_integration_test/674/display/redirect) | true | `/test dm-integration-test`
pull-dm-compatibility-test | 65a351a63d7fe525c2631763051d2d462959e20e | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_compatibility_test/641/display/redirect) | true | `/test dm-compatibility-test`
pull-cdc-integration-kafka-test | 65a351a63d7fe525c2631763051d2d462959e20e | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/669/display/redirect) | true | `/test cdc-integration-kafka-test`

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

<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 -->
CharlesCheung96: Please add basic integration test for the new frontier. 
ti-chi-bot[bot]: @hicqu: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cdc-integration-kafka-test | 65a351a63d7fe525c2631763051d2d462959e20e | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/669/display/redirect) | true | `/test cdc-integration-kafka-test`
pull-dm-compatibility-test | 244af95950cf798b6fb39e7676c3d7fc62ac5a72 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_compatibility_test/647/display/redirect) | true | `/test dm-compatibility-test`

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

<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 -->
hicqu: /retest
ti-chi-bot[bot]: @hicqu: 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-compatibility-test | 244af95950cf798b6fb39e7676c3d7fc62ac5a72 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_compatibility_test/648/display/redirect) | true | `/test dm-compatibility-test`
pull-cdc-integration-kafka-test | 244af95950cf798b6fb39e7676c3d7fc62ac5a72 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/676/display/redirect) | true | `/test cdc-integration-kafka-test`

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

<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 -->
hicqu: /retest
hicqu: /retest
ti-chi-bot[bot]: @hicqu: 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-compatibility-test | 244af95950cf798b6fb39e7676c3d7fc62ac5a72 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_compatibility_test/651/display/redirect) | true | `/test dm-compatibility-test`
pull-cdc-integration-kafka-test | 244af95950cf798b6fb39e7676c3d7fc62ac5a72 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/679/display/redirect) | true | `/test cdc-integration-kafka-test`

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

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cdc-integration-kafka-test | 244af95950cf798b6fb39e7676c3d7fc62ac5a72 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/679/display/redirect) | true | `/test cdc-integration-kafka-test`
pull-dm-compatibility-test | 244af95950cf798b6fb39e7676c3d7fc62ac5a72 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_compatibility_test/653/display/redirect) | true | `/test dm-compatibility-test`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
asddongmen: I think it is worth mention this will also improve TiCDC ability of replicating big table. 
If I am right, after this PR, a single TiCDC node(8c, 16gb) can replicate a table that has more than 1.5 million regions. 
ti-chi-bot[bot]: @hicqu: 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-cdc-integration-kafka-test | e0f26d14bc3533ee585d40217484237337397104 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/690/display/redirect) | true | `/test cdc-integration-kafka-test`

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

<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 -->
hicqu: /retest
ti-chi-bot[bot]: @hicqu: 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-cdc-integration-kafka-test | e0f26d14bc3533ee585d40217484237337397104 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/695/display/redirect) | true | `/test cdc-integration-kafka-test`

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

<details>

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

Patch:
diff --git a/cdc/kv/client.go b/cdc/kv/client.go
index f15ac1f25..4245f1ac1 100644
--- a/cdc/kv/client.go
+++ b/cdc/kv/client.go
@@ -19,7 +19,6 @@ import (
 	"io"
 	"math/rand"
 	"strconv"
-	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -1519,7 +1518,7 @@ func (s *eventFeedSession) logSlowRegions(ctx context.Context) error {
 		}
 
 		attr := s.rangeLock.CollectLockedRangeAttrs(nil)
-		ckptTime := oracle.GetTimeFromTS(attr.SlowestRegion.CheckpointTs)
+		ckptTime := oracle.GetTimeFromTS(attr.SlowestRegion.ResolvedTs)
 		currTime := s.client.pdClock.CurrentTime()
 		log.Info("event feed starts to check locked regions",
 			zap.String("namespace", s.changefeed.Namespace),
@@ -1555,16 +1554,12 @@ func (s *eventFeedSession) logSlowRegions(ctx context.Context) error {
 				zap.Any("slowRegion", attr.SlowestRegion))
 		}
 		if len(attr.Holes) > 0 {
-			holes := make([]string, 0, len(attr.Holes))
-			for _, hole := range attr.Holes {
-				holes = append(holes, fmt.Sprintf("[%s,%s)", hole.StartKey, hole.EndKey))
-			}
 			log.Info("event feed holes exist",
 				zap.String("namespace", s.changefeed.Namespace),
 				zap.String("changefeed", s.changefeed.ID),
 				zap.Int64("tableID", s.tableID),
 				zap.String("tableName", s.tableName),
-				zap.String("holes", strings.Join(holes, ", ")))
+				zap.Any("holes", attr.Holes))
 		}
 	}
 }
diff --git a/cdc/kv/region_state.go b/cdc/kv/region_state.go
index 1263eaee9..321e22c5e 100644
--- a/cdc/kv/region_state.go
+++ b/cdc/kv/region_state.go
@@ -53,7 +53,7 @@ func newSingleRegionInfo(
 }
 
 func (s singleRegionInfo) resolvedTs() uint64 {
-	return s.lockedRange.CheckpointTs.Load()
+	return s.lockedRange.ResolvedTs.Load()
 }
 
 type regionFeedState struct {
@@ -134,18 +134,18 @@ func (s *regionFeedState) getRegionID() uint64 {
 }
 
 func (s *regionFeedState) getLastResolvedTs() uint64 {
-	return s.sri.lockedRange.CheckpointTs.Load()
+	return s.sri.lockedRange.ResolvedTs.Load()
 }
 
 // updateResolvedTs update the resolved ts of the current region feed
 func (s *regionFeedState) updateResolvedTs(resolvedTs uint64) {
 	state := s.sri.lockedRange
 	for {
-		last := state.CheckpointTs.Load()
+		last := state.ResolvedTs.Load()
 		if last > resolvedTs {
 			return
 		}
-		if state.CheckpointTs.CompareAndSwap(last, resolvedTs) {
+		if state.ResolvedTs.CompareAndSwap(last, resolvedTs) {
 			break
 		}
 	}
diff --git a/cdc/kv/regionlock/region_range_lock.go b/cdc/kv/regionlock/region_range_lock.go
index 8d4a91991..9502e58f4 100644
--- a/cdc/kv/regionlock/region_range_lock.go
+++ b/cdc/kv/regionlock/region_range_lock.go
@@ -31,84 +31,213 @@ import (
 )
 
 type rangeTsEntry struct {
-	// Only startKey is necessary. End key can be inferred by the next item since the map always keeps a continuous
-	// range.
+	// Only startKey is necessary. End key can be inferred by the next item,
+	// since the map always keeps a continuous range.
 	startKey []byte
 	ts       uint64
+
+	// rangeTsEntry is used in rangeTsMap. rangeTsMap.set will associate a timestamp to a given key range,
+	// and rangeTsMap.unset will revoke the relationship. `isSet` indicates whether a key range is
+	// generated from a set or unset operation.
+	isSet bool
 }
 
-func rangeTsEntryWithKey(key []byte) *rangeTsEntry {
-	return &rangeTsEntry{
-		startKey: key,
-	}
+func rangeTsEntryWithKey(key []byte) rangeTsEntry {
+	return rangeTsEntry{startKey: key}
 }
 
-func rangeTsEntryLess(a, b *rangeTsEntry) bool {
+func rangeTsEntryLess(a, b rangeTsEntry) bool {
 	return bytes.Compare(a.startKey, b.startKey) < 0
 }
 
 // rangeTsMap represents a map from key range to a timestamp. It supports
 // range set and calculating min value among a specified range.
 type rangeTsMap struct {
-	m *btree.BTreeG[*rangeTsEntry]
+	m     *btree.BTreeG[rangeTsEntry]
+	start []byte
+	end   []byte
 }
 
 // newRangeTsMap creates a RangeTsMap.
 func newRangeTsMap(startKey, endKey []byte, startTs uint64) *rangeTsMap {
 	m := &rangeTsMap{
-		m: btree.NewG(16, rangeTsEntryLess),
+		m:     btree.NewG(16, rangeTsEntryLess),
+		start: startKey,
+		end:   endKey,
 	}
-	m.Set(startKey, endKey, startTs)
+	m.set(startKey, endKey, startTs)
 	return m
 }
 
-// Set sets the corresponding ts of the given range to the specified value.
-func (m *rangeTsMap) Set(startKey, endKey []byte, ts uint64) {
-	if _, ok := m.m.Get(rangeTsEntryWithKey(endKey)); !ok {
-		// To calculate the minimal ts, the default value is math.MaxUint64
-		tailTs := uint64(math.MaxUint64)
-		m.m.DescendLessOrEqual(rangeTsEntryWithKey(endKey), func(i *rangeTsEntry) bool {
-			tailTs = i.ts
+func (m *rangeTsMap) clone() (res *rangeTsMap) {
+	res = &rangeTsMap{
+		m:     btree.NewG(16, rangeTsEntryLess),
+		start: m.start,
+		end:   m.end,
+	}
+	m.m.Ascend(func(i rangeTsEntry) bool {
+		res.m.ReplaceOrInsert(i)
+		return true
+	})
+	return
+}
+
+// set the timestamp of range [startKey, endKey) to ts.
+// RegionRangeLock uses rangeTsMap to store unsubscribed regions.
+// So `set` should be called after a region gets unlocked from RegionRangeLock.
+// Note: It leaves the timestamp of range [endKey, +∞) intact.
+func (m *rangeTsMap) set(startKey, endKey []byte, ts uint64) {
+	startEntry := rangeTsEntryWithKey(startKey)
+	endEntry := rangeTsEntryWithKey(endKey)
+
+	// Check and update the range overlapped with endKey is it exists.
+	if !bytes.Equal(m.end, endKey) && !m.m.Has(endEntry) {
+		var found bool
+		var endKeyOverlapped rangeTsEntry
+		m.m.DescendLessOrEqual(endEntry, func(i rangeTsEntry) bool {
+			found = true
+			endKeyOverlapped.ts = i.ts
+			endKeyOverlapped.isSet = i.isSet
 			return false
 		})
-		m.m.ReplaceOrInsert(&rangeTsEntry{
-			startKey: endKey,
-			ts:       tailTs,
-		})
+		if found {
+			if endKeyOverlapped.isSet {
+				log.Panic("rangeTsMap double set")
+			}
+			endKeyOverlapped.startKey = endKey
+			m.m.ReplaceOrInsert(endKeyOverlapped)
+		}
 	}
 
-	entriesToDelete := make([]*rangeTsEntry, 0)
-	m.m.AscendRange(rangeTsEntryWithKey(startKey), rangeTsEntryWithKey(endKey),
-		func(i *rangeTsEntry) bool {
-			entriesToDelete = append(entriesToDelete, i)
-			return true
+	// Check and update the range overlapped with startKey if it exists.
+	if !m.m.Has(startEntry) {
+		var found bool
+		var startKeyOverlapped rangeTsEntry
+		m.m.DescendLessOrEqual(startEntry, func(i rangeTsEntry) bool {
+			found = true
+			startKeyOverlapped.isSet = i.isSet
+			return false
 		})
+		if found && startKeyOverlapped.isSet {
+			log.Panic("rangeTsMap double set")
+		}
+	}
 
+	// Check and delete all covered ranges.
+	entriesToDelete := make([]rangeTsEntry, 0)
+	m.m.AscendRange(startEntry, endEntry, func(i rangeTsEntry) bool {
+		if i.isSet {
+			log.Panic("rangeTsMap double set")
+		}
+		entriesToDelete = append(entriesToDelete, i)
+		return true
+	})
 	for _, e := range entriesToDelete {
 		m.m.Delete(e)
 	}
 
-	m.m.ReplaceOrInsert(&rangeTsEntry{
-		startKey: startKey,
-		ts:       ts,
+	m.m.ReplaceOrInsert(rangeTsEntry{startKey: startKey, ts: ts, isSet: true})
+}
+
+// RegionRangeLock uses rangeTsMap to store unsubscribed regions.
+// So `unset` should be called after a region gets locked in RegionRangeLock.
+func (m *rangeTsMap) unset(startKey, endKey []byte) {
+	var neighbor rangeTsEntry
+	var exist bool
+	startEntry := rangeTsEntryWithKey(startKey)
+	endEntry := rangeTsEntryWithKey(endKey)
+
+	// Check and update the range overlapped with endKey is it exists.
+	if !bytes.Equal(m.end, endKey) {
+		if neighbor, exist = m.m.Get(endEntry); !exist {
+			var found bool
+			var endKeyOverlapped rangeTsEntry
+			m.m.DescendLessOrEqual(endEntry, func(i rangeTsEntry) bool {
+				found = true
+				endKeyOverlapped.ts = i.ts
+				endKeyOverlapped.isSet = i.isSet
+				return false
+			})
+			if found {
+				if !endKeyOverlapped.isSet {
+					log.Panic("rangeTsMap double unset")
+				}
+				endKeyOverlapped.startKey = endKey
+				m.m.ReplaceOrInsert(endKeyOverlapped)
+			}
+		} else if !neighbor.isSet {
+			m.m.Delete(neighbor)
+		}
+	}
+
+	// Check and update the range overlapped with startKey if it exists.
+	exist = false
+	m.m.DescendLessOrEqual(startEntry, func(i rangeTsEntry) bool {
+		if bytes.Compare(i.startKey, startKey) < 0 {
+			neighbor = i
+			exist = true
+			return false
+		}
+		return true
+	})
+	shouldInsert := !exist || neighbor.isSet
+
+	// Check and delete all covered ranges.
+	entriesToDelete := make([]rangeTsEntry, 0)
+	m.m.AscendRange(startEntry, endEntry, func(i rangeTsEntry) bool {
+		if !i.isSet {
+			log.Panic("rangeTsMap double unset")
+		}
+		entriesToDelete = append(entriesToDelete, i)
+		return true
 	})
+	for _, e := range entriesToDelete {
+		m.m.Delete(e)
+	}
+
+	if shouldInsert {
+		m.m.ReplaceOrInsert(rangeTsEntry{startKey: startKey, isSet: false})
+	}
 }
 
-// GetMin gets the min ts value among the given range. endKey must be greater than startKey.
-func (m *rangeTsMap) GetMin(startKey, endKey []byte) uint64 {
+func (m *rangeTsMap) getMinTsInRange(startKey, endKey []byte) uint64 {
 	var ts uint64 = math.MaxUint64
-	m.m.DescendLessOrEqual(rangeTsEntryWithKey(startKey), func(i *rangeTsEntry) bool {
-		ts = i.ts
-		return false
-	})
-	m.m.AscendRange(rangeTsEntryWithKey(startKey), rangeTsEntryWithKey(endKey),
-		func(i *rangeTsEntry) bool {
-			thisTs := i.ts
-			if ts > thisTs {
-				ts = thisTs
+
+	startEntry := rangeTsEntryWithKey(startKey)
+	if _, ok := m.m.Get(startEntry); !ok {
+		m.m.DescendLessOrEqual(startEntry, func(i rangeTsEntry) bool {
+			if !i.isSet {
+				log.Panic("rangeTsMap get after unset")
 			}
-			return true
+			ts = i.ts
+			return false
 		})
+	}
+
+	endEntry := rangeTsEntryWithKey(endKey)
+	m.m.AscendRange(startEntry, endEntry, func(i rangeTsEntry) bool {
+		if !i.isSet {
+			log.Panic("rangeTsMap get after unset")
+		}
+		if ts > i.ts {
+			ts = i.ts
+		}
+		return true
+	})
+
+	return ts
+}
+
+func (m *rangeTsMap) getMinTs() uint64 {
+	var ts uint64 = math.MaxUint64
+
+	m.m.Ascend(func(i rangeTsEntry) bool {
+		if i.isSet && ts > i.ts {
+			ts = i.ts
+		}
+		return true
+	})
+
 	return ts
 }
 
@@ -150,12 +279,11 @@ type RegionRangeLock struct {
 	totalSpan         tablepb.Span
 	changefeedLogInfo string
 
-	mu                sync.Mutex
-	rangeCheckpointTs *rangeTsMap
-	rangeLock         *btree.BTreeG[*rangeLockEntry]
-	regionIDLock      map[uint64]*rangeLockEntry
-	stopped           bool
-	refCount          uint64
+	mu              sync.RWMutex
+	rangeResolvedTs *rangeTsMap
+	rangeLock       *btree.BTreeG[*rangeLockEntry]
+	regionIDLock    map[uint64]*rangeLockEntry
+	stopped         bool
 }
 
 // NewRegionRangeLock creates a new RegionRangeLock.
@@ -167,7 +295,7 @@ func NewRegionRangeLock(
 		id:                id,
 		totalSpan:         tablepb.Span{StartKey: startKey, EndKey: endKey},
 		changefeedLogInfo: changefeedLogInfo,
-		rangeCheckpointTs: newRangeTsMap(startKey, endKey, startTs),
+		rangeResolvedTs:   newRangeTsMap(startKey, endKey, startTs),
 		rangeLock:         btree.NewG(16, rangeLockEntryLess),
 		regionIDLock:      make(map[uint64]*rangeLockEntry),
 	}
@@ -218,31 +346,32 @@ func (l *RegionRangeLock) tryLockRange(startKey, endKey []byte, regionID, versio
 	overlappingEntries := l.getOverlappedEntries(startKey, endKey, regionID)
 
 	if len(overlappingEntries) == 0 {
-		checkpointTs := l.rangeCheckpointTs.GetMin(startKey, endKey)
+		resolvedTs := l.rangeResolvedTs.getMinTsInRange(startKey, endKey)
 		newEntry := &rangeLockEntry{
 			startKey: startKey,
 			endKey:   endKey,
 			regionID: regionID,
 			version:  version,
 		}
-		newEntry.state.CheckpointTs.Store(checkpointTs)
+		newEntry.state.ResolvedTs.Store(resolvedTs)
 		newEntry.state.Created = time.Now()
 		l.rangeLock.ReplaceOrInsert(newEntry)
 		l.regionIDLock[regionID] = newEntry
 
+		l.rangeResolvedTs.unset(startKey, endKey)
 		log.Debug("range locked",
 			zap.String("changefeed", l.changefeedLogInfo),
-			zap.Uint64("lockID", l.id), zap.Uint64("regionID", regionID),
+			zap.Uint64("lockID", l.id),
+			zap.Uint64("regionID", regionID),
 			zap.Uint64("version", version),
-			zap.Uint64("checkpointTs", checkpointTs),
+			zap.Uint64("resolvedTs", resolvedTs),
 			zap.String("startKey", hex.EncodeToString(startKey)),
 			zap.String("endKey", hex.EncodeToString(endKey)))
 
-		l.refCount += 1
 		return LockRangeResult{
-			Status:       LockRangeStatusSuccess,
-			CheckpointTs: checkpointTs,
-			LockedRange:  &newEntry.state,
+			Status:      LockRangeStatusSuccess,
+			ResolvedTs:  resolvedTs,
+			LockedRange: &newEntry.state,
 		}, nil
 	}
 
@@ -348,11 +477,11 @@ func (l *RegionRangeLock) LockRange(
 	return res
 }
 
-// UnlockRange unlocks a range and update checkpointTs of the range to specified value.
+// UnlockRange unlocks a range and update resolvedTs of the range to specified value.
 // If it returns true it means it is stopped and all ranges are unlocked correctly.
 func (l *RegionRangeLock) UnlockRange(
 	startKey, endKey []byte, regionID, version uint64,
-	checkpointTs ...uint64,
+	resolvedTs ...uint64,
 ) (drained bool) {
 	l.mu.Lock()
 	defer l.mu.Unlock()
@@ -382,8 +511,7 @@ func (l *RegionRangeLock) UnlockRange(
 			zap.String("regionIDLockEntry", l.regionIDLock[regionID].String()))
 	}
 	delete(l.regionIDLock, regionID)
-	l.refCount -= 1
-	drained = l.stopped && l.refCount == 0
+	drained = l.stopped && len(l.regionIDLock) == 0
 
 	if entry.version != version || !bytes.Equal(entry.endKey, endKey) {
 		log.Panic("unlocking region doesn't match the locked region",
@@ -403,18 +531,18 @@ func (l *RegionRangeLock) UnlockRange(
 		panic("unreachable")
 	}
 
-	var newCheckpointTs uint64
-	if len(checkpointTs) > 0 {
-		newCheckpointTs = checkpointTs[0]
+	var newResolvedTs uint64
+	if len(resolvedTs) > 0 {
+		newResolvedTs = resolvedTs[0]
 	} else {
-		newCheckpointTs = entry.state.CheckpointTs.Load()
+		newResolvedTs = entry.state.ResolvedTs.Load()
 	}
 
-	l.rangeCheckpointTs.Set(startKey, endKey, newCheckpointTs)
+	l.rangeResolvedTs.set(startKey, endKey, newResolvedTs)
 	log.Debug("unlocked range",
 		zap.String("changefeed", l.changefeedLogInfo),
 		zap.Uint64("lockID", l.id), zap.Uint64("regionID", entry.regionID),
-		zap.Uint64("checkpointTs", newCheckpointTs),
+		zap.Uint64("resolvedTs", newResolvedTs),
 		zap.String("startKey", hex.EncodeToString(startKey)),
 		zap.String("endKey", hex.EncodeToString(endKey)))
 	return
@@ -422,16 +550,16 @@ func (l *RegionRangeLock) UnlockRange(
 
 // LockedRanges returns count of locked ranges.
 func (l *RegionRangeLock) LockedRanges() int {
-	l.mu.Lock()
-	defer l.mu.Unlock()
+	l.mu.RLock()
+	defer l.mu.RUnlock()
 	return l.rangeLock.Len()
 }
 
 // RefCount returns how many ranges are locked.
 func (l *RegionRangeLock) RefCount() uint64 {
-	l.mu.Lock()
-	defer l.mu.Unlock()
-	return l.refCount
+	l.mu.RLock()
+	defer l.mu.RUnlock()
+	return uint64(len(l.regionIDLock))
 }
 
 // Stop stops the instance.
@@ -439,7 +567,7 @@ func (l *RegionRangeLock) Stop() (drained bool) {
 	l.mu.Lock()
 	defer l.mu.Unlock()
 	l.stopped = true
-	return l.stopped && l.refCount == 0
+	return l.stopped && len(l.regionIDLock) == 0
 }
 
 const (
@@ -455,7 +583,7 @@ const (
 
 // LockRangeResult represents the result of LockRange method of RegionRangeLock.
 // If Status is LockRangeStatusSuccess:
-//   - CheckpointTs will be the minimal checkpoint ts among the locked range;
+//   - ResolvedTs will be the minimal checkpoint ts among the locked range;
 //   - LockedRange is for recording real-time state changes;
 //
 // If Status is LockRangeStatusWait, it means the lock cannot be acquired immediately. WaitFn must be invoked to
@@ -464,31 +592,31 @@ const (
 // If Status is LockRangeStatusStale, it means the LockRange request is stale because there's already a overlapping
 // locked range, whose version is greater or equals to the requested one.
 type LockRangeResult struct {
-	Status       int
-	CheckpointTs uint64
-	LockedRange  *LockedRange
-	WaitFn       func() LockRangeResult
-	RetryRanges  []tablepb.Span
+	Status      int
+	ResolvedTs  uint64
+	LockedRange *LockedRange
+	WaitFn      func() LockRangeResult
+	RetryRanges []tablepb.Span
 }
 
 // LockedRange is returned by `RegionRangeLock.LockRange`, which can be used to
 // collect informations for the range. And collected informations can be accessed
 // by iterating `RegionRangeLock`.
 type LockedRange struct {
-	CheckpointTs atomic.Uint64
-	Initialzied  atomic.Bool
-	Created      time.Time
+	ResolvedTs  atomic.Uint64
+	Initialzied atomic.Bool
+	Created     time.Time
 }
 
 // CollectLockedRangeAttrs collects locked range attributes.
 func (l *RegionRangeLock) CollectLockedRangeAttrs(
 	action func(regionID, version uint64, state *LockedRange, span tablepb.Span),
 ) (r CollectedLockedRangeAttrs) {
-	l.mu.Lock()
-	defer l.mu.Unlock()
+	l.mu.RLock()
+	defer l.mu.RUnlock()
 	r.LockedRegionCount = l.rangeLock.Len()
-	r.FastestRegion.CheckpointTs = 0
-	r.SlowestRegion.CheckpointTs = math.MaxUint64
+	r.FastestRegion.ResolvedTs = 0
+	r.SlowestRegion.ResolvedTs = math.MaxUint64
 
 	lastEnd := l.totalSpan.StartKey
 	l.rangeLock.Ascend(func(item *rangeLockEntry) bool {
@@ -497,18 +625,20 @@ func (l *RegionRangeLock) CollectLockedRangeAttrs(
 			action(item.regionID, item.version, &item.state, span)
 		}
 		if spanz.EndCompare(lastEnd, item.startKey) < 0 {
-			r.Holes = append(r.Holes, tablepb.Span{StartKey: lastEnd, EndKey: item.startKey})
+			span := tablepb.Span{StartKey: lastEnd, EndKey: item.startKey}
+			ts := l.rangeResolvedTs.getMinTsInRange(lastEnd, item.startKey)
+			r.Holes = append(r.Holes, HoleAttrs{Span: span, ResolvedTs: ts})
 		}
-		ckpt := item.state.CheckpointTs.Load()
-		if ckpt > r.FastestRegion.CheckpointTs {
+		ckpt := item.state.ResolvedTs.Load()
+		if ckpt > r.FastestRegion.ResolvedTs {
 			r.FastestRegion.RegionID = item.regionID
-			r.FastestRegion.CheckpointTs = ckpt
+			r.FastestRegion.ResolvedTs = ckpt
 			r.FastestRegion.Initialized = item.state.Initialzied.Load()
 			r.FastestRegion.Created = item.state.Created
 		}
-		if ckpt < r.SlowestRegion.CheckpointTs {
+		if ckpt < r.SlowestRegion.ResolvedTs {
 			r.SlowestRegion.RegionID = item.regionID
-			r.SlowestRegion.CheckpointTs = ckpt
+			r.SlowestRegion.ResolvedTs = ckpt
 			r.SlowestRegion.Initialized = item.state.Initialzied.Load()
 			r.SlowestRegion.Created = item.state.Created
 		}
@@ -516,7 +646,9 @@ func (l *RegionRangeLock) CollectLockedRangeAttrs(
 		return true
 	})
 	if spanz.EndCompare(lastEnd, l.totalSpan.EndKey) < 0 {
-		r.Holes = append(r.Holes, tablepb.Span{StartKey: lastEnd, EndKey: l.totalSpan.EndKey})
+		span := tablepb.Span{StartKey: lastEnd, EndKey: l.totalSpan.EndKey}
+		ts := l.rangeResolvedTs.getMinTsInRange(lastEnd, l.totalSpan.EndKey)
+		r.Holes = append(r.Holes, HoleAttrs{Span: span, ResolvedTs: ts})
 	}
 	return
 }
@@ -524,15 +656,47 @@ func (l *RegionRangeLock) CollectLockedRangeAttrs(
 // CollectedLockedRangeAttrs returns by `RegionRangeLock.CollectedLockedRangeAttrs`.
 type CollectedLockedRangeAttrs struct {
 	LockedRegionCount int
-	Holes             []tablepb.Span
-	FastestRegion     LockedRangeAttrs
-	SlowestRegion     LockedRangeAttrs
+
+	// If Holes isn't empty it means some regions are not captured at the time.
+	// Those regions can just split or merge, or be transferred, or temporarily be unavailable.
+	Holes []HoleAttrs
+
+	FastestRegion LockedRangeAttrs
+	SlowestRegion LockedRangeAttrs
 }
 
 // LockedRangeAttrs is like `LockedRange`, but only contains some read-only attributes.
 type LockedRangeAttrs struct {
-	RegionID     uint64
-	CheckpointTs uint64
-	Initialized  bool
-	Created      time.Time
+	RegionID    uint64
+	ResolvedTs  uint64
+	Initialized bool
+	Created     time.Time
+}
+
+// HoleAttrs is used for `CollectedLockedRangeAttrs`.
+type HoleAttrs struct {
+	Span       tablepb.Span
+	ResolvedTs uint64
+}
+
+// CalculateMinResolvedTs gets the minimum checkpoint timestamp from range lock.
+func (l *RegionRangeLock) CalculateMinResolvedTs() uint64 {
+	l.mu.RLock()
+	defer l.mu.RUnlock()
+
+	var minTs uint64 = math.MaxUint64
+	l.rangeLock.Ascend(func(item *rangeLockEntry) bool {
+		ts := item.state.ResolvedTs.Load()
+		if ts < minTs {
+			minTs = ts
+		}
+		return true
+	})
+
+	unlockedMinTs := l.rangeResolvedTs.getMinTs()
+	if unlockedMinTs < minTs {
+		minTs = unlockedMinTs
+	}
+
+	return minTs
 }
diff --git a/cdc/kv/shared_client.go b/cdc/kv/shared_client.go
index c9cfbfc50..69588d0f3 100644
--- a/cdc/kv/shared_client.go
+++ b/cdc/kv/shared_client.go
@@ -16,8 +16,6 @@ package kv
 import (
 	"context"
 	"encoding/binary"
-	"fmt"
-	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -124,6 +122,8 @@ type requestedTable struct {
 	rangeLock *regionlock.RegionRangeLock
 	eventCh   chan<- MultiplexingEvent
 
+	lastAdvanceTime atomic.Int64
+
 	// To handle table removing.
 	stopped atomic.Bool
 
@@ -671,7 +671,7 @@ func (s *SharedClient) resolveLock(ctx context.Context) error {
 	}
 
 	doResolve := func(regionID uint64, state *regionlock.LockedRange, maxVersion uint64) {
-		if state.CheckpointTs.Load() > maxVersion || !state.Initialzied.Load() {
+		if state.ResolvedTs.Load() > maxVersion || !state.Initialzied.Load() {
 			return
 		}
 		if lastRun, ok := resolveLastRun[regionID]; ok {
@@ -725,7 +725,7 @@ func (s *SharedClient) logSlowRegions(ctx context.Context) error {
 		s.totalSpans.RLock()
 		for subscriptionID, rt := range s.totalSpans.v {
 			attr := rt.rangeLock.CollectLockedRangeAttrs(nil)
-			ckptTime := oracle.GetTimeFromTS(attr.SlowestRegion.CheckpointTs)
+			ckptTime := oracle.GetTimeFromTS(attr.SlowestRegion.ResolvedTs)
 			if attr.SlowestRegion.Initialized {
 				if currTime.Sub(ckptTime) > 2*resolveLockMinInterval {
 					log.Info("event feed finds a initialized slow region",
@@ -748,15 +748,11 @@ func (s *SharedClient) logSlowRegions(ctx context.Context) error {
 					zap.Any("slowRegion", attr.SlowestRegion))
 			}
 			if len(attr.Holes) > 0 {
-				holes := make([]string, 0, len(attr.Holes))
-				for _, hole := range attr.Holes {
-					holes = append(holes, fmt.Sprintf("[%s,%s)", hole.StartKey, hole.EndKey))
-				}
 				log.Info("event feed holes exist",
 					zap.String("namespace", s.changefeed.Namespace),
 					zap.String("changefeed", s.changefeed.ID),
 					zap.Any("subscriptionID", subscriptionID),
-					zap.String("holes", strings.Join(holes, ", ")))
+					zap.Any("holes", attr.Holes))
 			}
 		}
 		s.totalSpans.RUnlock()
@@ -780,7 +776,7 @@ func (s *SharedClient) newRequestedTable(
 
 	rt.postUpdateRegionResolvedTs = func(regionID, _ uint64, state *regionlock.LockedRange, _ tablepb.Span) {
 		maxVersion := rt.staleLocksVersion.Load()
-		if state.CheckpointTs.Load() <= maxVersion && state.Initialzied.Load() {
+		if state.ResolvedTs.Load() <= maxVersion && state.Initialzied.Load() {
 			enter := time.Now()
 			s.resolveLockCh.In() <- resolveLockTask{regionID, maxVersion, state, enter}
 		}
diff --git a/cdc/kv/shared_region_worker.go b/cdc/kv/shared_region_worker.go
index de1448ce4..b2cca89be 100644
--- a/cdc/kv/shared_region_worker.go
+++ b/cdc/kv/shared_region_worker.go
@@ -39,6 +39,7 @@ type eventItem struct {
 	state *regionFeedState
 }
 
+// NOTE: all regions must come from the same requestedTable, and regions will never be empty.
 type resolvedTsBatch struct {
 	ts      uint64
 	regions []*regionFeedState
@@ -174,6 +175,14 @@ func (w *sharedRegionWorker) handleEventEntry(ctx context.Context, x *cdcpb.Even
 }
 
 func (w *sharedRegionWorker) handleResolvedTs(ctx context.Context, batch resolvedTsBatch) {
+	if w.client.config.KVClient.AdvanceIntervalInMs > 0 {
+		w.advanceTableSpan(ctx, batch)
+	} else {
+		w.forwardResolvedTsToPullerFrontier(ctx, batch)
+	}
+}
+
+func (w *sharedRegionWorker) forwardResolvedTsToPullerFrontier(ctx context.Context, batch resolvedTsBatch) {
 	resolvedSpans := make(map[SubscriptionID]*struct {
 		spans          []model.RegionComparableSpan
 		requestedTable *requestedTable
@@ -231,3 +240,45 @@ func (w *sharedRegionWorker) handleResolvedTs(ctx context.Context, batch resolve
 		}
 	}
 }
+
+func (w *sharedRegionWorker) advanceTableSpan(ctx context.Context, batch resolvedTsBatch) {
+	for _, state := range batch.regions {
+		if state.isStale() || !state.isInitialized() {
+			continue
+		}
+
+		regionID := state.getRegionID()
+		lastResolvedTs := state.getLastResolvedTs()
+		if batch.ts < lastResolvedTs {
+			log.Debug("The resolvedTs is fallen back in kvclient",
+				zap.String("namespace", w.changefeed.Namespace),
+				zap.String("changefeed", w.changefeed.ID),
+				zap.Uint64("regionID", regionID),
+				zap.Uint64("resolvedTs", batch.ts),
+				zap.Uint64("lastResolvedTs", lastResolvedTs))
+			continue
+		}
+		state.updateResolvedTs(batch.ts)
+	}
+
+	rt := batch.regions[0].sri.requestedTable
+	now := time.Now().UnixMilli()
+	lastAdvance := rt.lastAdvanceTime.Load()
+	if now-lastAdvance > int64(w.client.config.KVClient.AdvanceIntervalInMs) && rt.lastAdvanceTime.CompareAndSwap(lastAdvance, now) {
+		ts := rt.rangeLock.CalculateMinResolvedTs()
+		if ts > rt.startTs {
+			revent := model.RegionFeedEvent{
+				Resolved: &model.ResolvedSpans{
+					Spans:      []model.RegionComparableSpan{{Span: rt.span, Region: 0}},
+					ResolvedTs: ts,
+				},
+			}
+			x := rt.associateSubscriptionID(revent)
+			select {
+			case rt.eventCh <- x:
+				w.metrics.metricSendEventResolvedCounter.Add(float64(len(batch.regions)))
+			case <-ctx.Done():
+			}
+		}
+	}
+}
diff --git a/cdc/syncpointstore/mysql_syncpoint_store.go b/cdc/syncpointstore/mysql_syncpoint_store.go
index ce4e88df2..53aaa4f0c 100644
--- a/cdc/syncpointstore/mysql_syncpoint_store.go
+++ b/cdc/syncpointstore/mysql_syncpoint_store.go
@@ -26,17 +26,11 @@ import (
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/errorutil"
+	"github.com/pingcap/tiflow/pkg/filter"
 	"github.com/pingcap/tiflow/pkg/sink/mysql"
 	"go.uber.org/zap"
 )
 
-const (
-	// syncPointTableName is the name of table where all syncpoint maps sit
-	syncPointTableName string = "syncpoint_v1"
-	// schemaName is the name of database where syncPoint maps sit
-	schemaName = "tidb_cdc"
-)
-
 type mysqlSyncPointStore struct {
 	db                     *sql.DB
 	clusterID              string
@@ -88,7 +82,7 @@ func newMySQLSyncPointStore(
 }
 
 func (s *mysqlSyncPointStore) CreateSyncTable(ctx context.Context) error {
-	database := schemaName
+	database := filter.TiCDCSystemSchema
 	tx, err := s.db.BeginTx(ctx, nil)
 	if err != nil {
 		log.Error("create sync table: begin Tx fail", zap.Error(err))
@@ -120,7 +114,7 @@ func (s *mysqlSyncPointStore) CreateSyncTable(ctx context.Context) error {
 		INDEX (created_at),
 		PRIMARY KEY (changefeed, primary_ts)
 	);`
-	query = fmt.Sprintf(query, syncPointTableName)
+	query = fmt.Sprintf(query, filter.SyncPointTable)
 	_, err = tx.Exec(query)
 	if err != nil {
 		err2 := tx.Rollback()
@@ -154,7 +148,7 @@ func (s *mysqlSyncPointStore) SinkSyncPoint(ctx context.Context,
 		return cerror.WrapError(cerror.ErrMySQLTxnError, err)
 	}
 	// insert ts map
-	query := "insert ignore into " + schemaName + "." + syncPointTableName +
+	query := "insert ignore into " + filter.TiCDCSystemSchema + "." + filter.SyncPointTable +
 		"(ticdc_cluster_id, changefeed, primary_ts, secondary_ts) VALUES (?,?,?,?)"
 	_, err = tx.Exec(query, s.clusterID, id.ID, checkpointTs, secondaryTs)
 	if err != nil {
@@ -186,8 +180,8 @@ func (s *mysqlSyncPointStore) SinkSyncPoint(ctx context.Context,
 	if time.Since(s.lastCleanSyncPointTime) >= s.syncPointRetention {
 		query = fmt.Sprintf(
 			"DELETE IGNORE FROM "+
-				schemaName+"."+
-				syncPointTableName+
+				filter.TiCDCSystemSchema+"."+
+				filter.SyncPointTable+
 				" WHERE ticdc_cluster_id = '%s' and changefeed = '%s' and created_at < (NOW() - INTERVAL %.2f SECOND)",
 			s.clusterID,
 			id.ID,
diff --git a/dm/syncer/ddl.go b/dm/syncer/ddl.go
index 47035e75f..fdca8906d 100644
--- a/dm/syncer/ddl.go
+++ b/dm/syncer/ddl.go
@@ -1173,7 +1173,7 @@ func (ddl *DDLWorker) AstToDDLEvent(qec *queryEventContext, info *ddlInfo) (et b
 					}
 				}
 			case ast.AlterTableReorganizePartition:
-				return bf.ReorganizePartion
+				return bf.ReorganizePartition
 			case ast.AlterTableRebuildPartition:
 				return bf.RebuildPartition
 			case ast.AlterTableCoalescePartitions:
diff --git a/go.mod b/go.mod
index 1d280d822..c6982dfbd 100644
--- a/go.mod
+++ b/go.mod
@@ -69,7 +69,7 @@ require (
 	github.com/pingcap/kvproto v0.0.0-20231222062942-c0c73f41d0b2
 	github.com/pingcap/log v1.1.1-0.20230317032135-a0d097d16e22
 	github.com/pingcap/tidb v1.1.0-beta.0.20240105042433-54d8a1416ab0
-	github.com/pingcap/tidb-tools v0.0.0-20231228035519-c4bdf178b3d6
+	github.com/pingcap/tidb-tools v0.0.0-20240131075425-5bb51786a1dc
 	github.com/pingcap/tidb/pkg/parser v0.0.0-20231229060758-e19e06e1bc19
 	github.com/prometheus/client_golang v1.18.0
 	github.com/prometheus/client_model v0.5.0
diff --git a/go.sum b/go.sum
index 72a4e2408..d9f3afc89 100644
--- a/go.sum
+++ b/go.sum
@@ -867,8 +867,8 @@ github.com/pingcap/sysutil v1.0.1-0.20230407040306-fb007c5aff21 h1:QV6jqlfOkh8hq
 github.com/pingcap/sysutil v1.0.1-0.20230407040306-fb007c5aff21/go.mod h1:QYnjfA95ZaMefyl1NO8oPtKeb8pYUdnDVhQgf+qdpjM=
 github.com/pingcap/tidb v1.1.0-beta.0.20240105042433-54d8a1416ab0 h1:EAeL37UVvIMvAG5JP0qDSBWZvV4tKn0gUbSP05vjtqU=
 github.com/pingcap/tidb v1.1.0-beta.0.20240105042433-54d8a1416ab0/go.mod h1:1jrwjaNIwD9JmFR3NghQ/xu/PCuPQVLhK8sshyFadN0=
-github.com/pingcap/tidb-tools v0.0.0-20231228035519-c4bdf178b3d6 h1:NJQd14hU0pLj8vS+vYtgFw4bXrFIXa/HfMdxALHtgVU=
-github.com/pingcap/tidb-tools v0.0.0-20231228035519-c4bdf178b3d6/go.mod h1:7m72uoBoTrr8JNNA7iFBH824j3clIJx34zlUWJ45Q6Y=
+github.com/pingcap/tidb-tools v0.0.0-20240131075425-5bb51786a1dc h1:z9gMlNH+AMmp62EniIRtyVVgOsrFcuxiUu6Ty60uEMU=
+github.com/pingcap/tidb-tools v0.0.0-20240131075425-5bb51786a1dc/go.mod h1:7m72uoBoTrr8JNNA7iFBH824j3clIJx34zlUWJ45Q6Y=
 github.com/pingcap/tidb/pkg/parser v0.0.0-20231229060758-e19e06e1bc19 h1:nmFYQ4q3m0v7AleVVyiijMtEzfxCh8zqMeuD1npsy+w=
 github.com/pingcap/tidb/pkg/parser v0.0.0-20231229060758-e19e06e1bc19/go.mod h1:yRkiqLFwIqibYg2P7h4bclHjHcJiIFRLKhGRyBcKYus=
 github.com/pingcap/tipb v0.0.0-20230919054518-dfd7d194838f h1:NCiI4Wyu4GkViLGTu6cYcxt79LZ1SenBBQX1OwEV6Jg=
diff --git a/pkg/config/kvclient.go b/pkg/config/kvclient.go
index 9bdd8060e..f6a9c49af 100644
--- a/pkg/config/kvclient.go
+++ b/pkg/config/kvclient.go
@@ -22,6 +22,8 @@ type KVClientConfig struct {
 	WorkerConcurrent uint `toml:"worker-concurrent" json:"worker-concurrent"`
 	// how many grpc streams will be established to every TiKV node
 	GrpcStreamConcurrent uint `toml:"grpc-stream-concurrent" json:"grpc-stream-concurrent"`
+	// Advance table ResolvedTs interval.
+	AdvanceIntervalInMs uint `toml:"advance-interval-in-ms" json:"advance-interval-in-ms"`
 	// how many goroutines to maintain frontiers.
 	FrontierConcurrent uint `toml:"frontier-concurrent" json:"frontier-concurrent"`
 	// background workerpool size, the workrpool is shared by all goroutines in cdc server
diff --git a/pkg/config/server_config.go b/pkg/config/server_config.go
index 91f40e3f6..b362fe094 100644
--- a/pkg/config/server_config.go
+++ b/pkg/config/server_config.go
@@ -111,6 +111,7 @@ var defaultServerConfig = &ServerConfig{
 		EnableMultiplexing:   true,
 		WorkerConcurrent:     8,
 		GrpcStreamConcurrent: 1,
+		AdvanceIntervalInMs:  300,
 		FrontierConcurrent:   8,
 		WorkerPoolSize:       0, // 0 will use NumCPU() * 2
 		RegionScanLimit:      40,
diff --git a/pkg/filter/expr_filter.go b/pkg/filter/expr_filter.go
index 61835c6d6..25b04cf73 100644
--- a/pkg/filter/expr_filter.go
+++ b/pkg/filter/expr_filter.go
@@ -419,6 +419,9 @@ func (f *dmlExprFilter) shouldSkipDML(
 	rawRow model.RowChangedDatums,
 	ti *model.TableInfo,
 ) (bool, error) {
+	if len(f.rules) == 0 {
+		return false, nil
+	}
 	// for defense purpose, normally the row and ti should not be nil.
 	if ti == nil || row == nil || rawRow.IsEmpty() {
 		return false, nil
diff --git a/pkg/filter/filter.go b/pkg/filter/filter.go
index f1e464f7b..a3c017c34 100644
--- a/pkg/filter/filter.go
+++ b/pkg/filter/filter.go
@@ -14,51 +14,77 @@
 package filter
 
 import (
+	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	timodel "github.com/pingcap/tidb/pkg/parser/model"
 	tfilter "github.com/pingcap/tidb/pkg/util/table-filter"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/pkg/config"
 )
 
-// allowDDLList is a list of DDL types that can be applied to cdc's schema storage.
-// It's a white list.
-var allowDDLList = []timodel.ActionType{
-	timodel.ActionCreateSchema,
-	timodel.ActionDropSchema,
-	timodel.ActionCreateTable,
-	timodel.ActionDropTable,
-	timodel.ActionAddColumn,
-	timodel.ActionDropColumn,
-	timodel.ActionAddIndex,
-	timodel.ActionDropIndex,
-	timodel.ActionTruncateTable,
-	timodel.ActionModifyColumn,
-	timodel.ActionRenameTable,
-	timodel.ActionRenameTables,
-	timodel.ActionSetDefaultValue,
-	timodel.ActionModifyTableComment,
-	timodel.ActionRenameIndex,
-	timodel.ActionAddTablePartition,
-	timodel.ActionDropTablePartition,
-	timodel.ActionCreateView,
-	timodel.ActionModifyTableCharsetAndCollate,
-	timodel.ActionTruncateTablePartition,
-	timodel.ActionDropView,
-	timodel.ActionRecoverTable,
-	timodel.ActionModifySchemaCharsetAndCollate,
-	timodel.ActionAddPrimaryKey,
-	timodel.ActionDropPrimaryKey,
-	timodel.ActionAddColumns,  // Removed in TiDB v6.2.0, see https://github.com/pingcap/tidb/pull/35862.
-	timodel.ActionDropColumns, // Removed in TiDB v6.2.0
-	timodel.ActionRebaseAutoID,
-	timodel.ActionAlterIndexVisibility,
-	timodel.ActionMultiSchemaChange,
-	timodel.ActionExchangeTablePartition,
-	timodel.ActionReorganizePartition,
-	timodel.ActionAlterTTLInfo,
-	timodel.ActionAlterTTLRemove,
-	timodel.ActionAlterTablePartitioning,
-	timodel.ActionRemovePartitioning,
+const (
+	// SyncPointTable is the tale name use to write ts-map when sync-point is enable.
+	SyncPointTable = "syncpoint_v1"
+	// TiCDCSystemSchema is the schema only use by TiCDC.
+	TiCDCSystemSchema = "tidb_cdc"
+)
+
+// ddlWhiteListMap is a map of all DDL types that can be applied to cdc's schema storage.
+var ddlWhiteListMap = map[timodel.ActionType]bf.EventType{
+	// schema related DDLs
+	timodel.ActionCreateSchema:                  bf.CreateDatabase,
+	timodel.ActionDropSchema:                    bf.DropDatabase,
+	timodel.ActionModifySchemaCharsetAndCollate: bf.ModifySchemaCharsetAndCollate,
+
+	// table related DDLs
+	timodel.ActionCreateTable:                  bf.CreateTable,
+	timodel.ActionDropTable:                    bf.DropTable,
+	timodel.ActionTruncateTable:                bf.TruncateTable,
+	timodel.ActionRenameTable:                  bf.RenameTable,
+	timodel.ActionRenameTables:                 bf.RenameTable,
+	timodel.ActionRecoverTable:                 bf.RecoverTable,
+	timodel.ActionModifyTableComment:           bf.ModifyTableComment,
+	timodel.ActionModifyTableCharsetAndCollate: bf.ModifyTableCharsetAndCollate,
+
+	// view related DDLs
+	timodel.ActionCreateView: bf.CreateView,
+	timodel.ActionDropView:   bf.DropView,
+
+	// partition related DDLs
+	timodel.ActionAddTablePartition:      bf.AddTablePartition,
+	timodel.ActionDropTablePartition:     bf.DropTablePartition,
+	timodel.ActionTruncateTablePartition: bf.TruncateTablePartition,
+	timodel.ActionExchangeTablePartition: bf.ExchangePartition,
+	timodel.ActionReorganizePartition:    bf.ReorganizePartition,
+	timodel.ActionAlterTablePartitioning: bf.AlterTablePartitioning,
+	timodel.ActionRemovePartitioning:     bf.RemovePartitioning,
+
+	// column related DDLs
+	timodel.ActionAddColumn:       bf.AddColumn,
+	timodel.ActionDropColumn:      bf.DropColumn,
+	timodel.ActionModifyColumn:    bf.ModifyColumn,
+	timodel.ActionSetDefaultValue: bf.SetDefaultValue,
+
+	// index related DDLs
+	timodel.ActionRebaseAutoID:         bf.RebaseAutoID,
+	timodel.ActionAddPrimaryKey:        bf.AddPrimaryKey,
+	timodel.ActionDropPrimaryKey:       bf.DropPrimaryKey,
+	timodel.ActionAddIndex:             bf.CreateIndex,
+	timodel.ActionDropIndex:            bf.DropIndex,
+	timodel.ActionRenameIndex:          bf.RenameIndex,
+	timodel.ActionAlterIndexVisibility: bf.AlterIndexVisibility,
+
+	// TTL related DDLs
+	timodel.ActionAlterTTLInfo:   bf.AlterTTLInfo,
+	timodel.ActionAlterTTLRemove: bf.AlterTTLRemove,
+
+	// difficult to classify DDLs
+	timodel.ActionMultiSchemaChange: bf.MultiSchemaChange,
+
+	// deprecated DDLs,see https://github.com/pingcap/tidb/pull/35862.
+	// DDL types below are deprecated in TiDB v6.2.0, but we still keep them here
+	// In case that some users will use TiCDC to replicate data from TiDB v6.1.x.
+	timodel.ActionAddColumns:  bf.AddColumn,
+	timodel.ActionDropColumns: bf.DropColumn,
 }
 
 // Filter are safe for concurrent use.
@@ -110,7 +136,7 @@ func NewFilter(cfg *config.ReplicaConfig, tz string) (Filter, error) {
 	if err != nil {
 		return nil, err
 	}
-	sqlEventFilter, err := newSQLEventFilter(cfg.Filter, cfg.SQLMode)
+	sqlEventFilter, err := newSQLEventFilter(cfg.Filter)
 	if err != nil {
 		return nil, err
 	}
@@ -215,12 +241,8 @@ func (f *filter) shouldIgnoreStartTs(ts uint64) bool {
 }
 
 func isAllowedDDL(actionType timodel.ActionType) bool {
-	for _, action := range allowDDLList {
-		if actionType == action {
-			return true
-		}
-	}
-	return false
+	_, ok := ddlWhiteListMap[actionType]
+	return ok
 }
 
 // IsSchemaDDL returns true if the action type is a schema DDL.
diff --git a/pkg/filter/sql_event_filter.go b/pkg/filter/sql_event_filter.go
index b18618eb7..d9c1bf794 100644
--- a/pkg/filter/sql_event_filter.go
+++ b/pkg/filter/sql_event_filter.go
@@ -14,14 +14,9 @@
 package filter
 
 import (
-	"fmt"
-	"sync"
-
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
-	"github.com/pingcap/tidb/pkg/parser"
-	"github.com/pingcap/tidb/pkg/parser/mysql"
 	tfilter "github.com/pingcap/tidb/pkg/util/table-filter"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/pkg/config"
@@ -84,8 +79,8 @@ func newSQLEventFilterRule(cfg *config.EventFilterRule) (*sqlEventRule, error) {
 }
 
 func verifyIgnoreEvents(types []bf.EventType) error {
-	typesMap := make(map[bf.EventType]struct{}, len(supportedEventTypes))
-	for _, et := range supportedEventTypes {
+	typesMap := make(map[bf.EventType]struct{}, len(SupportedEventTypes()))
+	for _, et := range SupportedEventTypes() {
 		typesMap[et] = struct{}{}
 	}
 	for _, et := range types {
@@ -98,28 +93,11 @@ func verifyIgnoreEvents(types []bf.EventType) error {
 
 // sqlEventFilter is a filter that filters DDL/DML event by its type or query.
 type sqlEventFilter struct {
-	// Please be careful, parser.Parser is not thread safe.
-	pLock sync.Mutex
-	// Currently, parser is only used to parse ddl query.
-	// So we can use a lock to protect it.
-	// If we want to use it to parse dml query in the future,
-	// we should create a parser for each goroutine.
-	ddlParser *parser.Parser
-	rules     []*sqlEventRule
+	rules []*sqlEventRule
 }
 
-func newSQLEventFilter(cfg *config.FilterConfig, sqlMode string) (*sqlEventFilter, error) {
-	p := parser.New()
-	mode, err := mysql.GetSQLMode(sqlMode)
-	if err != nil {
-		log.Error("failed to get sql mode", zap.Error(err))
-		return nil, cerror.ErrInvalidReplicaConfig.FastGenByArgs(fmt.Sprintf("invalid sqlMode %s", sqlMode))
-	}
-	p.SetSQLMode(mode)
-
-	res := &sqlEventFilter{
-		ddlParser: p,
-	}
+func newSQLEventFilter(cfg *config.FilterConfig) (*sqlEventFilter, error) {
+	res := &sqlEventFilter{}
 	for _, rule := range cfg.EventFilters {
 		if err := res.addRule(rule); err != nil {
 			return nil, errors.Trace(err)
@@ -155,17 +133,12 @@ func (f *sqlEventFilter) getRules(schema, table string) []*sqlEventRule {
 
 // skipDDLEvent skips ddl event by its type and query.
 func (f *sqlEventFilter) shouldSkipDDL(ddl *model.DDLEvent) (bool, error) {
+	if len(f.rules) == 0 {
+		return false, nil
+	}
 	schema := ddl.TableInfo.TableName.Schema
 	table := ddl.TableInfo.TableName.Table
-	log.Info("sql event filter handle ddl event",
-		zap.Any("ddlType", ddl.Type), zap.String("schema", schema),
-		zap.String("table", table), zap.String("query", ddl.Query))
-	f.pLock.Lock()
-	evenType, err := ddlToEventType(f.ddlParser, ddl.Query, ddl.Type)
-	f.pLock.Unlock()
-	if err != nil {
-		return false, err
-	}
+	evenType := ddlToEventType(ddl.Type)
 	if evenType == bf.NullEvent {
 		log.Warn("sql event filter unsupported ddl type, do nothing",
 			zap.String("type", ddl.Type.String()),
@@ -185,12 +158,31 @@ func (f *sqlEventFilter) shouldSkipDDL(ddl *model.DDLEvent) (bool, error) {
 		if action == bf.Ignore {
 			return true, nil
 		}
+
+		// If the ddl is alter table's subtype,
+		// we need try to filter it by bf.AlterTable.
+		if isAlterTable(ddl.Type) {
+			action, err = rule.bf.Filter(
+				binlogFilterSchemaPlaceholder,
+				binlogFilterTablePlaceholder,
+				bf.AlterTable, ddl.Query)
+			if err != nil {
+				return false, errors.Trace(err)
+			}
+			if action == bf.Ignore {
+				return true, nil
+			}
+		}
 	}
 	return false, nil
 }
 
 // shouldSkipDML skips dml event by its type.
 func (f *sqlEventFilter) shouldSkipDML(event *model.RowChangedEvent) (bool, error) {
+	if len(f.rules) == 0 {
+		return false, nil
+	}
+
 	var et bf.EventType
 	switch {
 	case event.IsInsert():
@@ -216,29 +208,3 @@ func (f *sqlEventFilter) shouldSkipDML(event *model.RowChangedEvent) (bool, erro
 	}
 	return false, nil
 }
-
-var supportedEventTypes = []bf.EventType{
-	bf.AllDML,
-	bf.AllDDL,
-
-	// dml events
-	bf.InsertEvent,
-	bf.UpdateEvent,
-	bf.DeleteEvent,
-
-	// ddl events
-	bf.CreateSchema,
-	bf.CreateDatabase,
-	bf.DropSchema,
-	bf.DropDatabase,
-	bf.CreateTable,
-	bf.DropTable,
-	bf.RenameTable,
-	bf.TruncateTable,
-	bf.AlterTable,
-	bf.CreateView,
-	bf.DropView,
-	bf.AddTablePartition,
-	bf.DropTablePartition,
-	bf.TruncateTablePartition,
-}
diff --git a/pkg/filter/utils.go b/pkg/filter/utils.go
index 1a45d29cd..7a431ee58 100644
--- a/pkg/filter/utils.go
+++ b/pkg/filter/utils.go
@@ -17,7 +17,6 @@ import (
 	"fmt"
 
 	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
-	"github.com/pingcap/tidb/pkg/parser"
 	timodel "github.com/pingcap/tidb/pkg/parser/model"
 	tifilter "github.com/pingcap/tidb/pkg/util/filter"
 	tfilter "github.com/pingcap/tidb/pkg/util/table-filter"
@@ -27,7 +26,14 @@ import (
 
 // isSysSchema returns true if the given schema is a system schema
 func isSysSchema(db string) bool {
-	return tifilter.IsSystemSchema(db)
+	switch db {
+	// TiCDCSystemSchema is used by TiCDC only.
+	// Tables in TiCDCSystemSchema should not be replicated by cdc.
+	case TiCDCSystemSchema:
+		return true
+	default:
+		return tifilter.IsSystemSchema(db)
+	}
 }
 
 // VerifyTableRules checks the table filter rules in the configuration
@@ -51,37 +57,89 @@ func VerifyTableRules(cfg *config.FilterConfig) (tfilter.Filter, error) {
 }
 
 // ddlToEventType get event type from ddl query.
-func ddlToEventType(p *parser.Parser, query string, jobType timodel.ActionType) (bf.EventType, error) {
-	// Since `Parser` will return a AlterTable type `ast.StmtNode` for table partition related DDL,
-	// we need to check the ActionType of a ddl at first.
-	switch jobType {
-	case timodel.ActionAddTablePartition:
-		return bf.AddTablePartition, nil
-	case timodel.ActionDropTablePartition:
-		return bf.DropTablePartition, nil
-	case timodel.ActionTruncateTablePartition:
-		return bf.TruncateTablePartition, nil
-	}
-	stmt, err := p.ParseOneStmt(query, "", "")
-	if err != nil {
-		return bf.NullEvent, cerror.WrapError(cerror.ErrConvertDDLToEventTypeFailed, err, query)
+func ddlToEventType(jobType timodel.ActionType) bf.EventType {
+	evenType, ok := ddlWhiteListMap[jobType]
+	if ok {
+		return evenType
 	}
-	et := bf.AstToDDLEvent(stmt)
-	// `Parser` will return a `AlterTable` type `ast.StmtNode` for a query like:
-	// `alter table t1 add index (xxx)` and will return a `CreateIndex` type
-	// `ast.StmtNode` for a query like: `create index i on t1 (xxx)`.
-	// So we cast index related DDL to `AlterTable` event type for the sake of simplicity.
-	switch et {
-	case bf.DropIndex:
-		return bf.AlterTable, nil
-	case bf.CreateIndex:
-		return bf.AlterTable, nil
+	return bf.NullEvent
+}
+
+var alterTableSubType = []timodel.ActionType{
+	// table related DDLs
+	timodel.ActionRenameTable,
+	timodel.ActionRenameTables,
+	timodel.ActionModifyTableComment,
+	timodel.ActionModifyTableCharsetAndCollate,
+
+	// partition related DDLs
+	timodel.ActionAddTablePartition,
+	timodel.ActionDropTablePartition,
+	timodel.ActionTruncateTablePartition,
+	timodel.ActionExchangeTablePartition,
+	timodel.ActionReorganizePartition,
+	timodel.ActionAlterTablePartitioning,
+	timodel.ActionRemovePartitioning,
+
+	// column related DDLs
+	timodel.ActionAddColumn,
+	timodel.ActionDropColumn,
+	timodel.ActionModifyColumn,
+	timodel.ActionSetDefaultValue,
+
+	// index related DDLs
+	timodel.ActionRebaseAutoID,
+	timodel.ActionAddPrimaryKey,
+	timodel.ActionDropPrimaryKey,
+	timodel.ActionAddIndex,
+	timodel.ActionDropIndex,
+	timodel.ActionRenameIndex,
+	timodel.ActionAlterIndexVisibility,
+
+	// TTL related DDLs
+	timodel.ActionAlterTTLInfo,
+	timodel.ActionAlterTTLRemove,
+
+	// difficult to classify DDLs
+	timodel.ActionMultiSchemaChange,
+
+	// deprecated DDLs,see https://github.com/pingcap/tidb/pull/35862.
+	// DDL types below are deprecated in TiDB v6.2.0, but we still keep them here
+	// In case that some users will use TiCDC to replicate data from TiDB v6.1.x.
+	timodel.ActionAddColumns,
+	timodel.ActionDropColumns,
+}
+
+// isAlterTable returns true if the given job type is alter table's subtype.
+func isAlterTable(jobType timodel.ActionType) bool {
+	for _, t := range alterTableSubType {
+		if t == jobType {
+			return true
+		}
 	}
-	return et, nil
+	return false
 }
 
 // SupportedEventTypes returns the supported event types.
 func SupportedEventTypes() []bf.EventType {
+	supportedEventTypes := []bf.EventType{
+		bf.AllDML,
+		bf.AllDDL,
+
+		// dml events
+		bf.InsertEvent,
+		bf.UpdateEvent,
+		bf.DeleteEvent,
+
+		// ddl events
+		bf.AlterTable,
+		bf.CreateSchema,
+		bf.DropSchema,
+	}
+
+	for _, ddlType := range ddlWhiteListMap {
+		supportedEventTypes = append(supportedEventTypes, ddlType)
+	}
 	return supportedEventTypes
 }
 


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