Repository: pingcap/tiflow
Pull request #11192 (2024-05-29 06:45:35)

filter(ticdc): Fix verify table rules 
<!--
Thank you for contributing to TiFlow! 
Please read MD's [CONTRIBUTING](https://github.com/pingcap/tiflow/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.
-->

### What problem does this PR solve?
<!--
Please create an issue first to describe the problem.

There MUST be one line starting with "Issue Number:  " and 
linking the relevant issues via the "close" or "ref".

For more info, check https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue.
 -->

Issue Number: ref #10457

### What is changed and how it works?


### Check List <!--REMOVE the items that are not applicable-->

#### Tests <!-- At least one of them must be included. -->

 - No code

#### 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 [hongyunyan](https://github.com/hongyunyan), **ensuring that each of them provides their approval before proceeding**. For more information see [the 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":["hongyunyan"]} -->
ti-chi-bot[bot]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/pingcap/tiflow/pull/11192#pullrequestreview-2084456980" 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-05-29 06:53:59.456637715 +0000 UTC m=+2845793.213773285`: :ballot_box_with_check: agreed by [asddongmen](https://github.com/asddongmen).
ti-chi-bot[bot]: [LGTM Timeline notifier]
---
Timeline:

- `2024-05-29 06:53:59.456637715 +0000 UTC m=+2845793.213773285`: :ballot_box_with_check: agreed by [asddongmen](https://github.com/asddongmen).
- `2024-05-29 07:01:35.192062298 +0000 UTC m=+2846248.949197856`: :ballot_box_with_check: agreed by [CharlesCheung96](https://github.com/CharlesCheung96).
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/pingcap/tiflow/pull/11192?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) Report
Attention: Patch coverage is `50.00000%` with `2 lines` in your changes are missing coverage. Please review.
> Project coverage is 57.6028%. Comparing base [(`c07bce4`)](https://app.codecov.io/gh/pingcap/tiflow/commit/c07bce4e337cc63e5cd20b2f2632076f8cb8eb55?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) to head [(`c878aa5`)](https://app.codecov.io/gh/pingcap/tiflow/commit/c878aa5c645d0c6966e476e18dd23481d702618a?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap).

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


| [Components](https://app.codecov.io/gh/pingcap/tiflow/pull/11192/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/11192/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `61.4107% <50.0000%> (-0.0200%)` | :arrow_down: |
| [dm](https://app.codecov.io/gh/pingcap/tiflow/pull/11192/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `51.1880% <ø> (-0.0242%)` | :arrow_down: |
| [engine](https://app.codecov.io/gh/pingcap/tiflow/pull/11192/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `63.3667% <ø> (-0.0283%)` | :arrow_down: |

| [Flag](https://app.codecov.io/gh/pingcap/tiflow/pull/11192/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/11192/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `57.6028% <50.0000%> (-0.0226%)` | :arrow_down: |

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     #11192        +/-   ##
================================================
- Coverage   57.6254%   57.6028%   -0.0226%     
================================================
  Files           849        849                
  Lines        125934     125933         -1     
================================================
- Hits          72570      72541        -29     
- Misses        47966      47987        +21     
- Partials       5398       5405         +7     
```

</details>
ti-chi-bot[bot]: @hongyunyan: 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-mysql-test | 94e3e6d7aa2c9a856abaa7382b613939b9e74987 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_test/2030/display/redirect) | true | `/test cdc-integration-mysql-test`

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

<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]: @hongyunyan: 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-mysql-test | 94e3e6d7aa2c9a856abaa7382b613939b9e74987 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_test/2030/display/redirect) | true | `/test cdc-integration-mysql-test`
pull-dm-integration-test | 94e3e6d7aa2c9a856abaa7382b613939b9e74987 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_integration_test/2185/display/redirect) | true | `/test dm-integration-test`

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

<details>

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

Patch:
diff --git a/cdc/api/v2/api_helpers.go b/cdc/api/v2/api_helpers.go
index 73f256b26..05b93b6e6 100644
--- a/cdc/api/v2/api_helpers.go
+++ b/cdc/api/v2/api_helpers.go
@@ -56,6 +56,8 @@ const RegisterImportTaskPrefix = "/tidb/brie/import"
 
 // APIV2Helpers is a collections of helper functions of OpenAPIV2.
 // Defining it as an interface to make APIs more testable.
+// Note: Every method in this interface should check the context before returning.
+// If the context is canceled, the method should return an error immediately.
 type APIV2Helpers interface {
 	// verifyCreateChangefeedConfig verifies the changefeedConfig,
 	// and yield a valid changefeedInfo or error
@@ -104,18 +106,22 @@ type APIV2Helpers interface {
 	// getEtcdClient returns an Etcd client given the PD endpoints and
 	// tls config
 	getEtcdClient(
+		ctx context.Context,
 		pdAddrs []string,
 		tlsConfig *tls.Config,
 	) (*clientv3.Client, error)
 
 	// getKVCreateTiStore wraps kv.createTiStore method to increase testability
 	createTiStore(
+		ctx context.Context,
 		pdAddrs []string,
 		credential *security.Credential,
 	) (tidbkv.Storage, error)
 
 	// getVerifiedTables wraps entry.VerifyTables to increase testability
-	getVerifiedTables(replicaConfig *config.ReplicaConfig,
+	getVerifiedTables(
+		ctx context.Context,
+		replicaConfig *config.ReplicaConfig,
 		storage tidbkv.Storage, startTs uint64,
 		scheme string, topic string, protocol config.Protocol,
 	) (ineligibleTables,
@@ -257,7 +263,8 @@ func (APIV2HelpersImpl) verifyCreateChangefeedConfig(
 }
 
 // verifyUpstream verifies the upstream config before updating a changefeed
-func (h APIV2HelpersImpl) verifyUpstream(ctx context.Context,
+func (h APIV2HelpersImpl) verifyUpstream(
+	ctx context.Context,
 	changefeedConfig *ChangefeedConfig,
 	cfInfo *model.ChangeFeedInfo,
 ) error {
@@ -401,7 +408,8 @@ func (APIV2HelpersImpl) verifyUpdateChangefeedConfig(
 // overrideCheckpointTs is the checkpointTs of the changefeed that specified by the user.
 // or it is the checkpointTs of the changefeed before it is paused.
 // we need to check weather the resuming changefeed is gc safe or not.
-func (APIV2HelpersImpl) verifyResumeChangefeedConfig(ctx context.Context,
+func (APIV2HelpersImpl) verifyResumeChangefeedConfig(
+	ctx context.Context,
 	pdClient pd.Client,
 	gcServiceID string,
 	changefeedID model.ChangeFeedID,
@@ -440,7 +448,8 @@ func (APIV2HelpersImpl) verifyResumeChangefeedConfig(ctx context.Context,
 }
 
 // getPDClient returns a PDClient given the PD cluster addresses and a credential
-func (APIV2HelpersImpl) getPDClient(ctx context.Context,
+func (APIV2HelpersImpl) getPDClient(
+	ctx context.Context,
 	pdAddrs []string,
 	credential *security.Credential,
 ) (pd.Client, error) {
@@ -472,7 +481,9 @@ func (APIV2HelpersImpl) getPDClient(ctx context.Context,
 }
 
 func (h APIV2HelpersImpl) getEtcdClient(
-	pdAddrs []string, tlsCfg *tls.Config,
+	ctx context.Context,
+	pdAddrs []string,
+	tlsCfg *tls.Config,
 ) (*clientv3.Client, error) {
 	conf := config.GetGlobalServerConfig()
 	grpcTLSOption, err := conf.Security.ToGRPCDialOption()
@@ -481,7 +492,7 @@ func (h APIV2HelpersImpl) getEtcdClient(
 	}
 	logConfig := &logutil.DefaultZapLoggerConfig
 	logConfig.Level = zap.NewAtomicLevelAt(zapcore.ErrorLevel)
-	return clientv3.New(
+	res, err := clientv3.New(
 		clientv3.Config{
 			Endpoints:   pdAddrs,
 			TLS:         tlsCfg,
@@ -504,16 +515,33 @@ func (h APIV2HelpersImpl) getEtcdClient(
 			},
 		},
 	)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+	if ctx.Err() != nil {
+		return nil, errors.Trace(ctx.Err())
+	}
+	return res, nil
 }
 
 // getTiStore wrap the kv.createTiStore method to increase testability
-func (h APIV2HelpersImpl) createTiStore(pdAddrs []string,
+func (h APIV2HelpersImpl) createTiStore(
+	ctx context.Context,
+	pdAddrs []string,
 	credential *security.Credential,
 ) (tidbkv.Storage, error) {
-	return kv.CreateTiStore(strings.Join(pdAddrs, ","), credential)
+	res, err := kv.CreateTiStore(strings.Join(pdAddrs, ","), credential)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+	if ctx.Err() != nil {
+		return nil, errors.Trace(ctx.Err())
+	}
+	return res, nil
 }
 
 func (h APIV2HelpersImpl) getVerifiedTables(
+	ctx context.Context,
 	replicaConfig *config.ReplicaConfig,
 	storage tidbkv.Storage, startTs uint64,
 	scheme string, topic string, protocol config.Protocol,
@@ -550,5 +578,9 @@ func (h APIV2HelpersImpl) getVerifiedTables(
 		return nil, nil, err
 	}
 
+	if ctx.Err() != nil {
+		return nil, nil, errors.Trace(ctx.Err())
+	}
+
 	return ineligibleTables, eligibleTables, nil
 }
diff --git a/cdc/api/v2/api_helpers_mock.go b/cdc/api/v2/api_helpers_mock.go
index 2ec2f12b1..613d99d10 100644
--- a/cdc/api/v2/api_helpers_mock.go
+++ b/cdc/api/v2/api_helpers_mock.go
@@ -43,33 +43,33 @@ func (m *MockAPIV2Helpers) EXPECT() *MockAPIV2HelpersMockRecorder {
 }
 
 // createTiStore mocks base method.
-func (m *MockAPIV2Helpers) createTiStore(pdAddrs []string, credential *security.Credential) (kv.Storage, error) {
+func (m *MockAPIV2Helpers) createTiStore(ctx context.Context, pdAddrs []string, credential *security.Credential) (kv.Storage, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "createTiStore", pdAddrs, credential)
+	ret := m.ctrl.Call(m, "createTiStore", ctx, pdAddrs, credential)
 	ret0, _ := ret[0].(kv.Storage)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
 
 // createTiStore indicates an expected call of createTiStore.
-func (mr *MockAPIV2HelpersMockRecorder) createTiStore(pdAddrs, credential interface{}) *gomock.Call {
+func (mr *MockAPIV2HelpersMockRecorder) createTiStore(ctx, pdAddrs, credential interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "createTiStore", reflect.TypeOf((*MockAPIV2Helpers)(nil).createTiStore), pdAddrs, credential)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "createTiStore", reflect.TypeOf((*MockAPIV2Helpers)(nil).createTiStore), ctx, pdAddrs, credential)
 }
 
 // getEtcdClient mocks base method.
-func (m *MockAPIV2Helpers) getEtcdClient(pdAddrs []string, tlsConfig *tls.Config) (*v3.Client, error) {
+func (m *MockAPIV2Helpers) getEtcdClient(ctx context.Context, pdAddrs []string, tlsConfig *tls.Config) (*v3.Client, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "getEtcdClient", pdAddrs, tlsConfig)
+	ret := m.ctrl.Call(m, "getEtcdClient", ctx, pdAddrs, tlsConfig)
 	ret0, _ := ret[0].(*v3.Client)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
 
 // getEtcdClient indicates an expected call of getEtcdClient.
-func (mr *MockAPIV2HelpersMockRecorder) getEtcdClient(pdAddrs, tlsConfig interface{}) *gomock.Call {
+func (mr *MockAPIV2HelpersMockRecorder) getEtcdClient(ctx, pdAddrs, tlsConfig interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "getEtcdClient", reflect.TypeOf((*MockAPIV2Helpers)(nil).getEtcdClient), pdAddrs, tlsConfig)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "getEtcdClient", reflect.TypeOf((*MockAPIV2Helpers)(nil).getEtcdClient), ctx, pdAddrs, tlsConfig)
 }
 
 // getPDClient mocks base method.
@@ -88,9 +88,9 @@ func (mr *MockAPIV2HelpersMockRecorder) getPDClient(ctx, pdAddrs, credential int
 }
 
 // getVerifiedTables mocks base method.
-func (m *MockAPIV2Helpers) getVerifiedTables(replicaConfig *config.ReplicaConfig, storage kv.Storage, startTs uint64, scheme, topic string, protocol config.Protocol) ([]model.TableName, []model.TableName, error) {
+func (m *MockAPIV2Helpers) getVerifiedTables(ctx context.Context, replicaConfig *config.ReplicaConfig, storage kv.Storage, startTs uint64, scheme, topic string, protocol config.Protocol) ([]model.TableName, []model.TableName, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "getVerifiedTables", replicaConfig, storage, startTs, scheme, topic, protocol)
+	ret := m.ctrl.Call(m, "getVerifiedTables", ctx, replicaConfig, storage, startTs, scheme, topic, protocol)
 	ret0, _ := ret[0].([]model.TableName)
 	ret1, _ := ret[1].([]model.TableName)
 	ret2, _ := ret[2].(error)
@@ -98,9 +98,9 @@ func (m *MockAPIV2Helpers) getVerifiedTables(replicaConfig *config.ReplicaConfig
 }
 
 // getVerifiedTables indicates an expected call of getVerifiedTables.
-func (mr *MockAPIV2HelpersMockRecorder) getVerifiedTables(replicaConfig, storage, startTs, scheme, topic, protocol interface{}) *gomock.Call {
+func (mr *MockAPIV2HelpersMockRecorder) getVerifiedTables(ctx, replicaConfig, storage, startTs, scheme, topic, protocol interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "getVerifiedTables", reflect.TypeOf((*MockAPIV2Helpers)(nil).getVerifiedTables), replicaConfig, storage, startTs, scheme, topic, protocol)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "getVerifiedTables", reflect.TypeOf((*MockAPIV2Helpers)(nil).getVerifiedTables), ctx, replicaConfig, storage, startTs, scheme, topic, protocol)
 }
 
 // verifyCreateChangefeedConfig mocks base method.
diff --git a/cdc/api/v2/changefeed.go b/cdc/api/v2/changefeed.go
index 7174f1429..68cb5f9f4 100644
--- a/cdc/api/v2/changefeed.go
+++ b/cdc/api/v2/changefeed.go
@@ -86,7 +86,7 @@ func (h *OpenAPIV2) createChangefeed(c *gin.Context) {
 	defer pdClient.Close()
 
 	// verify tables todo: del kvstore
-	kvStorage, err := h.helpers.createTiStore(cfg.PDAddrs, credential)
+	kvStorage, err := h.helpers.createTiStore(ctx, cfg.PDAddrs, credential)
 	if err != nil {
 		_ = c.Error(cerror.WrapError(cerror.ErrNewStore, err))
 		return
@@ -143,7 +143,7 @@ func (h *OpenAPIV2) createChangefeed(c *gin.Context) {
 		return
 	}
 
-	cli, err := h.helpers.getEtcdClient(cfg.PDAddrs, tlsCfg)
+	cli, err := h.helpers.getEtcdClient(ctx, cfg.PDAddrs, tlsCfg)
 	if err != nil {
 		_ = c.Error(err)
 		return
@@ -328,8 +328,8 @@ func (h *OpenAPIV2) verifyTable(c *gin.Context) {
 		cfg.PDConfig = getUpstreamPDConfig(up)
 	}
 	credential := cfg.PDConfig.toCredential()
-
-	kvStore, err := h.helpers.createTiStore(cfg.PDAddrs, credential)
+	ctx := c.Request.Context()
+	kvStore, err := h.helpers.createTiStore(ctx, cfg.PDAddrs, credential)
 	if err != nil {
 		_ = c.Error(err)
 		return
@@ -347,7 +347,7 @@ func (h *OpenAPIV2) verifyTable(c *gin.Context) {
 	protocol, _ := config.ParseSinkProtocolFromString(util.GetOrZero(replicaCfg.Sink.Protocol))
 
 	ineligibleTables, eligibleTables, err := h.helpers.
-		getVerifiedTables(replicaCfg, kvStore, cfg.StartTs, scheme, topic, protocol)
+		getVerifiedTables(ctx, replicaCfg, kvStore, cfg.StartTs, scheme, topic, protocol)
 	if err != nil {
 		_ = c.Error(err)
 		return
@@ -464,7 +464,7 @@ func (h *OpenAPIV2) updateChangefeed(c *gin.Context) {
 	if len(updateCfConfig.PDAddrs) != 0 || upManager == nil {
 		pdAddrs := updateCfConfig.PDAddrs
 		credentials := updateCfConfig.PDConfig.toCredential()
-		storage, err = h.helpers.createTiStore(pdAddrs, credentials)
+		storage, err = h.helpers.createTiStore(ctx, pdAddrs, credentials)
 		if err != nil {
 			_ = c.Error(errors.Trace(err))
 		}
diff --git a/metrics/grafana/ticdc.json b/metrics/grafana/ticdc.json
index bcc960682..0f731ef73 100644
--- a/metrics/grafana/ticdc.json
+++ b/metrics/grafana/ticdc.json
@@ -7465,11 +7465,11 @@
           "targets": [
             {
               "exemplar": true,
-              "expr": "sum(ticdc_sink_kafka_producer_request_latency{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", namespace=~\"$namespace\",changefeed=~\"$changefeed\", instance=~\"$ticdc_instance\"}) by (namespace,changefeed, instance, broker)",
+              "expr": "sum(ticdc_sink_kafka_producer_request_latency{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", namespace=~\"$namespace\",changefeed=~\"$changefeed\", instance=~\"$ticdc_instance\"}) by (namespace,changefeed, instance, broker, type)",
               "format": "time_series",
               "interval": "",
               "intervalFactor": 1,
-              "legendFormat": "{{namespace}}-{{changefeed}}-{{instance}}-{{broker}}",
+              "legendFormat": "{{namespace}}-{{changefeed}}-{{instance}}-{{broker}}-{{type}}",
               "refId": "A"
             }
           ],
@@ -7671,11 +7671,11 @@
           "targets": [
             {
               "exemplar": true,
-              "expr": "sum(ticdc_sink_kafka_producer_records_per_request{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", namespace=~\"$namespace\",changefeed=~\"$changefeed\", instance=~\"$ticdc_instance\"}) by (namespace,changefeed, instance, broker)",
+              "expr": "sum(ticdc_sink_kafka_producer_records_per_request{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", namespace=~\"$namespace\",changefeed=~\"$changefeed\", instance=~\"$ticdc_instance\"}) by (namespace,changefeed, instance, type)",
               "format": "time_series",
               "interval": "",
               "intervalFactor": 1,
-              "legendFormat": "{{namespace}}-{{changefeed}}-{{instance}}-{{broker}}",
+              "legendFormat": "{{namespace}}-{{changefeed}}-{{instance}}-{{type}}",
               "refId": "A"
             }
           ],
@@ -7772,20 +7772,12 @@
           "targets": [
             {
               "exemplar": true,
-              "expr": "sum(ticdc_sink_kafka_producer_compression_ratio{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", namespace=~\"$namespace\",changefeed=~\"$changefeed\", instance=~\"$ticdc_instance\"}) by (namespace,changefeed, instance)",
+              "expr": "sum(ticdc_sink_kafka_producer_compression_ratio{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", namespace=~\"$namespace\",changefeed=~\"$changefeed\", instance=~\"$ticdc_instance\"}) by (namespace,changefeed, instance, type)",
               "format": "time_series",
               "interval": "",
               "intervalFactor": 1,
-              "legendFormat": "{{namespace}}-{{changefeed}}-{{instance}}",
+              "legendFormat": "{{namespace}}-{{changefeed}}-{{instance}}-{{type}}",
               "refId": "A"
-            },
-            {
-              "exemplar": true,
-              "expr": "histogram_quantile(0.99, sum(rate(ticdc_kafka_sink_codec_compression_ratio_bucket{k8s_cluster=\"$k8s_cluster\",tidb_cluster=\"$tidb_cluster\",namespace=~\"$namespace\",changefeed=~\"$changefeed\",instance=~\"$ticdc_instance\"}[1m])) by (le,namespace,changefeed,instance))",
-              "hide": false,
-              "interval": "",
-              "legendFormat": "{{namespace}}-{{changefeed}}-{{instance}}-Codec-P99",
-              "refId": "B"
             }
           ],
           "thresholds": [],
diff --git a/pkg/filter/utils.go b/pkg/filter/utils.go
index be42603e6..a844e062e 100644
--- a/pkg/filter/utils.go
+++ b/pkg/filter/utils.go
@@ -40,15 +40,11 @@ func isSysSchema(db string) bool {
 // and returns an invalid rule error if the verification fails,
 // otherwise it will return a table filter.
 func VerifyTableRules(cfg *config.FilterConfig) (tfilter.Filter, error) {
-	var f tfilter.Filter
-	var err error
-	if len(cfg.Rules) != 0 {
-		rules := cfg.Rules
-		if len(rules) == 0 {
-			rules = []string{"*.*"}
-		}
-		f, err = tfilter.Parse(rules)
+	rules := cfg.Rules
+	if len(rules) == 0 {
+		rules = []string{"*.*"}
 	}
+	f, err := tfilter.Parse(rules)
 	if err != nil {
 		return nil, cerror.WrapError(cerror.ErrFilterRuleInvalid, err, cfg)
 	}
diff --git a/pkg/sink/kafka/metrics.go b/pkg/sink/kafka/metrics.go
index b78e6a699..68d7095b3 100644
--- a/pkg/sink/kafka/metrics.go
+++ b/pkg/sink/kafka/metrics.go
@@ -49,7 +49,7 @@ var (
 			Subsystem: "sink",
 			Name:      "kafka_producer_request_latency",
 			Help:      "The request latency for all brokers.",
-		}, []string{"namespace", "changefeed", "broker"})
+		}, []string{"namespace", "changefeed", "broker", "type"})
 	// Histogram update by `compression-ratio`.
 	compressionRatioGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
@@ -57,8 +57,7 @@ var (
 			Subsystem: "sink",
 			Name:      "kafka_producer_compression_ratio",
 			Help:      "The compression ratio times 100 of record batches for all topics.",
-		}, []string{"namespace", "changefeed"})
-
+		}, []string{"namespace", "changefeed", "type"})
 	// updated by `records-per-request`.
 	recordsPerRequestGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
@@ -66,7 +65,7 @@ var (
 			Subsystem: "sink",
 			Name:      "kafka_producer_records_per_request",
 			Help:      "The P99 number of records per request for all topics.",
-		}, []string{"namespace", "changefeed"})
+		}, []string{"namespace", "changefeed", "type"})
 
 	// Meter mark by 1 once a response received.
 	responseRateGauge = prometheus.NewGaugeVec(
diff --git a/pkg/sink/kafka/metrics_collector.go b/pkg/sink/kafka/metrics_collector.go
index ff3c1c82b..4eced9a88 100644
--- a/pkg/sink/kafka/metrics_collector.go
+++ b/pkg/sink/kafka/metrics_collector.go
@@ -51,6 +51,8 @@ const (
 	requestLatencyInMsMetricNamePrefix = "request-latency-in-ms-for-broker-"
 	requestsInFlightMetricNamePrefix   = "requests-in-flight-for-broker-"
 	responseRateMetricNamePrefix       = "response-rate-for-broker-"
+
+	p99 = "p99"
 )
 
 type saramaMetricsCollector struct {
@@ -128,19 +130,23 @@ func (m *saramaMetricsCollector) updateBrokers(ctx context.Context) {
 func (m *saramaMetricsCollector) collectProducerMetrics() {
 	namespace := m.changefeedID.Namespace
 	changefeedID := m.changefeedID.ID
-
 	compressionRatioMetric := m.registry.Get(compressionRatioMetricName)
 	if histogram, ok := compressionRatioMetric.(metrics.Histogram); ok {
 		compressionRatioGauge.
-			WithLabelValues(namespace, changefeedID).
+			WithLabelValues(namespace, changefeedID, "avg").
 			Set(histogram.Snapshot().Mean() / 100)
+		compressionRatioGauge.WithLabelValues(namespace, changefeedID, p99).
+			Set(histogram.Snapshot().Percentile(0.99) / 100)
 	}
 
 	recordsPerRequestMetric := m.registry.Get(recordsPerRequestMetricName)
 	if histogram, ok := recordsPerRequestMetric.(metrics.Histogram); ok {
 		recordsPerRequestGauge.
-			WithLabelValues(namespace, changefeedID).
+			WithLabelValues(namespace, changefeedID, "avg").
 			Set(histogram.Snapshot().Mean())
+		recordsPerRequestGauge.
+			WithLabelValues(namespace, changefeedID, p99).
+			Set(histogram.Snapshot().Percentile(0.99))
 	}
 }
 
@@ -149,13 +155,12 @@ func (m *saramaMetricsCollector) collectBrokerMetrics() {
 	changefeedID := m.changefeedID.ID
 	for id := range m.brokers {
 		brokerID := strconv.Itoa(int(id))
-
 		outgoingByteRateMetric := m.registry.Get(
 			getBrokerMetricName(outgoingByteRateMetricNamePrefix, brokerID))
 		if meter, ok := outgoingByteRateMetric.(metrics.Meter); ok {
 			OutgoingByteRateGauge.
 				WithLabelValues(namespace, changefeedID, brokerID).
-				Set(meter.Snapshot().RateMean())
+				Set(meter.Snapshot().Rate1())
 		}
 
 		requestRateMetric := m.registry.Get(
@@ -163,15 +168,18 @@ func (m *saramaMetricsCollector) collectBrokerMetrics() {
 		if meter, ok := requestRateMetric.(metrics.Meter); ok {
 			RequestRateGauge.
 				WithLabelValues(namespace, changefeedID, brokerID).
-				Set(meter.Snapshot().RateMean())
+				Set(meter.Snapshot().Rate1())
 		}
 
 		requestLatencyMetric := m.registry.Get(
 			getBrokerMetricName(requestLatencyInMsMetricNamePrefix, brokerID))
 		if histogram, ok := requestLatencyMetric.(metrics.Histogram); ok {
 			RequestLatencyGauge.
-				WithLabelValues(namespace, changefeedID, brokerID).
-				Set(histogram.Snapshot().Mean() / 1000) // convert millisecond to second.
+				WithLabelValues(namespace, changefeedID, brokerID, "avg").
+				Set(histogram.Snapshot().Mean() / 1000)
+			RequestLatencyGauge.
+				WithLabelValues(namespace, changefeedID, brokerID, p99).
+				Set(histogram.Snapshot().Percentile(0.99) / 1000)
 		}
 
 		requestsInFlightMetric := m.registry.Get(getBrokerMetricName(
@@ -187,7 +195,7 @@ func (m *saramaMetricsCollector) collectBrokerMetrics() {
 		if meter, ok := responseRateMetric.(metrics.Meter); ok {
 			responseRateGauge.
 				WithLabelValues(namespace, changefeedID, brokerID).
-				Set(meter.Snapshot().RateMean())
+				Set(meter.Snapshot().Rate1())
 		}
 	}
 }
diff --git a/pkg/sink/kafka/v2/metrics_collector.go b/pkg/sink/kafka/v2/metrics_collector.go
index dd4286b5b..70eee28b8 100644
--- a/pkg/sink/kafka/v2/metrics_collector.go
+++ b/pkg/sink/kafka/v2/metrics_collector.go
@@ -81,7 +81,7 @@ func (m *MetricsCollector) collectMetrics() {
 	// since kafka-go does not support per broker metrics, so we add `v2` as the broker ID.
 	pkafka.RequestRateGauge.WithLabelValues(m.changefeedID.Namespace, m.changefeedID.ID, "v2").
 		Set(float64(statistics.Writes / 5))
-	pkafka.RequestLatencyGauge.WithLabelValues(m.changefeedID.Namespace, m.changefeedID.ID, "v2").
+	pkafka.RequestLatencyGauge.WithLabelValues(m.changefeedID.Namespace, m.changefeedID.ID, "v2", "avg").
 		Set(statistics.WriteTime.Avg.Seconds())
 	pkafka.OutgoingByteRateGauge.WithLabelValues(m.changefeedID.Namespace, m.changefeedID.ID, "v2").
 		Set(float64(statistics.Bytes / 5))
@@ -103,7 +103,7 @@ func (m *MetricsCollector) cleanupMetrics() {
 	pkafka.RequestRateGauge.
 		DeleteLabelValues(m.changefeedID.Namespace, m.changefeedID.ID, "v2")
 	pkafka.RequestLatencyGauge.
-		DeleteLabelValues(m.changefeedID.Namespace, m.changefeedID.ID, "v2")
+		DeleteLabelValues(m.changefeedID.Namespace, m.changefeedID.ID, "v2", "avg")
 	pkafka.OutgoingByteRateGauge.
 		DeleteLabelValues(m.changefeedID.Namespace, m.changefeedID.ID, "v2")
 


Test patch:
diff --git a/cdc/api/v2/changefeed_test.go b/cdc/api/v2/changefeed_test.go
index 3fbe004b0..b8295cf40 100644
--- a/cdc/api/v2/changefeed_test.go
+++ b/cdc/api/v2/changefeed_test.go
@@ -135,7 +135,7 @@ func TestCreateChangefeed(t *testing.T) {
 		getPDClient(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(pdClient, nil).AnyTimes()
 	helpers.EXPECT().
-		createTiStore(gomock.Any(), gomock.Any()).
+		createTiStore(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil, cerrors.ErrNewStore).
 		Times(1)
 	cfConfig.PDAddrs = []string{"http://127.0.0.1:2379", "http://127.0.0.1:2382"}
@@ -153,7 +153,7 @@ func TestCreateChangefeed(t *testing.T) {
 
 	// case 4: failed to verify tables
 	helpers.EXPECT().
-		createTiStore(gomock.Any(), gomock.Any()).
+		createTiStore(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil, nil).
 		AnyTimes()
 	helpers.EXPECT().
@@ -177,9 +177,9 @@ func TestCreateChangefeed(t *testing.T) {
 
 	// case 5:
 	helpers.EXPECT().
-		getEtcdClient(gomock.Any(), gomock.Any()).
+		getEtcdClient(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(testEtcdCluster.RandClient(), nil)
-	helpers.EXPECT().getVerifiedTables(gomock.Any(), gomock.Any(), gomock.Any(),
+	helpers.EXPECT().getVerifiedTables(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
 		gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil, nil, nil).
 		AnyTimes()
@@ -222,7 +222,7 @@ func TestCreateChangefeed(t *testing.T) {
 
 	// case 6: success
 	helpers.EXPECT().
-		getEtcdClient(gomock.Any(), gomock.Any()).
+		getEtcdClient(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(testEtcdCluster.RandClient(), nil)
 	mo.EXPECT().
 		CreateChangefeed(gomock.Any(), gomock.Any(), gomock.Any()).
@@ -446,7 +446,7 @@ func TestUpdateChangefeed(t *testing.T) {
 		verifyUpstream(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil).AnyTimes()
 	helpers.EXPECT().
-		createTiStore(gomock.Any(), gomock.Any()).
+		createTiStore(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil, nil).
 		AnyTimes()
 	mockCapture.EXPECT().GetUpstreamManager().Return(nil, nil).AnyTimes()
@@ -650,7 +650,7 @@ func TestVerifyTable(t *testing.T) {
 	body, err := json.Marshal(&updateCfg)
 	require.Nil(t, err)
 	helpers.EXPECT().
-		createTiStore(gomock.Any(), gomock.Any()).
+		createTiStore(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil, cerrors.ErrNewStore).
 		Times(1)
 
@@ -665,10 +665,10 @@ func TestVerifyTable(t *testing.T) {
 
 	// case 3: getVerifiedTables failed
 	helpers.EXPECT().
-		createTiStore(gomock.Any(), gomock.Any()).
+		createTiStore(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil, nil).
 		AnyTimes()
-	helpers.EXPECT().getVerifiedTables(gomock.Any(), gomock.Any(), gomock.Any(),
+	helpers.EXPECT().getVerifiedTables(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
 		gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil, nil, cerrors.ErrFilterRuleInvalid).
 		Times(1)
@@ -690,7 +690,7 @@ func TestVerifyTable(t *testing.T) {
 	ineligible := []model.TableName{
 		{Schema: "test", Table: "invalidTable"},
 	}
-	helpers.EXPECT().getVerifiedTables(gomock.Any(), gomock.Any(), gomock.Any(),
+	helpers.EXPECT().getVerifiedTables(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
 		gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(eligible, ineligible, nil)
 
@@ -702,6 +702,21 @@ func TestVerifyTable(t *testing.T) {
 	err = json.NewDecoder(w.Body).Decode(&resp)
 	require.Nil(t, err)
 	require.Equal(t, http.StatusOK, w.Code)
+
+	// case 5: context canceled
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	helpers.EXPECT().getVerifiedTables(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
+		gomock.Any(), gomock.Any(), gomock.Any()).
+		Return(nil, nil, ctx.Err()).
+		Times(1)
+	w = httptest.NewRecorder()
+	req, _ = http.NewRequestWithContext(ctx, verify.method, verify.url, bytes.NewReader(body))
+	router.ServeHTTP(w, req)
+	respErr = model.HTTPError{}
+	err = json.NewDecoder(w.Body).Decode(&respErr)
+	require.Nil(t, err)
+	require.Contains(t, respErr.Error, "context canceled")
 }
 
 func TestResumeChangefeed(t *testing.T) {
diff --git a/pkg/filter/utils_test.go b/pkg/filter/utils_test.go
index b3f926d52..3a1cb0f57 100644
--- a/pkg/filter/utils_test.go
+++ b/pkg/filter/utils_test.go
@@ -48,6 +48,7 @@ func TestVerifyTableRules(t *testing.T) {
 		cfg      *config.FilterConfig
 		hasError bool
 	}{
+		{&config.FilterConfig{}, false},
 		{&config.FilterConfig{Rules: []string{""}}, false},
 		{&config.FilterConfig{Rules: []string{"*.*"}}, false},
 		{&config.FilterConfig{Rules: []string{"test.*ms"}}, false},
@@ -56,7 +57,10 @@ func TestVerifyTableRules(t *testing.T) {
 		{&config.FilterConfig{Rules: []string{"*.*", "*.*.*", "*.*.*.*"}}, true},
 	}
 	for _, c := range cases {
-		_, err := VerifyTableRules(c.cfg)
+		f, err := VerifyTableRules(c.cfg)
 		require.Equal(t, c.hasError, err != nil, "case: %s", c.cfg.Rules)
+		if !c.hasError {
+			require.True(t, f != nil)
+		}
 	}
 }
diff --git a/pkg/sink/kafka/v2/metrics_collector_test.go b/pkg/sink/kafka/v2/metrics_collector_test.go
index 6f0e48161..46fb3cb49 100644
--- a/pkg/sink/kafka/v2/metrics_collector_test.go
+++ b/pkg/sink/kafka/v2/metrics_collector_test.go
@@ -40,7 +40,7 @@ func TestMetricsCollector(t *testing.T) {
 		{[]string{changefeedID.Namespace, changefeedID.ID}, pkafka.BatchMessageCountGauge},
 		{[]string{changefeedID.Namespace, changefeedID.ID}, pkafka.BatchSizeGauge},
 		{[]string{changefeedID.Namespace, changefeedID.ID, "v2"}, pkafka.RequestRateGauge},
-		{[]string{changefeedID.Namespace, changefeedID.ID, "v2"}, pkafka.RequestLatencyGauge},
+		{[]string{changefeedID.Namespace, changefeedID.ID, "v2", "avg"}, pkafka.RequestLatencyGauge},
 		{[]string{changefeedID.Namespace, changefeedID.ID, "v2"}, pkafka.OutgoingByteRateGauge},
 		{[]string{changefeedID.Namespace, changefeedID.ID}, pkafka.ClientRetryGauge},
 		{[]string{changefeedID.Namespace, changefeedID.ID}, pkafka.ClientErrorGauge},