Repository: pingcap/tiflow
Pull request #5542 (2022-05-23 14:39:17)

Rollup PR #5499, #5269, #5558, #4643, #4375, #4235 and #5477
This is an automated cherry-pick of #5477

### What problem does this PR solve?

Issue Number: close #5468 

### What is changed and how it works?

* config(ticdc): use 1 minute kv retry time (#5499)
* cdc,retry: fix leader missing by extending region retry duration (#5269)
* kv(ticdc): set debug level for resolved ts fallback log (#5558)
* kv,cmd(ticdc): refinement about resolve lock (#4643)
* metrics(ticdc): fix min resolved ts/checkpoint table ID metrics (#4375)
* cdc,pkg,tests(ticdc): unify cdc log style (#4235)
* kv(ticdc): fix data loss when upstream txn conflicts during scan (#5477)

### Check List <!--REMOVE the items that are not applicable-->

#### Tests <!-- At least one of them must be included. -->

 - Unit test

##### Will it cause performance regression or break compatibility?

It caches more events in memory during incremental scan, it cause OOM when upstream workload is write-heavy.

##### Do you need to update user documentation, design documentation or monitoring documentation?

No.

### Release note <!-- bugfixes or new feature need a release note -->

```release-note
Fix data loss when upstream transaction conflicts during cdc reconnection
Fix replication interruption due to leader missing by extending region retry duration
Fix min resolved ts/checkpoint table ID metrics
```


ti-chi-bot: [REVIEW NOTIFICATION]

This pull request has not been approved.


To complete the [pull request process](https://book.prow.tidb.io/#/en/workflows/pr), please ask the reviewers in the [list](https://prow.tidb.io/tichi/repos/pingcap/tiflow/pulls/5542/owners) to review by filling `/cc @reviewer` in the comment.
After your PR has acquired the required number of LGTMs, you can assign this pull request to the committer in the [list](https://prow.tidb.io/tichi/repos/pingcap/tiflow/pulls/5542/owners) by filling  `/assign @committer` in the comment to help you merge this pull request.

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

<details>

Reviewer can indicate their review by submitting an approval review.
Reviewer can cancel approval by submitting a request changes review.
</details>

<!--Review Notification Identifier-->
ti-chi-bot: @ti-chi-bot: This cherry pick PR is for a release branch and has not yet been approved by release team.
Adding the `do-not-merge/cherry-pick-not-approved` label.

To merge this cherry pick, it must first be approved by the collaborators.

**AFTER** it has been approved by collaborators, please ping the release team in a comment to request a cherry pick review.


<details>

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

overvenus: /merge
ti-chi-bot: This pull request has been accepted and is ready to merge. <details>Commit hash: 29a08173d8cc168fe7717737c699ed536e4c6ab6</details>
ti-chi-bot: @ti-chi-bot: Your PR was out of date, I have automatically updated it for you.

At the same time I will also trigger all tests for you:

/run-all-tests

If the CI test fails, you just re-trigger the test that failed and the bot will merge the PR for you after the CI passes.


<details>

Instructions for interacting with me using PR comments are available [here](https://prow.tidb.io/command-help).  If you have questions or suggestions related to my behavior, please file an issue against the [ti-community-infra/tichi](https://github.com/ti-community-infra/tichi/issues/new?title=Prow%20issue:) repository.
</details>
ti-chi-bot: Merge canceled because a new commit is pushed.
overvenus: /merge
ti-chi-bot: This pull request has been accepted and is ready to merge. <details>Commit hash: 709a65c8cdab87e19a274478061fbe81d2993920</details>

Patch:
diff --git a/Makefile b/Makefile
index 96ca66bfb..1e515ed8d 100644
--- a/Makefile
+++ b/Makefile
@@ -172,6 +172,8 @@ fmt: tools/bin/gofumports tools/bin/shfmt
 	tools/bin/gofumports -l -w $(FILES) 2>&1 | $(FAIL_ON_STDOUT)
 	@echo "run shfmt"
 	tools/bin/shfmt -d -w .
+	@echo "check log style"
+	scripts/check-log-style.sh
 
 errdoc: tools/bin/errdoc-gen
 	@echo "generator errors.toml"
diff --git a/cdc/capture/capture.go b/cdc/capture/capture.go
index 78161a0a7..61c835177 100644
--- a/cdc/capture/capture.go
+++ b/cdc/capture/capture.go
@@ -218,8 +218,8 @@ func (c *Capture) reset(ctx context.Context) error {
 	}
 
 	log.Info("init capture",
-		zap.String("capture-id", c.info.ID),
-		zap.String("capture-addr", c.info.AdvertiseAddr))
+		zap.String("captureID", c.info.ID),
+		zap.String("captureAddr", c.info.AdvertiseAddr))
 	return nil
 }
 
@@ -254,7 +254,7 @@ func (c *Capture) Run(ctx context.Context) error {
 		//   2. the parent context canceled, it means that the caller of the capture hope the capture to exit, and this loop will return in the above `select` block
 		// TODO: make sure the internal cancel should return the real error instead of context.Canceled
 		if cerror.ErrCaptureSuicide.Equal(err) || context.Canceled == errors.Cause(err) {
-			log.Info("capture recovered", zap.String("capture-id", c.info.ID))
+			log.Info("capture recovered", zap.String("captureID", c.info.ID))
 			continue
 		}
 		return errors.Trace(err)
@@ -422,8 +422,8 @@ func (c *Capture) campaignOwner(ctx cdcContext.Context) error {
 		}
 
 		log.Info("campaign owner successfully",
-			zap.String("capture-id", c.info.ID),
-			zap.Int64("owner-rev", ownerRev))
+			zap.String("captureID", c.info.ID),
+			zap.Int64("ownerRev", ownerRev))
 
 		owner := c.newOwner(c.pdClient)
 		c.setOwner(owner)
diff --git a/cdc/entry/mounter.go b/cdc/entry/mounter.go
index 546b8dda6..7409a0db0 100644
--- a/cdc/entry/mounter.go
+++ b/cdc/entry/mounter.go
@@ -179,6 +179,9 @@ func (m *mounterImpl) unmarshalAndMountRowChanged(ctx context.Context, raw *mode
 	if err != nil {
 		return nil, err
 	}
+	if len(raw.OldValue) == 0 && len(raw.Value) == 0 {
+		log.Warn("empty value and old value", zap.Any("row", raw))
+	}
 	baseInfo := baseKVEntry{
 		StartTs:         raw.StartTs,
 		CRTs:            raw.CRTs,
@@ -538,7 +541,7 @@ func getDefaultOrZeroValue(col *timodel.ColumnInfo) (interface{}, int, string, e
 		default:
 			d = table.GetZeroValue(col)
 			if d.IsNull() {
-				log.Error("meet unsupported column type", zap.String("column info", col.String()))
+				log.Error("meet unsupported column type", zap.String("columnInfo", col.String()))
 			}
 		}
 	}
diff --git a/cdc/entry/schema_storage.go b/cdc/entry/schema_storage.go
index d99a1956b..1aa0d3fe3 100644
--- a/cdc/entry/schema_storage.go
+++ b/cdc/entry/schema_storage.go
@@ -795,7 +795,7 @@ func (s *schemaStorageImpl) GetSnapshot(ctx context.Context, ts uint64) (*schema
 			logTime = now
 		}
 		return err
-	}, retry.WithBackoffBaseDelay(10), retry.WithInfiniteTries(), retry.WithIsRetryableErr(isRetryable))
+	}, retry.WithBackoffBaseDelay(10), retry.WithIsRetryableErr(isRetryable))
 
 	return snap, err
 }
diff --git a/cdc/http_router.go b/cdc/http_router.go
index 9138b97b2..0efb7e03a 100644
--- a/cdc/http_router.go
+++ b/cdc/http_router.go
@@ -153,7 +153,7 @@ func logMiddleware() gin.HandlerFunc {
 			zap.String("ip", c.ClientIP()),
 			zap.String("user-agent", c.Request.UserAgent()),
 			zap.Error(stdErr),
-			zap.Duration("cost", cost),
+			zap.Duration("duration", cost),
 		)
 	}
 }
diff --git a/cdc/kv/client.go b/cdc/kv/client.go
index 4dbab5480..0edc10add 100644
--- a/cdc/kv/client.go
+++ b/cdc/kv/client.go
@@ -293,7 +293,6 @@ type CDCKVClient interface {
 		ctx context.Context,
 		span regionspan.ComparableSpan,
 		ts uint64,
-		enableOldValue bool,
 		lockResolver txnutil.LockResolver,
 		isPullerInit PullerInitialization,
 		eventCh chan<- model.RegionFeedEvent,
@@ -307,6 +306,7 @@ var NewCDCKVClient = NewCDCClient
 type CDCClient struct {
 	pd pd.Client
 
+	config    *config.KVClientConfig
 	clusterID uint64
 
 	grpcPool GrpcPool
@@ -318,7 +318,14 @@ type CDCClient struct {
 }
 
 // NewCDCClient creates a CDCClient instance
-func NewCDCClient(ctx context.Context, pd pd.Client, kvStorage tikv.Storage, grpcPool GrpcPool, regionCache *tikv.RegionCache) (c CDCKVClient) {
+func NewCDCClient(
+	ctx context.Context,
+	pd pd.Client,
+	kvStorage tikv.Storage,
+	grpcPool GrpcPool,
+	regionCache *tikv.RegionCache,
+	cfg *config.KVClientConfig,
+) (c CDCKVClient) {
 	clusterID := pd.GetClusterID(ctx)
 
 	var store TiKVStorage
@@ -333,6 +340,7 @@ func NewCDCClient(ctx context.Context, pd pd.Client, kvStorage tikv.Storage, grp
 
 	c = &CDCClient{
 		clusterID:      clusterID,
+		config:         cfg,
 		pd:             pd,
 		kvStorage:      store,
 		grpcPool:       grpcPool,
@@ -393,14 +401,12 @@ type PullerInitialization interface {
 // The `Start` and `End` field in input span must be memcomparable encoded.
 func (c *CDCClient) EventFeed(
 	ctx context.Context, span regionspan.ComparableSpan, ts uint64,
-	enableOldValue bool,
 	lockResolver txnutil.LockResolver,
 	isPullerInit PullerInitialization,
 	eventCh chan<- model.RegionFeedEvent,
 ) error {
 	s := newEventFeedSession(ctx, c, c.regionCache, c.kvStorage, span,
-		lockResolver, isPullerInit,
-		enableOldValue, ts, eventCh)
+		lockResolver, isPullerInit, ts, eventCh)
 	return s.eventFeed(ctx, ts)
 }
 
@@ -441,8 +447,7 @@ type eventFeedSession struct {
 	// The queue is used to store region that reaches limit
 	rateLimitQueue []regionErrorInfo
 
-	rangeLock      *regionspan.RegionRangeLock
-	enableOldValue bool
+	rangeLock *regionspan.RegionRangeLock
 
 	// To identify metrics of different eventFeedSession
 	id                string
@@ -468,25 +473,22 @@ func newEventFeedSession(
 	totalSpan regionspan.ComparableSpan,
 	lockResolver txnutil.LockResolver,
 	isPullerInit PullerInitialization,
-	enableOldValue bool,
 	startTs uint64,
 	eventCh chan<- model.RegionFeedEvent,
 ) *eventFeedSession {
 	id := strconv.FormatUint(allocID(), 10)
-	kvClientCfg := config.GetGlobalServerConfig().KVClient
 	return &eventFeedSession{
 		client:            client,
 		regionCache:       regionCache,
 		kvStorage:         kvStorage,
 		totalSpan:         totalSpan,
 		eventCh:           eventCh,
-		regionRouter:      NewSizedRegionRouter(ctx, kvClientCfg.RegionScanLimit),
+		regionRouter:      NewSizedRegionRouter(ctx, client.config.RegionScanLimit),
 		regionCh:          make(chan singleRegionInfo, defaultRegionChanSize),
 		errCh:             make(chan regionErrorInfo, defaultRegionChanSize),
 		requestRangeCh:    make(chan rangeRequestTask, defaultRegionChanSize),
 		rateLimitQueue:    make([]regionErrorInfo, 0, defaultRegionRateLimitQueueSize),
 		rangeLock:         regionspan.NewRegionRangeLock(totalSpan.Start, totalSpan.End, startTs),
-		enableOldValue:    enableOldValue,
 		lockResolver:      lockResolver,
 		isPullerInit:      isPullerInit,
 		id:                id,
@@ -693,11 +695,8 @@ func (s *eventFeedSession) requestRegionToStore(
 		}
 		requestID := allocID()
 
-		extraOp := kvrpcpb.ExtraOp_Noop
-		if s.enableOldValue {
-			extraOp = kvrpcpb.ExtraOp_ReadOldValue
-		}
-
+		// Always read old value.
+		extraOp := kvrpcpb.ExtraOp_ReadOldValue
 		rpcCtx := sri.rpcCtx
 		regionID := rpcCtx.Meta.GetId()
 		req := &cdcpb.ChangeDataRequest{
@@ -897,6 +896,8 @@ func (s *eventFeedSession) divideAndSendEventFeedToRegions(
 	nextSpan := span
 	captureAddr := util.CaptureAddrFromCtx(ctx)
 
+	// Max backoff 500ms.
+	scanRegionMaxBackoff := int64(500)
 	for {
 		var (
 			regions []*tikv.Region
@@ -926,7 +927,8 @@ func (s *eventFeedSession) divideAndSendEventFeedToRegions(
 			}
 			log.Debug("ScanRegions", zap.Stringer("span", nextSpan), zap.Reflect("regions", metas))
 			return nil
-		}, retry.WithBackoffMaxDelay(50), retry.WithMaxTries(100), retry.WithIsRetryableErr(cerror.IsRetryableError))
+		}, retry.WithBackoffMaxDelay(scanRegionMaxBackoff),
+			retry.WithTotalRetryDuratoin(time.Duration(s.client.config.RegionRetryDuration)))
 		if retryErr != nil {
 			return retryErr
 		}
@@ -1105,7 +1107,7 @@ func (s *eventFeedSession) receiveFromStream(
 	changefeedID := util.ChangefeedIDFromCtx(ctx)
 	metricSendEventBatchResolvedSize := batchResolvedEventSize.WithLabelValues(captureAddr, changefeedID)
 
-	// always create a new region worker, because `receiveFromStreamV2` is ensured
+	// always create a new region worker, because `receiveFromStream` is ensured
 	// to call exactly once from outter code logic
 	worker := newRegionWorker(s, addr)
 
@@ -1180,7 +1182,7 @@ func (s *eventFeedSession) receiveFromStream(
 				regionCount = len(cevent.ResolvedTs.Regions)
 			}
 			log.Warn("change data event size too large",
-				zap.Int("size", size), zap.Int("event length", len(cevent.Events)),
+				zap.Int("size", size), zap.Int("eventLen", len(cevent.Events)),
 				zap.Int("resolved region count", regionCount))
 		}
 
@@ -1330,7 +1332,7 @@ func (s *eventFeedSession) getStreamCancel(storeAddr string) (cancel context.Can
 	return
 }
 
-func assembleRowEvent(regionID uint64, entry *cdcpb.Event_Row, enableOldValue bool) (model.RegionFeedEvent, error) {
+func assembleRowEvent(regionID uint64, entry *cdcpb.Event_Row) (model.RegionFeedEvent, error) {
 	var opType model.OpType
 	switch entry.GetOpType() {
 	case cdcpb.Event_Row_DELETE:
@@ -1350,14 +1352,10 @@ func assembleRowEvent(regionID uint64, entry *cdcpb.Event_Row, enableOldValue bo
 			StartTs:  entry.StartTs,
 			CRTs:     entry.CommitTs,
 			RegionID: regionID,
+			OldValue: entry.GetOldValue(),
 		},
 	}
 
-	// when old-value is disabled, it is still possible for the tikv to send a event containing the old value
-	// we need avoid a old-value sent to downstream when old-value is disabled
-	if enableOldValue {
-		revent.Val.OldValue = entry.GetOldValue()
-	}
 	return revent, nil
 }
 
diff --git a/cdc/kv/matcher.go b/cdc/kv/matcher.go
index 4643f496f..530e7ac25 100644
--- a/cdc/kv/matcher.go
+++ b/cdc/kv/matcher.go
@@ -23,6 +23,7 @@ type matcher struct {
 	// TODO : clear the single prewrite
 	unmatchedValue map[matchKey]*cdcpb.Event_Row
 	cachedCommit   []*cdcpb.Event_Row
+	cachedRollback []*cdcpb.Event_Row
 }
 
 type matchKey struct {
@@ -58,8 +59,15 @@ func (m *matcher) putPrewriteRow(row *cdcpb.Event_Row) {
 
 // matchRow matches the commit event with the cached prewrite event
 // the Value and OldValue will be assigned if a matched prewrite event exists.
-func (m *matcher) matchRow(row *cdcpb.Event_Row) bool {
+func (m *matcher) matchRow(row *cdcpb.Event_Row, initialized bool) bool {
 	if value, exist := m.unmatchedValue[newMatchKey(row)]; exist {
+		// TiKV may send a fake prewrite event with empty value caused by txn heartbeat.
+		//
+		// We need to skip match if the region is not initialized,
+		// as prewrite events may be sent out of order.
+		if !initialized && len(value.GetValue()) == 0 {
+			return false
+		}
 		row.Value = value.GetValue()
 		row.OldValue = value.GetOldValue()
 		delete(m.unmatchedValue, newMatchKey(row))
@@ -72,13 +80,16 @@ func (m *matcher) cacheCommitRow(row *cdcpb.Event_Row) {
 	m.cachedCommit = append(m.cachedCommit, row)
 }
 
-func (m *matcher) matchCachedRow() []*cdcpb.Event_Row {
+func (m *matcher) matchCachedRow(initialized bool) []*cdcpb.Event_Row {
+	if !initialized {
+		log.Panic("must be initialized before match cahced rows")
+	}
 	cachedCommit := m.cachedCommit
 	m.cachedCommit = nil
 	top := 0
 	for i := 0; i < len(cachedCommit); i++ {
 		cacheEntry := cachedCommit[i]
-		ok := m.matchRow(cacheEntry)
+		ok := m.matchRow(cacheEntry, true)
 		if !ok {
 			// when cdc receives a commit log without a corresponding
 			// prewrite log before initialized, a committed log  with
@@ -97,3 +108,19 @@ func (m *matcher) matchCachedRow() []*cdcpb.Event_Row {
 func (m *matcher) rollbackRow(row *cdcpb.Event_Row) {
 	delete(m.unmatchedValue, newMatchKey(row))
 }
+
+func (m *matcher) cacheRollbackRow(row *cdcpb.Event_Row) {
+	m.cachedRollback = append(m.cachedRollback, row)
+}
+
+func (m *matcher) matchCachedRollbackRow(initialized bool) {
+	if !initialized {
+		log.Panic("must be initialized before match cahced rollback rows")
+	}
+	rollback := m.cachedRollback
+	m.cachedRollback = nil
+	for i := 0; i < len(rollback); i++ {
+		cacheEntry := rollback[i]
+		m.rollbackRow(cacheEntry)
+	}
+}
diff --git a/cdc/kv/region_worker.go b/cdc/kv/region_worker.go
index 0317baea1..42b57d839 100644
--- a/cdc/kv/region_worker.go
+++ b/cdc/kv/region_worker.go
@@ -154,23 +154,20 @@ type regionWorker struct {
 
 	metrics *regionWorkerMetrics
 
-	enableOldValue bool
-	storeAddr      string
+	storeAddr string
 }
 
 func newRegionWorker(s *eventFeedSession, addr string) *regionWorker {
-	cfg := config.GetGlobalServerConfig().KVClient
 	worker := &regionWorker{
-		session:        s,
-		inputCh:        make(chan *regionStatefulEvent, regionWorkerInputChanSize),
-		outputCh:       s.eventCh,
-		errorCh:        make(chan error, 1),
-		statesManager:  newRegionStateManager(-1),
-		rtsManager:     newRegionTsManager(),
-		rtsUpdateCh:    make(chan *regionTsInfo, 1024),
-		enableOldValue: s.enableOldValue,
-		storeAddr:      addr,
-		concurrent:     cfg.WorkerConcurrent,
+		session:       s,
+		inputCh:       make(chan *regionStatefulEvent, regionWorkerInputChanSize),
+		outputCh:      s.eventCh,
+		errorCh:       make(chan error, 1),
+		statesManager: newRegionStateManager(-1),
+		rtsManager:    newRegionTsManager(),
+		rtsUpdateCh:   make(chan *regionTsInfo, 1024),
+		storeAddr:     addr,
+		concurrent:    s.client.config.WorkerConcurrent,
 	}
 	return worker
 }
@@ -352,6 +349,7 @@ func (w *regionWorker) resolveLock(ctx context.Context) error {
 						continue
 					}
 					log.Warn("region not receiving resolved event from tikv or resolved ts is not pushing for too long time, try to resolve lock",
+						zap.String("addr", w.storeAddr),
 						zap.Uint64("regionID", rts.regionID),
 						zap.Stringer("span", state.getRegionSpan()),
 						zap.Duration("duration", sinceLastResolvedTs),
@@ -633,16 +631,16 @@ func (w *regionWorker) handleEventEntry(
 		case cdcpb.Event_INITIALIZED:
 			if time.Since(state.startFeedTime) > 20*time.Second {
 				log.Warn("The time cost of initializing is too much",
-					zap.Duration("timeCost", time.Since(state.startFeedTime)),
+					zap.Duration("duration", time.Since(state.startFeedTime)),
 					zap.Uint64("regionID", regionID))
 			}
 			w.metrics.metricPullEventInitializedCounter.Inc()
 
 			state.initialized = true
 			w.session.regionRouter.Release(state.sri.rpcCtx.Addr)
-			cachedEvents := state.matcher.matchCachedRow()
+			cachedEvents := state.matcher.matchCachedRow(state.initialized)
 			for _, cachedEvent := range cachedEvents {
-				revent, err := assembleRowEvent(regionID, cachedEvent, w.enableOldValue)
+				revent, err := assembleRowEvent(regionID, cachedEvent)
 				if err != nil {
 					return errors.Trace(err)
 				}
@@ -653,16 +651,17 @@ func (w *regionWorker) handleEventEntry(
 					return errors.Trace(ctx.Err())
 				}
 			}
+			state.matcher.matchCachedRollbackRow(state.initialized)
 		case cdcpb.Event_COMMITTED:
 			w.metrics.metricPullEventCommittedCounter.Inc()
-			revent, err := assembleRowEvent(regionID, entry, w.enableOldValue)
+			revent, err := assembleRowEvent(regionID, entry)
 			if err != nil {
 				return errors.Trace(err)
 			}
 
 			if entry.CommitTs <= state.lastResolvedTs {
 				logPanic("The CommitTs must be greater than the resolvedTs",
-					zap.String("Event Type", "COMMITTED"),
+					zap.String("EventType", "COMMITTED"),
 					zap.Uint64("CommitTs", entry.CommitTs),
 					zap.Uint64("resolvedTs", state.lastResolvedTs),
 					zap.Uint64("regionID", regionID))
@@ -681,13 +680,13 @@ func (w *regionWorker) handleEventEntry(
 			w.metrics.metricPullEventCommitCounter.Inc()
 			if entry.CommitTs <= state.lastResolvedTs {
 				logPanic("The CommitTs must be greater than the resolvedTs",
-					zap.String("Event Type", "COMMIT"),
+					zap.String("EventType", "COMMIT"),
 					zap.Uint64("CommitTs", entry.CommitTs),
 					zap.Uint64("resolvedTs", state.lastResolvedTs),
 					zap.Uint64("regionID", regionID))
 				return errUnreachable
 			}
-			ok := state.matcher.matchRow(entry)
+			ok := state.matcher.matchRow(entry, state.initialized)
 			if !ok {
 				if !state.initialized {
 					state.matcher.cacheCommitRow(entry)
@@ -699,7 +698,7 @@ func (w *regionWorker) handleEventEntry(
 					entry.GetType(), entry.GetOpType())
 			}
 
-			revent, err := assembleRowEvent(regionID, entry, w.enableOldValue)
+			revent, err := assembleRowEvent(regionID, entry)
 			if err != nil {
 				return errors.Trace(err)
 			}
@@ -712,6 +711,10 @@ func (w *regionWorker) handleEventEntry(
 			}
 		case cdcpb.Event_ROLLBACK:
 			w.metrics.metricPullEventRollbackCounter.Inc()
+			if !state.initialized {
+				state.matcher.cacheRollbackRow(entry)
+				continue
+			}
 			state.matcher.rollbackRow(entry)
 		}
 	}
@@ -738,8 +741,8 @@ func (w *regionWorker) handleResolvedTs(
 	}
 
 	if resolvedTs < state.lastResolvedTs {
-		log.Warn("The resolvedTs is fallen back in kvclient",
-			zap.String("Event Type", "RESOLVED"),
+		log.Debug("The resolvedTs is fallen back in kvclient",
+			zap.String("EventType", "RESOLVED"),
 			zap.Uint64("resolvedTs", resolvedTs),
 			zap.Uint64("lastResolvedTs", state.lastResolvedTs),
 			zap.Uint64("regionID", regionID))
diff --git a/cdc/metrics.go b/cdc/metrics.go
index 1af38df21..c6d4a645a 100644
--- a/cdc/metrics.go
+++ b/cdc/metrics.go
@@ -33,6 +33,7 @@ import (
 	"github.com/pingcap/tiflow/pkg/orchestrator"
 	"github.com/pingcap/tiflow/pkg/p2p"
 	"github.com/prometheus/client_golang/prometheus"
+	tikvmetrics "github.com/tikv/client-go/v2/metrics"
 )
 
 var registry = prometheus.NewRegistry()
@@ -61,4 +62,10 @@ func init() {
 	redowriter.InitMetrics(registry)
 	db.InitMetrics(registry)
 	kafka.InitMetrics(registry)
+	// TiKV client metrics, including metrics about resolved and region cache.
+	originalRegistry := prometheus.DefaultRegisterer
+	prometheus.DefaultRegisterer = registry
+	tikvmetrics.InitMetrics("ticdc", "tikvclient")
+	tikvmetrics.RegisterMetrics()
+	prometheus.DefaultRegisterer = originalRegistry
 }
diff --git a/cdc/model/changefeed.go b/cdc/model/changefeed.go
index 989e9fe67..564a47aad 100644
--- a/cdc/model/changefeed.go
+++ b/cdc/model/changefeed.go
@@ -302,9 +302,9 @@ func (info *ChangeFeedInfo) fixState() {
 
 	if state != info.State {
 		log.Info("handle old owner inconsistent state",
-			zap.String("old state", string(info.State)),
-			zap.String("admin job type", info.AdminJobType.String()),
-			zap.String("new state", string(state)))
+			zap.String("oldState", string(info.State)),
+			zap.String("adminJob", info.AdminJobType.String()),
+			zap.String("newState", string(state)))
 		info.State = state
 	}
 }
@@ -353,8 +353,8 @@ func (info *ChangeFeedInfo) fixSinkProtocol() {
 	} else {
 		if needsFix(info.Config.Sink.Protocol) {
 			log.Info("handle incompatible protocol from sink config",
-				zap.String("old protocol", info.Config.Sink.Protocol),
-				zap.String("fixed protocol", openProtocolStr))
+				zap.String("oldProtocol", info.Config.Sink.Protocol),
+				zap.String("fixedProtocol", openProtocolStr))
 			info.Config.Sink.Protocol = openProtocolStr
 		}
 	}
diff --git a/cdc/model/kv.go b/cdc/model/kv.go
index 584b7ef38..aa86d55dc 100644
--- a/cdc/model/kv.go
+++ b/cdc/model/kv.go
@@ -84,8 +84,10 @@ type RawKVEntry struct {
 }
 
 func (v *RawKVEntry) String() string {
-	return fmt.Sprintf("OpType: %v, Key: %s, Value: %s, StartTs: %d, CRTs: %d, RegionID: %d",
-		v.OpType, string(v.Key), string(v.Value), v.StartTs, v.CRTs, v.RegionID)
+	// TODO: redact values.
+	return fmt.Sprintf(
+		"OpType: %v, Key: %s, Value: %s, OldValue: %s, StartTs: %d, CRTs: %d, RegionID: %d",
+		v.OpType, string(v.Key), string(v.Value), string(v.OldValue), v.StartTs, v.CRTs, v.RegionID)
 }
 
 // ApproximateDataSize calculate the approximate size of protobuf binary
diff --git a/cdc/owner/changefeed.go b/cdc/owner/changefeed.go
index f2f455352..be630734e 100644
--- a/cdc/owner/changefeed.go
+++ b/cdc/owner/changefeed.go
@@ -116,7 +116,7 @@ func (c *changefeed) Tick(ctx cdcContext.Context, state *orchestrator.Changefeed
 	})
 	state.CheckCaptureAlive(ctx.GlobalVars().CaptureInfo.ID)
 	if err := c.tick(ctx, state, captures); err != nil {
-		log.Error("an error occurred in Owner", zap.String("changefeedID", c.state.ID), zap.Error(err))
+		log.Error("an error occurred in Owner", zap.String("changefeed", c.state.ID), zap.Error(err))
 		var code string
 		if rfcCode, ok := cerror.RFCCode(err); ok {
 			code = string(rfcCode)
@@ -331,7 +331,7 @@ func (c *changefeed) releaseResources(ctx cdcContext.Context) {
 	cancel()
 	// We don't need to wait sink Close, pass a canceled context is ok
 	if err := c.sink.close(canceledCtx); err != nil {
-		log.Warn("Closing sink failed in Owner", zap.String("changefeedID", c.state.ID), zap.Error(err))
+		log.Warn("Closing sink failed in Owner", zap.String("changefeed", c.state.ID), zap.Error(err))
 	}
 	c.wg.Wait()
 	c.scheduler.Close(ctx)
@@ -480,7 +480,7 @@ func (c *changefeed) handleBarrier(ctx cdcContext.Context) (uint64, error) {
 		}
 		c.feedStateManager.MarkFinished()
 	default:
-		log.Panic("Unknown barrier type", zap.Int("barrier type", int(barrierTp)))
+		log.Panic("Unknown barrier type", zap.Int("barrierType", int(barrierTp)))
 	}
 	return barrierTs, nil
 }
diff --git a/cdc/owner/ddl_puller.go b/cdc/owner/ddl_puller.go
index eba0480b6..1ce895fb0 100644
--- a/cdc/owner/ddl_puller.go
+++ b/cdc/owner/ddl_puller.go
@@ -26,6 +26,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/puller"
 	"github.com/pingcap/tiflow/cdc/sorter/memory"
+	"github.com/pingcap/tiflow/pkg/config"
 	cdcContext "github.com/pingcap/tiflow/pkg/context"
 	"github.com/pingcap/tiflow/pkg/filter"
 	"github.com/pingcap/tiflow/pkg/regionspan"
@@ -70,12 +71,13 @@ func newDDLPuller(ctx cdcContext.Context, startTs uint64) (DDLPuller, error) {
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
+	kvCfg := config.GetGlobalServerConfig().KVClient
 	var plr puller.Puller
 	kvStorage := ctx.GlobalVars().KVStorage
 	// kvStorage can be nil only in the test
 	if kvStorage != nil {
 		plr = puller.NewPuller(ctx, pdCli, ctx.GlobalVars().GrpcPool, ctx.GlobalVars().RegionCache, kvStorage, startTs,
-			[]regionspan.Span{regionspan.GetDDLSpan(), regionspan.GetAddIndexDDLSpan()}, false)
+			[]regionspan.Span{regionspan.GetDDLSpan(), regionspan.GetAddIndexDDLSpan()}, kvCfg)
 	}
 
 	return &ddlPullerImpl{
@@ -90,7 +92,7 @@ func newDDLPuller(ctx cdcContext.Context, startTs uint64) (DDLPuller, error) {
 func (h *ddlPullerImpl) Run(ctx cdcContext.Context) error {
 	ctx, cancel := cdcContext.WithCancel(ctx)
 	h.cancel = cancel
-	log.Debug("DDL puller started", zap.String("changefeed-id", ctx.ChangefeedVars().ID))
+	log.Debug("DDL puller started", zap.String("changefeed", ctx.ChangefeedVars().ID))
 	stdCtx := util.PutTableInfoInCtx(ctx, -1, puller.DDLPullerTableName)
 	stdCtx = util.PutChangefeedIDInCtx(stdCtx, ctx.ChangefeedVars().ID)
 	errg, stdCtx := errgroup.WithContext(stdCtx)
@@ -149,9 +151,9 @@ func (h *ddlPullerImpl) Run(ctx cdcContext.Context) error {
 				duration := h.clock.Since(lastResolvedTsAdanvcedTime)
 				if duration > ownerDDLPullerStuckWarnTimeout {
 					log.Warn("ddl puller resolved ts has not advanced",
-						zap.String("changefeed-id", ctx.ChangefeedVars().ID),
+						zap.String("changefeed", ctx.ChangefeedVars().ID),
 						zap.Duration("duration", duration),
-						zap.Uint64("resolved-ts", h.resolvedTS))
+						zap.Uint64("resolvedTs", h.resolvedTS))
 				}
 			case e := <-rawDDLCh:
 				if err := receiveDDL(e); err != nil {
diff --git a/cdc/owner/ddl_sink.go b/cdc/owner/ddl_sink.go
index c06c2e882..7601f951d 100644
--- a/cdc/owner/ddl_sink.go
+++ b/cdc/owner/ddl_sink.go
@@ -168,7 +168,7 @@ func (s *ddlSinkImpl) run(ctx cdcContext.Context, id model.ChangeFeedID, info *m
 				}
 				// If DDL executing failed, and the error can not be ignored, throw an error and pause the changefeed
 				log.Error("Execute DDL failed",
-					zap.String("ChangeFeedID", ctx.ChangefeedVars().ID),
+					zap.String("changefeed", ctx.ChangefeedVars().ID),
 					zap.Error(err),
 					zap.Reflect("ddl", ddl))
 				ctx.Throw(errors.Trace(err))
diff --git a/cdc/owner/feed_state_manager.go b/cdc/owner/feed_state_manager.go
index 9b19b212b..d6be7a717 100644
--- a/cdc/owner/feed_state_manager.go
+++ b/cdc/owner/feed_state_manager.go
@@ -172,7 +172,7 @@ func (m *feedStateManager) PushAdminJob(job *model.AdminJob) {
 	switch job.Type {
 	case model.AdminStop, model.AdminResume, model.AdminRemove:
 	default:
-		log.Panic("Can not handle this job", zap.String("changefeedID", m.state.ID),
+		log.Panic("Can not handle this job", zap.String("changefeed", m.state.ID),
 			zap.String("changefeedState", string(m.state.Info.State)), zap.Any("job", job))
 	}
 	m.pushAdminJob(job)
@@ -183,13 +183,13 @@ func (m *feedStateManager) handleAdminJob() (jobsPending bool) {
 	if job == nil || job.CfID != m.state.ID {
 		return false
 	}
-	log.Info("handle admin job", zap.String("changefeedID", m.state.ID), zap.Reflect("job", job))
+	log.Info("handle admin job", zap.String("changefeed", m.state.ID), zap.Reflect("job", job))
 	switch job.Type {
 	case model.AdminStop:
 		switch m.state.Info.State {
 		case model.StateNormal, model.StateError:
 		default:
-			log.Warn("can not pause the changefeed in the current state", zap.String("changefeedID", m.state.ID),
+			log.Warn("can not pause the changefeed in the current state", zap.String("changefeed", m.state.ID),
 				zap.String("changefeedState", string(m.state.Info.State)), zap.Any("job", job))
 			return
 		}
@@ -201,7 +201,7 @@ func (m *feedStateManager) handleAdminJob() (jobsPending bool) {
 		case model.StateNormal, model.StateError, model.StateFailed,
 			model.StateStopped, model.StateFinished, model.StateRemoved:
 		default:
-			log.Warn("can not remove the changefeed in the current state", zap.String("changefeedID", m.state.ID),
+			log.Warn("can not remove the changefeed in the current state", zap.String("changefeed", m.state.ID),
 				zap.String("changefeedState", string(m.state.Info.State)), zap.Any("job", job))
 			return
 		}
@@ -218,12 +218,12 @@ func (m *feedStateManager) handleAdminJob() (jobsPending bool) {
 			return nil, true, nil
 		})
 		checkpointTs := m.state.Info.GetCheckpointTs(m.state.Status)
-		log.Info("the changefeed is removed", zap.String("changefeed-id", m.state.ID), zap.Uint64("checkpoint-ts", checkpointTs))
+		log.Info("the changefeed is removed", zap.String("changefeed", m.state.ID), zap.Uint64("checkpointTs", checkpointTs))
 	case model.AdminResume:
 		switch m.state.Info.State {
 		case model.StateFailed, model.StateError, model.StateStopped, model.StateFinished:
 		default:
-			log.Warn("can not resume the changefeed in the current state", zap.String("changefeedID", m.state.ID),
+			log.Warn("can not resume the changefeed in the current state", zap.String("changefeed", m.state.ID),
 				zap.String("changefeedState", string(m.state.Info.State)), zap.Any("job", job))
 			return
 		}
@@ -248,7 +248,7 @@ func (m *feedStateManager) handleAdminJob() (jobsPending bool) {
 		switch m.state.Info.State {
 		case model.StateNormal:
 		default:
-			log.Warn("can not finish the changefeed in the current state", zap.String("changefeedID", m.state.ID),
+			log.Warn("can not finish the changefeed in the current state", zap.String("changefeed", m.state.ID),
 				zap.String("changefeedState", string(m.state.Info.State)), zap.Any("job", job))
 			return
 		}
@@ -341,7 +341,7 @@ func (m *feedStateManager) errorsReportedByProcessors() []*model.RunningError {
 				runningErrors = make(map[string]*model.RunningError)
 			}
 			runningErrors[position.Error.Code] = position.Error
-			log.Error("processor report an error", zap.String("changefeedID", m.state.ID), zap.String("captureID", captureID), zap.Any("error", position.Error))
+			log.Error("processor report an error", zap.String("changefeed", m.state.ID), zap.String("captureID", captureID), zap.Any("error", position.Error))
 			m.state.PatchTaskPosition(captureID, func(position *model.TaskPosition) (*model.TaskPosition, bool, error) {
 				if position == nil {
 					return nil, false, nil
diff --git a/cdc/owner/owner.go b/cdc/owner/owner.go
index 1a516bdba..eeef3b451 100644
--- a/cdc/owner/owner.go
+++ b/cdc/owner/owner.go
@@ -370,7 +370,8 @@ func (o *Owner) clusterVersionConsistent(captures map[model.CaptureID]*model.Cap
 	for _, capture := range captures {
 		if myVersion != capture.Version {
 			if o.logLimiter.Allow() {
-				log.Warn("the capture version is different with the owner", zap.Reflect("capture", capture), zap.String("owner-version", myVersion))
+				log.Warn("the capture version is different with the owner",
+					zap.Reflect("capture", capture), zap.String("ownerVer", myVersion))
 			}
 			return false
 		}
diff --git a/cdc/owner/scheduler.go b/cdc/owner/scheduler.go
index 79c77117e..351da5fb8 100644
--- a/cdc/owner/scheduler.go
+++ b/cdc/owner/scheduler.go
@@ -84,7 +84,7 @@ func NewSchedulerV2(
 	if err := ret.registerPeerMessageHandlers(ctx); err != nil {
 		return nil, err
 	}
-	log.Debug("scheduler created", zap.Uint64("checkpoint-ts", checkpointTs))
+	log.Debug("scheduler created", zap.Uint64("checkpointTs", checkpointTs))
 	return ret, nil
 }
 
@@ -234,7 +234,7 @@ func (s *schedulerV2) trySendMessage(
 		if cerror.ErrPeerMessageClientClosed.Equal(err) {
 			log.Warn("peer messaging client is closed while trying to send a message through it. "+
 				"Report a bug if this warning repeats",
-				zap.String("changefeed-id", s.changeFeedID),
+				zap.String("changefeed", s.changeFeedID),
 				zap.String("target", target))
 			return false, nil
 		}
@@ -245,7 +245,7 @@ func (s *schedulerV2) trySendMessage(
 }
 
 func (s *schedulerV2) Close(ctx context.Context) {
-	log.Debug("scheduler closed", zap.String("changefeed-id", s.changeFeedID))
+	log.Debug("scheduler closed", zap.String("changefeed", s.changeFeedID))
 	s.deregisterPeerMessageHandlers(ctx)
 }
 
diff --git a/cdc/owner/scheduler_v1.go b/cdc/owner/scheduler_v1.go
index 003fd5d7e..247f0db70 100644
--- a/cdc/owner/scheduler_v1.go
+++ b/cdc/owner/scheduler_v1.go
@@ -310,7 +310,7 @@ func (s *oldScheduler) cleanUpFinishedOperations() {
 	for captureID := range s.state.TaskStatuses {
 		s.state.PatchTaskStatus(captureID, func(status *model.TaskStatus) (*model.TaskStatus, bool, error) {
 			if status == nil {
-				log.Warn("task status of the capture is not found, may be the key in etcd was deleted", zap.String("captureID", captureID), zap.String("changeFeedID", s.state.ID))
+				log.Warn("task status of the capture is not found, may be the key in etcd was deleted", zap.String("captureID", captureID), zap.String("changefeed", s.state.ID))
 				return status, false, nil
 			}
 
@@ -362,9 +362,9 @@ func (s *oldScheduler) rebalanceByTableNum() (shouldUpdateState bool) {
 
 	log.Info("Start rebalancing",
 		zap.String("changefeed", s.state.ID),
-		zap.Int("table-num", totalTableNum),
-		zap.Int("capture-num", captureNum),
-		zap.Int("target-limit", upperLimitPerCapture))
+		zap.Int("tableNum", totalTableNum),
+		zap.Int("captureNum", captureNum),
+		zap.Int("targetLimit", upperLimitPerCapture))
 
 	for captureID, taskStatus := range s.state.TaskStatuses {
 		tableNum2Remove := len(taskStatus.Tables) - upperLimitPerCapture
@@ -391,9 +391,9 @@ func (s *oldScheduler) rebalanceByTableNum() (shouldUpdateState bool) {
 				}
 				status.RemoveTable(tableID, s.state.Status.CheckpointTs, false)
 				log.Info("Rebalance: Move table",
-					zap.Int64("table-id", tableID),
+					zap.Int64("tableID", tableID),
 					zap.String("capture", captureID),
-					zap.String("changefeed-id", s.state.ID))
+					zap.String("changefeed", s.state.ID))
 				return status, true, nil
 			})
 			tableNum2Remove--
diff --git a/cdc/processor/agent.go b/cdc/processor/agent.go
index ea7684ba9..d261f7a03 100644
--- a/cdc/processor/agent.go
+++ b/cdc/processor/agent.go
@@ -316,8 +316,8 @@ func (a *agentImpl) Barrier(_ context.Context) (done bool) {
 		sinceLastAdvanced := a.clock.Since(a.barrierLastCleared)
 		if sinceLastAdvanced > barrierNotAdvancingWarnDuration && a.barrierLogRateLimiter.Allow() {
 			log.Warn("processor send barrier not advancing, report a bug if this log repeats",
-				zap.String("changefeed-id", a.changeFeed),
-				zap.String("owner-id", a.ownerCaptureID),
+				zap.String("changefeed", a.changeFeed),
+				zap.String("ownerID", a.ownerCaptureID),
 				zap.Duration("duration", sinceLastAdvanced))
 		}
 	}()
@@ -331,7 +331,7 @@ func (a *agentImpl) Barrier(_ context.Context) (done bool) {
 		// We need to wait for the sync request anyways, and
 		// there would not be any table to replicate for now.
 		log.Debug("waiting for owner to request sync",
-			zap.String("changefeed-id", a.changeFeed))
+			zap.String("changefeed", a.changeFeed))
 		return false
 	}
 
@@ -380,7 +380,7 @@ func (a *agentImpl) Close() error {
 	log.Debug("processor messenger: closing", zap.Stack("stack"))
 	if err := a.deregisterPeerMessageHandlers(); err != nil {
 		log.Warn("failed to deregister processor message handlers",
-			zap.String("changefeed-id", a.changeFeed),
+			zap.String("changefeed", a.changeFeed),
 			zap.Error(err))
 		return errors.Trace(err)
 	}
@@ -410,7 +410,7 @@ func (a *agentImpl) trySendMessage(
 		if cerror.ErrPeerMessageClientClosed.Equal(err) {
 			log.Warn("peer messaging client is closed while trying to send a message through it. "+
 				"Report a bug if this warning repeats",
-				zap.String("changefeed-id", a.changeFeed),
+				zap.String("changefeed", a.changeFeed),
 				zap.String("target", target))
 			return false, nil
 		}
@@ -494,5 +494,5 @@ func (a *agentImpl) printNoClientWarning(target model.CaptureID) {
 		return
 	}
 	log.Warn("processor: no message client found for owner, retry later",
-		zap.String("owner-id", target))
+		zap.String("ownerID", target))
 }
diff --git a/cdc/processor/pipeline/puller.go b/cdc/processor/pipeline/puller.go
index 653f9e2b7..930fd046c 100644
--- a/cdc/processor/pipeline/puller.go
+++ b/cdc/processor/pipeline/puller.go
@@ -19,6 +19,7 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/puller"
+	"github.com/pingcap/tiflow/pkg/config"
 	cdcContext "github.com/pingcap/tiflow/pkg/context"
 	"github.com/pingcap/tiflow/pkg/pipeline"
 	"github.com/pingcap/tiflow/pkg/regionspan"
@@ -66,10 +67,12 @@ func (n *pullerNode) InitWithWaitGroup(ctx pipeline.NodeContext, wg *errgroup.Gr
 	ctxC = util.PutTableInfoInCtx(ctxC, n.tableID, n.tableName)
 	ctxC = util.PutCaptureAddrInCtx(ctxC, ctx.GlobalVars().CaptureInfo.AdvertiseAddr)
 	ctxC = util.PutChangefeedIDInCtx(ctxC, ctx.ChangefeedVars().ID)
+	kvCfg := config.GetGlobalServerConfig().KVClient
+	ctxC = util.PutRoleInCtx(ctxC, util.RoleProcessor)
 	// NOTICE: always pull the old value internally
 	// See also: https://github.com/pingcap/tiflow/issues/2301.
 	plr := puller.NewPuller(ctxC, ctx.GlobalVars().PDClient, ctx.GlobalVars().GrpcPool, ctx.GlobalVars().RegionCache, ctx.GlobalVars().KVStorage,
-		n.replicaInfo.StartTs, n.tableSpan(ctx), true)
+		n.replicaInfo.StartTs, n.tableSpan(ctx), kvCfg)
 	n.wg.Go(func() error {
 		ctx.Throw(errors.Trace(plr.Run(ctxC)))
 		return nil
diff --git a/cdc/processor/pipeline/sorter.go b/cdc/processor/pipeline/sorter.go
index c7ee17555..90ecad4b6 100644
--- a/cdc/processor/pipeline/sorter.go
+++ b/cdc/processor/pipeline/sorter.go
@@ -105,7 +105,7 @@ func (n *sorterNode) StartActorNode(ctx pipeline.NodeContext, isTableActorMode b
 	case model.SortUnified, model.SortInFile /* `file` becomes an alias of `unified` for backward compatibility */ :
 		if sortEngine == model.SortInFile {
 			log.Warn("File sorter is obsolete and replaced by unified sorter. Please revise your changefeed settings",
-				zap.String("changefeed-id", ctx.ChangefeedVars().ID), zap.String("table-name", n.tableName))
+				zap.String("changefeed", ctx.ChangefeedVars().ID), zap.String("tableName", n.tableName))
 		}
 
 		if config.GetGlobalServerConfig().Debug.EnableDBSorter {
diff --git a/cdc/processor/pipeline/table.go b/cdc/processor/pipeline/table.go
index 70ab2a002..c3a01e732 100644
--- a/cdc/processor/pipeline/table.go
+++ b/cdc/processor/pipeline/table.go
@@ -189,9 +189,9 @@ func NewTablePipeline(ctx cdcContext.Context,
 
 	perTableMemoryQuota := serverConfig.GetGlobalServerConfig().PerTableMemoryQuota
 	log.Debug("creating table flow controller",
-		zap.String("changefeed-id", ctx.ChangefeedVars().ID),
-		zap.String("table-name", tableName),
-		zap.Int64("table-id", tableID),
+		zap.String("changefeed", ctx.ChangefeedVars().ID),
+		zap.String("tableName", tableName),
+		zap.Int64("tableID", tableID),
 		zap.Uint64("quota", perTableMemoryQuota))
 	flowController := common.NewTableFlowController(perTableMemoryQuota)
 	config := ctx.ChangefeedVars().Info.Config
diff --git a/cdc/processor/processor.go b/cdc/processor/processor.go
index 1ccf6bc12..206128820 100644
--- a/cdc/processor/processor.go
+++ b/cdc/processor/processor.go
@@ -106,7 +106,7 @@ func (p *processor) AddTable(ctx cdcContext.Context, tableID model.TableID) (boo
 	}
 
 	log.Info("adding table",
-		zap.Int64("table-id", tableID),
+		zap.Int64("tableID", tableID),
 		cdcContext.ZapFieldChangefeed(ctx))
 	err := p.addTable(ctx, tableID, &model.TableReplicaInfo{})
 	if err != nil {
@@ -664,6 +664,7 @@ func (p *processor) createAndDriveSchemaStorage(ctx cdcContext.Context) (entry.S
 	kvStorage := ctx.GlobalVars().KVStorage
 	ddlspans := []regionspan.Span{regionspan.GetDDLSpan(), regionspan.GetAddIndexDDLSpan()}
 	checkpointTs := p.changefeed.Info.GetCheckpointTs(p.changefeed.Status)
+	kvCfg := config.GetGlobalServerConfig().KVClient
 	stdCtx := util.PutTableInfoInCtx(ctx, -1, puller.DDLPullerTableName)
 	stdCtx = util.PutChangefeedIDInCtx(stdCtx, ctx.ChangefeedVars().ID)
 	ddlPuller := puller.NewPuller(
@@ -672,7 +673,7 @@ func (p *processor) createAndDriveSchemaStorage(ctx cdcContext.Context) (entry.S
 		ctx.GlobalVars().GrpcPool,
 		ctx.GlobalVars().RegionCache,
 		ctx.GlobalVars().KVStorage,
-		checkpointTs, ddlspans, false)
+		checkpointTs, ddlspans, kvCfg)
 	meta, err := kv.GetSnapshotMeta(kvStorage, checkpointTs)
 	if err != nil {
 		return nil, errors.Trace(err)
diff --git a/cdc/puller/puller.go b/cdc/puller/puller.go
index 571f613a7..e39cd5ec6 100644
--- a/cdc/puller/puller.go
+++ b/cdc/puller/puller.go
@@ -24,6 +24,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/kv"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/puller/frontier"
+	"github.com/pingcap/tiflow/pkg/config"
 	"github.com/pingcap/tiflow/pkg/regionspan"
 	"github.com/pingcap/tiflow/pkg/txnutil"
 	"github.com/pingcap/tiflow/pkg/util"
@@ -52,15 +53,14 @@ type Puller interface {
 }
 
 type pullerImpl struct {
-	kvCli          kv.CDCKVClient
-	kvStorage      tikv.Storage
-	checkpointTs   uint64
-	spans          []regionspan.ComparableSpan
-	outputCh       chan *model.RawKVEntry
-	tsTracker      frontier.Frontier
-	resolvedTs     uint64
-	initialized    int64
-	enableOldValue bool
+	kvCli        kv.CDCKVClient
+	kvStorage    tikv.Storage
+	checkpointTs uint64
+	spans        []regionspan.ComparableSpan
+	outputCh     chan *model.RawKVEntry
+	tsTracker    frontier.Frontier
+	resolvedTs   uint64
+	initialized  int64
 }
 
 // NewPuller create a new Puller fetch event start from checkpointTs
@@ -73,7 +73,7 @@ func NewPuller(
 	kvStorage tidbkv.Storage,
 	checkpointTs uint64,
 	spans []regionspan.Span,
-	enableOldValue bool,
+	cfg *config.KVClientConfig,
 ) Puller {
 	tikvStorage, ok := kvStorage.(tikv.Storage)
 	if !ok {
@@ -87,17 +87,16 @@ func NewPuller(
 	// the initial ts for frontier to 0. Once the puller level resolved ts
 	// initialized, the ts should advance to a non-zero value.
 	tsTracker := frontier.NewFrontier(0, comparableSpans...)
-	kvCli := kv.NewCDCKVClient(ctx, pdCli, tikvStorage, grpcPool, regionCache)
+	kvCli := kv.NewCDCKVClient(ctx, pdCli, tikvStorage, grpcPool, regionCache, cfg)
 	p := &pullerImpl{
-		kvCli:          kvCli,
-		kvStorage:      tikvStorage,
-		checkpointTs:   checkpointTs,
-		spans:          comparableSpans,
-		outputCh:       make(chan *model.RawKVEntry, defaultPullerOutputChanSize),
-		tsTracker:      tsTracker,
-		resolvedTs:     checkpointTs,
-		initialized:    0,
-		enableOldValue: enableOldValue,
+		kvCli:        kvCli,
+		kvStorage:    tikvStorage,
+		checkpointTs: checkpointTs,
+		spans:        comparableSpans,
+		outputCh:     make(chan *model.RawKVEntry, defaultPullerOutputChanSize),
+		tsTracker:    tsTracker,
+		resolvedTs:   checkpointTs,
+		initialized:  0,
 	}
 	return p
 }
@@ -114,7 +113,7 @@ func (p *pullerImpl) Run(ctx context.Context) error {
 		span := span
 
 		g.Go(func() error {
-			return p.kvCli.EventFeed(ctx, span, checkpointTs, p.enableOldValue, lockresolver, p, eventCh)
+			return p.kvCli.EventFeed(ctx, span, checkpointTs, lockresolver, p, eventCh)
 		})
 	}
 
diff --git a/cdc/redo/manager.go b/cdc/redo/manager.go
index 07ba2c1cb..a56f02cc6 100644
--- a/cdc/redo/manager.go
+++ b/cdc/redo/manager.go
@@ -282,7 +282,7 @@ func (m *ManagerImpl) AddTable(tableID model.TableID, startTs uint64) {
 		return m.tableIDs[i] >= tableID
 	})
 	if i < len(m.tableIDs) && m.tableIDs[i] == tableID {
-		log.Warn("add duplicated table in redo log manager", zap.Int64("table-id", tableID))
+		log.Warn("add duplicated table in redo log manager", zap.Int64("tableID", tableID))
 		return
 	}
 	if i == len(m.tableIDs) {
@@ -306,7 +306,7 @@ func (m *ManagerImpl) RemoveTable(tableID model.TableID) {
 		m.tableIDs = m.tableIDs[:len(m.tableIDs)-1]
 		delete(m.rtsMap, tableID)
 	} else {
-		log.Warn("remove a table not maintained in redo log manager", zap.Int64("table-id", tableID))
+		log.Warn("remove a table not maintained in redo log manager", zap.Int64("tableID", tableID))
 	}
 }
 
diff --git a/cdc/redo/reader/file.go b/cdc/redo/reader/file.go
index 976d4a474..47d176b02 100644
--- a/cdc/redo/reader/file.go
+++ b/cdc/redo/reader/file.go
@@ -184,7 +184,7 @@ func openSelectedFiles(ctx context.Context, dir, fixedType string, startTs uint6
 		ret, err := shouldOpen(startTs, name, fixedType)
 		if err != nil {
 			log.Warn("check selected log file fail",
-				zap.String("log file", name),
+				zap.String("logFile", name),
 				zap.Error(err))
 			continue
 		}
diff --git a/cdc/redo/writer/file.go b/cdc/redo/writer/file.go
index eb846c3c1..bc1dbbe85 100644
--- a/cdc/redo/writer/file.go
+++ b/cdc/redo/writer/file.go
@@ -203,12 +203,12 @@ func (w *Writer) runFlushToDisk(ctx context.Context, flushIntervalInMs int64) {
 		case <-ctx.Done():
 			err := w.Close()
 			if err != nil {
-				log.Error("runFlushToDisk close fail", zap.String("changefeedID", w.cfg.ChangeFeedID), zap.Error(err))
+				log.Error("runFlushToDisk close fail", zap.String("changefeed", w.cfg.ChangeFeedID), zap.Error(err))
 			}
 		case <-ticker.C:
 			err := w.Flush()
 			if err != nil {
-				log.Error("redo log flush fail", zap.String("changefeedID", w.cfg.ChangeFeedID), zap.Error(err))
+				log.Error("redo log flush fail", zap.String("changefeed", w.cfg.ChangeFeedID), zap.Error(err))
 			}
 		}
 	}
@@ -513,7 +513,7 @@ func (w *Writer) getShouldRemovedFiles(checkPointTs uint64) ([]os.FileInfo, erro
 		ret, err := w.shouldRemoved(checkPointTs, f)
 		if err != nil {
 			log.Warn("check removed log file fail",
-				zap.String("log file", f.Name()),
+				zap.String("logFile", f.Name()),
 				zap.Error(err))
 			continue
 		}
diff --git a/cdc/redo/writer/writer.go b/cdc/redo/writer/writer.go
index a2ffc7b18..831831e81 100644
--- a/cdc/redo/writer/writer.go
+++ b/cdc/redo/writer/writer.go
@@ -165,7 +165,7 @@ func NewLogWriter(
 	err = logWriter.initMeta(ctx)
 	if err != nil {
 		log.Warn("init redo meta fail",
-			zap.String("change feed", cfg.ChangeFeedID),
+			zap.String("changefeed", cfg.ChangeFeedID),
 			zap.Error(err))
 	}
 	if cfg.S3Storage {
@@ -276,12 +276,12 @@ func (l *LogWriter) runGC(ctx context.Context) {
 		case <-ctx.Done():
 			err := l.Close()
 			if err != nil {
-				log.Error("runGC close fail", zap.String("changefeedID", l.cfg.ChangeFeedID), zap.Error(err))
+				log.Error("runGC close fail", zap.String("changefeed", l.cfg.ChangeFeedID), zap.Error(err))
 			}
 		case <-ticker.C:
 			err := l.gc()
 			if err != nil {
-				log.Error("redo log GC fail", zap.String("changefeedID", l.cfg.ChangeFeedID), zap.Error(err))
+				log.Error("redo log GC fail", zap.String("changefeed", l.cfg.ChangeFeedID), zap.Error(err))
 			}
 		}
 	}
diff --git a/cdc/scheduler/agent.go b/cdc/scheduler/agent.go
index 8a6fd03b0..0c27f5a45 100644
--- a/cdc/scheduler/agent.go
+++ b/cdc/scheduler/agent.go
@@ -222,7 +222,7 @@ func (a *BaseAgent) Tick(ctx context.Context) error {
 		if op.Epoch != a.getEpoch() {
 			a.logger.Info("dispatch request epoch does not match",
 				zap.String("epoch", op.Epoch),
-				zap.String("expected-epoch", a.getEpoch()))
+				zap.String("expectedEpoch", a.getEpoch()))
 			continue
 		}
 		if _, ok := a.tableOperations[op.TableID]; ok {
@@ -370,8 +370,8 @@ func (a *BaseAgent) OnOwnerDispatchedTask(
 ) {
 	if !a.updateOwnerInfo(ownerCaptureID, ownerRev) {
 		a.logger.Info("task from stale owner ignored",
-			zap.Int64("table-id", tableID),
-			zap.Bool("is-delete", isDelete))
+			zap.Int64("tableID", tableID),
+			zap.Bool("isDelete", isDelete))
 		return
 	}
 
@@ -388,8 +388,8 @@ func (a *BaseAgent) OnOwnerDispatchedTask(
 	a.pendingOps.PushBack(op)
 
 	a.logger.Info("OnOwnerDispatchedTask",
-		zap.String("ownerCapture-id", ownerCaptureID),
-		zap.Int64("owner-rev", ownerRev),
+		zap.String("ownerCaptureID", ownerCaptureID),
+		zap.Int64("ownerRev", ownerRev),
 		zap.Any("op", op))
 }
 
@@ -404,8 +404,8 @@ func (a *BaseAgent) OnOwnerAnnounce(
 ) {
 	if !a.updateOwnerInfo(ownerCaptureID, ownerRev) {
 		a.logger.Info("sync request from stale owner ignored",
-			zap.String("owner-capture-id", ownerCaptureID),
-			zap.Int64("owner-rev", ownerRev))
+			zap.String("ownerCaptureID", ownerCaptureID),
+			zap.Int64("ownerRev", ownerRev))
 		return
 	}
 
@@ -414,8 +414,8 @@ func (a *BaseAgent) OnOwnerAnnounce(
 	a.needSyncNow.Store(true)
 
 	a.logger.Info("OnOwnerAnnounce",
-		zap.String("owner-capture-id", ownerCaptureID),
-		zap.Int64("owner-rev", ownerRev))
+		zap.String("ownerCaptureID", ownerCaptureID),
+		zap.Int64("ownerRev", ownerRev))
 }
 
 // updateOwnerInfo tries to update the stored ownerInfo, and returns false if the
@@ -441,7 +441,7 @@ func (a *BaseAgent) updateOwnerInfo(ownerCaptureID model.CaptureID, ownerRev int
 		a.ownerHasChanged.Store(true)
 
 		a.logger.Info("owner updated",
-			zap.Any("new-owner-info", a.ownerInfo))
+			zap.Any("newOwner", a.ownerInfo))
 
 		// Resets the deque so that pending operations from the previous owner
 		// will not be processed.
@@ -455,11 +455,11 @@ func (a *BaseAgent) updateOwnerInfo(ownerCaptureID model.CaptureID, ownerRev int
 	if a.ownerInfo.OwnerRev > ownerRev {
 		// the owner where the message just came from is stale.
 		a.logger.Info("message received from stale owner",
-			zap.Any("old-owner", ownerInfo{
+			zap.Any("oldOwner", ownerInfo{
 				OwnerCaptureID: ownerCaptureID,
 				OwnerRev:       ownerRev,
 			}),
-			zap.Any("current-owner", a.ownerInfo))
+			zap.Any("currentOwner", a.ownerInfo))
 
 		// Returning false indicates that we should reject the owner,
 		// because it is stale.
diff --git a/cdc/scheduler/agent_mock.go b/cdc/scheduler/agent_mock.go
index 94dcf6eb8..77e826a8d 100644
--- a/cdc/scheduler/agent_mock.go
+++ b/cdc/scheduler/agent_mock.go
@@ -102,7 +102,7 @@ func NewMockTableExecutor(t *testing.T) *MockTableExecutor {
 }
 
 func (e *MockTableExecutor) AddTable(ctx cdcContext.Context, tableID model.TableID) (bool, error) {
-	log.Info("AddTable", zap.Int64("table-id", tableID))
+	log.Info("AddTable", zap.Int64("tableID", tableID))
 	require.NotContains(e.t, e.Adding, tableID)
 	require.NotContains(e.t, e.Running, tableID)
 	require.NotContains(e.t, e.Removing, tableID)
@@ -115,7 +115,7 @@ func (e *MockTableExecutor) AddTable(ctx cdcContext.Context, tableID model.Table
 }
 
 func (e *MockTableExecutor) RemoveTable(ctx cdcContext.Context, tableID model.TableID) (bool, error) {
-	log.Info("RemoveTable", zap.Int64("table-id", tableID))
+	log.Info("RemoveTable", zap.Int64("tableID", tableID))
 	args := e.Called(ctx, tableID)
 	require.Contains(e.t, e.Running, tableID)
 	require.NotContains(e.t, e.Removing, tableID)
diff --git a/cdc/scheduler/balancer.go b/cdc/scheduler/balancer.go
index 0cfa61c22..1c2e99855 100644
--- a/cdc/scheduler/balancer.go
+++ b/cdc/scheduler/balancer.go
@@ -157,9 +157,9 @@ func (r *tableNumberBalancer) FindVictims(
 	upperLimitPerCapture := int(math.Ceil(float64(totalTableNum) / float64(captureNum)))
 
 	r.logger.Info("Start rebalancing",
-		zap.Int("table-num", totalTableNum),
-		zap.Int("capture-num", captureNum),
-		zap.Int("target-limit", upperLimitPerCapture))
+		zap.Int("tableNum", totalTableNum),
+		zap.Int("captureNum", captureNum),
+		zap.Int("targetLimit", upperLimitPerCapture))
 
 	var victims []*util.TableRecord
 	for _, tables := range tables.GetAllTablesGroupedByCaptures() {
@@ -201,7 +201,7 @@ func (r *tableNumberBalancer) FindVictims(
 			}
 
 			r.logger.Info("Rebalance: find victim table",
-				zap.Any("table-record", record))
+				zap.Any("tableRecord", record))
 			victims = append(victims, record)
 			tableNum2Remove--
 		}
diff --git a/cdc/scheduler/checkpoint_sender.go b/cdc/scheduler/checkpoint_sender.go
index 54d859ff1..db425fda2 100644
--- a/cdc/scheduler/checkpoint_sender.go
+++ b/cdc/scheduler/checkpoint_sender.go
@@ -79,7 +79,7 @@ func (s *checkpointTsSenderImpl) SendCheckpoint(ctx context.Context, provider ch
 	if s.enRouteCheckpointTs != noEnRouteCheckpointTs {
 		if !s.communicator.Barrier(ctx) {
 			s.logger.Debug("not sending checkpoint due to pending barrier",
-				zap.Duration("since-last-sent", time.Since(s.lastSendCheckpointTime)))
+				zap.Duration("sinceLastSent", time.Since(s.lastSendCheckpointTime)))
 			// We cannot proceed because the last checkpoint has not been acknowledged to have
 			// been received.
 			return nil
diff --git a/cdc/scheduler/schedule_dispatcher.go b/cdc/scheduler/schedule_dispatcher.go
index 7ba115050..7f1f01616 100644
--- a/cdc/scheduler/schedule_dispatcher.go
+++ b/cdc/scheduler/schedule_dispatcher.go
@@ -106,7 +106,7 @@ func NewBaseScheduleDispatcher(
 	checkpointTs model.Ts,
 ) *BaseScheduleDispatcher {
 	// logger is just the global logger with the `changefeed-id` field attached.
-	logger := log.L().With(zap.String("changefeed-id", changeFeedID))
+	logger := log.L().With(zap.String("changefeed", changeFeedID))
 
 	return &BaseScheduleDispatcher{
 		tables:               util.NewTableSet(),
@@ -228,7 +228,7 @@ func (s *BaseScheduleDispatcher) Tick(
 	for _, tableID := range toRemove {
 		record, ok := s.tables.GetTableRecord(tableID)
 		if !ok {
-			s.logger.Panic("table not found", zap.Int64("table-id", tableID))
+			s.logger.Panic("table not found", zap.Int64("tableID", tableID))
 		}
 		if record.Status != util.RunningTable {
 			// another operation is in progress
@@ -309,7 +309,7 @@ func (s *BaseScheduleDispatcher) syncCaptures(ctx context.Context) (capturesAllS
 			// removes expired captures from the captureSynced map
 			delete(s.captureStatus, captureID)
 			s.logger.Debug("syncCaptures: remove offline capture",
-				zap.String("capture-id", captureID))
+				zap.String("captureID", captureID))
 		}
 	}
 	for captureID := range s.captures {
@@ -333,7 +333,7 @@ func (s *BaseScheduleDispatcher) syncCaptures(ctx context.Context) (capturesAllS
 			if done {
 				s.captureStatus[captureID].SyncStatus = captureSyncSent
 				s.logger.Info("syncCaptures: sent sync request",
-					zap.String("capture-id", captureID))
+					zap.String("captureID", captureID))
 			}
 		case captureSyncFinished:
 			finishedCount++
@@ -359,8 +359,8 @@ func (s *BaseScheduleDispatcher) descheduleTablesFromDownCaptures() {
 			// gone capture.
 			removed := s.tables.RemoveTableRecordByCaptureID(captureID)
 			s.logger.Info("capture down, removing tables",
-				zap.String("capture-id", captureID),
-				zap.Any("removed-tables", removed))
+				zap.String("captureID", captureID),
+				zap.Any("removedTables", removed))
 			s.moveTableManager.OnCaptureRemoved(captureID)
 		}
 	}
@@ -421,7 +421,7 @@ func (s *BaseScheduleDispatcher) addTable(
 		CaptureID: target,
 		Status:    util.AddingTable,
 	}); !ok {
-		s.logger.Panic("duplicate table", zap.Int64("table-id", tableID))
+		s.logger.Panic("duplicate table", zap.Int64("tableID", tableID))
 	}
 	return true, nil
 }
@@ -432,7 +432,7 @@ func (s *BaseScheduleDispatcher) removeTable(
 ) (done bool, err error) {
 	record, ok := s.tables.GetTableRecord(tableID)
 	if !ok {
-		s.logger.Panic("table not found", zap.Int64("table-id", tableID))
+		s.logger.Panic("table not found", zap.Int64("tableID", tableID))
 	}
 	// need to delete table
 	captureID := record.CaptureID
@@ -455,8 +455,8 @@ func (s *BaseScheduleDispatcher) MoveTable(tableID model.TableID, target model.C
 	if !s.moveTableManager.Add(tableID, target) {
 		log.Info("Move Table command has been ignored, because the last user triggered"+
 			"move has not finished",
-			zap.Int64("table-id", tableID),
-			zap.String("target-capture", target))
+			zap.Int64("tableID", tableID),
+			zap.String("targetCapture", target))
 	}
 }
 
@@ -465,14 +465,14 @@ func (s *BaseScheduleDispatcher) handleMoveTableJobs(ctx context.Context) (bool,
 		func(ctx context.Context, tableID model.TableID, target model.CaptureID) (removeTableResult, error) {
 			_, ok := s.tables.GetTableRecord(tableID)
 			if !ok {
-				s.logger.Warn("table does not exist", zap.Int64("table-id", tableID))
+				s.logger.Warn("table does not exist", zap.Int64("tableID", tableID))
 				return removeTableResultGiveUp, nil
 			}
 
 			if _, ok := s.captures[target]; !ok {
 				s.logger.Warn("move table target does not exist",
-					zap.Int64("table-id", tableID),
-					zap.String("target-capture", target))
+					zap.Int64("tableID", tableID),
+					zap.String("targetCapture", target))
 				return removeTableResultGiveUp, nil
 			}
 
@@ -502,7 +502,7 @@ func (s *BaseScheduleDispatcher) rebalance(ctx context.Context) (done bool, err
 	for _, record := range tablesToRemove {
 		if record.Status != util.RunningTable {
 			s.logger.DPanic("unexpected table status",
-				zap.Any("table-record", record))
+				zap.Any("tableRecord", record))
 		}
 
 		epoch := s.captureStatus[record.CaptureID].Epoch
@@ -533,8 +533,8 @@ func (s *BaseScheduleDispatcher) OnAgentFinishedTableOperation(
 	defer s.mu.Unlock()
 
 	logger := s.logger.With(
-		zap.String("capture-id", captureID),
-		zap.Int64("table-id", tableID),
+		zap.String("captureID", captureID),
+		zap.Int64("tableID", tableID),
 		zap.String("epoch", epoch),
 	)
 
@@ -589,9 +589,9 @@ func (s *BaseScheduleDispatcher) OnAgentSyncTaskStatuses(
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	logger := s.logger.With(zap.String("capture-id", captureID))
+	logger := s.logger.With(zap.String("captureID", captureID))
 	logger.Info("scheduler received sync",
-		zap.String("capture-id", captureID),
+		zap.String("captureID", captureID),
 		zap.String("epoch", epoch))
 
 	if ce := logger.Check(zap.DebugLevel, "OnAgentSyncTaskStatuses"); ce != nil {
@@ -609,31 +609,31 @@ func (s *BaseScheduleDispatcher) OnAgentSyncTaskStatuses(
 
 	if _, ok := s.captureStatus[captureID]; !ok {
 		logger.Warn("received sync from a capture not previously tracked, ignore",
-			zap.Any("capture-status", s.captureStatus))
+			zap.Any("captureStatus", s.captureStatus))
 		return
 	}
 
 	for _, tableID := range adding {
 		if record, ok := s.tables.GetTableRecord(tableID); ok {
 			logger.Panic("duplicate table tasks",
-				zap.Int64("table-id", tableID),
-				zap.String("actual-capture-id", record.CaptureID))
+				zap.Int64("tableID", tableID),
+				zap.String("actualCaptureID", record.CaptureID))
 		}
 		s.tables.AddTableRecord(&util.TableRecord{TableID: tableID, CaptureID: captureID, Status: util.AddingTable})
 	}
 	for _, tableID := range running {
 		if record, ok := s.tables.GetTableRecord(tableID); ok {
 			logger.Panic("duplicate table tasks",
-				zap.Int64("table-id", tableID),
-				zap.String("actual-capture-id", record.CaptureID))
+				zap.Int64("tableID", tableID),
+				zap.String("actualCaptureID", record.CaptureID))
 		}
 		s.tables.AddTableRecord(&util.TableRecord{TableID: tableID, CaptureID: captureID, Status: util.RunningTable})
 	}
 	for _, tableID := range removing {
 		if record, ok := s.tables.GetTableRecord(tableID); ok {
 			logger.Panic("duplicate table tasks",
-				zap.Int64("table-id", tableID),
-				zap.String("actual-capture-id", record.CaptureID))
+				zap.Int64("tableID", tableID),
+				zap.String("actualCaptureID", record.CaptureID))
 		}
 		s.tables.AddTableRecord(&util.TableRecord{TableID: tableID, CaptureID: captureID, Status: util.RemovingTable})
 	}
@@ -648,9 +648,9 @@ func (s *BaseScheduleDispatcher) OnAgentCheckpoint(captureID model.CaptureID, ch
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	logger := s.logger.With(zap.String("capture-id", captureID),
-		zap.Uint64("checkpoint-ts", checkpointTs),
-		zap.Uint64("resolved-ts", resolvedTs))
+	logger := s.logger.With(zap.String("captureID", captureID),
+		zap.Uint64("checkpointTs", checkpointTs),
+		zap.Uint64("resolvedTs", resolvedTs))
 
 	status, ok := s.captureStatus[captureID]
 	if !ok || status.SyncStatus != captureSyncFinished {
diff --git a/cdc/scheduler/util/table_set.go b/cdc/scheduler/util/table_set.go
index 5c83b897c..c06e7fd69 100644
--- a/cdc/scheduler/util/table_set.go
+++ b/cdc/scheduler/util/table_set.go
@@ -153,7 +153,7 @@ func (s *TableSet) RemoveTableRecord(tableID model.TableID) bool {
 
 	captureIndexEntry, ok := s.captureIndex[record.CaptureID]
 	if !ok {
-		log.Panic("unreachable", zap.Int64("table-id", tableID))
+		log.Panic("unreachable", zap.Int64("tableID", tableID))
 	}
 	delete(captureIndexEntry, record.TableID)
 	if len(captureIndexEntry) == 0 {
diff --git a/cdc/server.go b/cdc/server.go
index f6e50bd56..dc848a88b 100644
--- a/cdc/server.go
+++ b/cdc/server.go
@@ -73,7 +73,7 @@ type Server struct {
 func NewServer(pdEndpoints []string) (*Server, error) {
 	conf := config.GetGlobalServerConfig()
 	log.Info("creating CDC server",
-		zap.Strings("pd-addrs", pdEndpoints),
+		zap.Strings("pd", pdEndpoints),
 		zap.Stringer("config", conf),
 	)
 
diff --git a/cdc/sink/codec/avro.go b/cdc/sink/codec/avro.go
index 13fbcf8fa..49e4b4509 100644
--- a/cdc/sink/codec/avro.go
+++ b/cdc/sink/codec/avro.go
@@ -367,7 +367,7 @@ func getAvroDataTypeFromColumn(col *model.Column) (interface{}, error) {
 	case mysql.TypeYear:
 		return "long", nil
 	default:
-		log.Panic("Unknown MySql type", zap.Reflect("mysql-type", col.Type))
+		log.Panic("Unknown MySql type", zap.Reflect("mysqlType", col.Type))
 		return "", errors.New("Unknown Mysql type")
 	}
 }
diff --git a/cdc/sink/codec/interface.go b/cdc/sink/codec/interface.go
index 996eb4d2d..0dafac1d1 100644
--- a/cdc/sink/codec/interface.go
+++ b/cdc/sink/codec/interface.go
@@ -180,7 +180,7 @@ func NewEventBatchEncoderBuilder(p config.Protocol, credential *security.Credent
 	case config.ProtocolCraft:
 		return newCraftEventBatchEncoderBuilder(opts), nil
 	default:
-		log.Warn("unknown codec protocol value of EventBatchEncoder, use open-protocol as the default", zap.Int("protocol_value", int(p)))
+		log.Warn("unknown codec protocol value of EventBatchEncoder, use open-protocol as the default", zap.Int("protocolValue", int(p)))
 		return newJSONEventBatchEncoderBuilder(opts), nil
 	}
 }
diff --git a/cdc/sink/codec/json.go b/cdc/sink/codec/json.go
index e1b4170d6..979dacbfa 100644
--- a/cdc/sink/codec/json.go
+++ b/cdc/sink/codec/json.go
@@ -496,7 +496,7 @@ func (d *JSONEventBatchEncoder) AppendRowChangedEvent(e *model.RowChangedEvent)
 		if message.Length() > d.maxMessageBytes {
 			// `len(d.messageBuf) == 1` is implied
 			log.Debug("Event does not fit into max-message-bytes. Adjust relevant configurations to avoid service interruptions.",
-				zap.Int("event-len", message.Length()), zap.Int("max-message-bytes", d.maxMessageBytes))
+				zap.Int("eventLen", message.Length()), zap.Int("max-message-bytes", d.maxMessageBytes))
 		}
 		d.curBatchSize++
 	}
diff --git a/cdc/sink/mysql.go b/cdc/sink/mysql.go
index 39f033045..688967d26 100644
--- a/cdc/sink/mysql.go
+++ b/cdc/sink/mysql.go
@@ -296,7 +296,10 @@ func (s *mysqlSink) execDDLWithMaxRetries(ctx context.Context, ddl *model.DDLEve
 			log.Warn("execute DDL with error, retry later", zap.String("query", ddl.Query), zap.Error(err))
 		}
 		return err
-	}, retry.WithBackoffBaseDelay(backoffBaseDelayInMs), retry.WithBackoffMaxDelay(backoffMaxDelayInMs), retry.WithMaxTries(defaultDDLMaxRetryTime), retry.WithIsRetryableErr(cerror.IsRetryableError))
+	}, retry.WithBackoffBaseDelay(backoffBaseDelayInMs),
+		retry.WithBackoffMaxDelay(backoffMaxDelayInMs),
+		retry.WithMaxTries(defaultDDLMaxRetry),
+		retry.WithIsRetryableErr(cerror.IsRetryableError))
 }
 
 func (s *mysqlSink) execDDL(ctx context.Context, ddl *model.DDLEvent) error {
@@ -491,13 +494,13 @@ func (s *mysqlSink) Barrier(ctx context.Context, tableID model.TableID) error {
 			maxResolvedTs, ok := s.tableMaxResolvedTs.Load(tableID)
 			log.Warn("Barrier doesn't return in time, may be stuck",
 				zap.Int64("tableID", tableID),
-				zap.Bool("has resolvedTs", ok),
+				zap.Bool("hasResolvedTs", ok),
 				zap.Any("resolvedTs", maxResolvedTs),
 				zap.Uint64("checkpointTs", s.getTableCheckpointTs(tableID)))
 		default:
 			v, ok := s.tableMaxResolvedTs.Load(tableID)
 			if !ok {
-				log.Info("No table resolvedTs is found", zap.Int64("table-id", tableID))
+				log.Info("No table resolvedTs is found", zap.Int64("tableID", tableID))
 				return nil
 			}
 			maxResolvedTs := v.(uint64)
@@ -604,7 +607,10 @@ func (s *mysqlSink) execDMLWithMaxRetries(ctx context.Context, dmls *preparedDML
 			zap.Int("num of Rows", dmls.rowCount),
 			zap.Int("bucket", bucket))
 		return nil
-	}, retry.WithBackoffBaseDelay(backoffBaseDelayInMs), retry.WithBackoffMaxDelay(backoffMaxDelayInMs), retry.WithMaxTries(defaultDMLMaxRetryTime), retry.WithIsRetryableErr(isRetryableDMLError))
+	}, retry.WithBackoffBaseDelay(backoffBaseDelayInMs),
+		retry.WithBackoffMaxDelay(backoffMaxDelayInMs),
+		retry.WithMaxTries(defaultDMLMaxRetry),
+		retry.WithIsRetryableErr(isRetryableDMLError))
 }
 
 type preparedDMLs struct {
diff --git a/cdc/sink/mysql_params.go b/cdc/sink/mysql_params.go
index 6506506cd..3b85435da 100644
--- a/cdc/sink/mysql_params.go
+++ b/cdc/sink/mysql_params.go
@@ -36,8 +36,6 @@ const (
 	DefaultWorkerCount = 16
 	DefaultMaxTxnRow   = 256
 
-	defaultDMLMaxRetryTime     = 8
-	defaultDDLMaxRetryTime     = 20
 	defaultTiDBTxnMode         = "optimistic"
 	defaultFlushInterval       = time.Millisecond * 50
 	defaultBatchReplaceEnabled = true
@@ -48,6 +46,11 @@ const (
 	defaultSafeMode            = true
 )
 
+var (
+	defaultDMLMaxRetry uint64 = 8
+	defaultDDLMaxRetry uint64 = 20
+)
+
 var defaultParams = &sinkParams{
 	workerCount:         DefaultWorkerCount,
 	maxTxnRow:           DefaultMaxTxnRow,
@@ -246,7 +249,7 @@ func generateDSNByParams(
 
 	dsnClone := dsnCfg.Clone()
 	dsnClone.Passwd = "******"
-	log.Info("sink uri is configured", zap.String("format dsn", dsnClone.FormatDSN()))
+	log.Info("sink uri is configured", zap.String("dsn", dsnClone.FormatDSN()))
 
 	return dsnCfg.FormatDSN(), nil
 }
diff --git a/cdc/sink/mysql_worker.go b/cdc/sink/mysql_worker.go
index b5e23a51b..a76bab79f 100644
--- a/cdc/sink/mysql_worker.go
+++ b/cdc/sink/mysql_worker.go
@@ -101,7 +101,7 @@ func (w *mysqlSinkWorker) run(ctx context.Context) (err error) {
 			stackSize := runtime.Stack(buf, false)
 			buf = buf[:stackSize]
 			err = cerror.ErrMySQLWorkerPanic.GenWithStack("mysql sink concurrent execute panic, stack: %v", string(buf))
-			log.Error("mysql sink worker panic", zap.Reflect("r", r), zap.Stack("stack trace"))
+			log.Error("mysql sink worker panic", zap.Reflect("r", r), zap.Stack("stacktrace"))
 		}
 	}()
 
diff --git a/cdc/sorter/unified/backend_pool.go b/cdc/sorter/unified/backend_pool.go
index 5e8b864d3..703882ab1 100644
--- a/cdc/sorter/unified/backend_pool.go
+++ b/cdc/sorter/unified/backend_pool.go
@@ -190,8 +190,8 @@ func (p *backEndPool) alloc(ctx context.Context) (backEnd, error) {
 	tableID, tableName := util.TableIDFromCtx(ctx)
 	log.Debug("Unified Sorter: trying to create file backEnd",
 		zap.String("filename", fname),
-		zap.Int64("table-id", tableID),
-		zap.String("table-name", tableName))
+		zap.Int64("tableID", tableID),
+		zap.String("tableName", tableName))
 
 	if err := checkDataDirSatisfied(); err != nil {
 		return nil, errors.Trace(err)
@@ -296,7 +296,8 @@ func (p *backEndPool) terminate() {
 		log.Debug("Unified Sorter backEnd removing file", zap.String("file", file))
 		err = os.RemoveAll(file)
 		if err != nil {
-			log.Warn("Unified Sorter clean-up failed: failed to remove", zap.String("file-name", file), zap.Error(err))
+			log.Warn("Unified Sorter clean-up failed: failed to remove",
+				zap.String("fileName", file), zap.Error(err))
 		}
 	}
 
@@ -331,7 +332,7 @@ func (p *backEndPool) lockSortDir() error {
 				"Make sure that another instance of TiCDC, or any other program, is not using the directory. "+
 				"If you believe you should not see this error, try deleting the lock file and resume the changefeed. "+
 				"Report a bug or contact support if the problem persists.",
-				zap.String("lock-file", lockFileName))
+				zap.String("lockFile", lockFileName))
 			return errors.Trace(err)
 		}
 		return cerrors.ErrSortDirLockError.Wrap(err).GenWithStackByCause()
@@ -352,7 +353,7 @@ func (p *backEndPool) unlockSortDir() error {
 func (p *backEndPool) cleanUpStaleFiles() error {
 	if p.dir == "" {
 		// guard against programmer error. Must be careful when we are deleting user files.
-		log.Panic("unexpected sort-dir", zap.String("sort-dir", p.dir))
+		log.Panic("unexpected sort-dir", zap.String("sortDir", p.dir))
 	}
 
 	files, err := filepath.Glob(filepath.Join(p.dir, fmt.Sprintf("%s-*", sortDirDataFileMagicPrefix)))
diff --git a/cdc/sorter/unified/heap_sorter.go b/cdc/sorter/unified/heap_sorter.go
index 53fe12fd1..4b54f8897 100644
--- a/cdc/sorter/unified/heap_sorter.go
+++ b/cdc/sorter/unified/heap_sorter.go
@@ -170,9 +170,9 @@ func (h *heapSorter) flush(ctx context.Context, maxResolvedTs uint64) error {
 	failpoint.Inject("sorterDebug", func() {
 		tableID, tableName := util.TableIDFromCtx(ctx)
 		log.Debug("Unified Sorter new flushTask",
-			zap.Int64("table-id", tableID),
-			zap.String("table-name", tableName),
-			zap.Int("heap-id", task.heapSorterID),
+			zap.Int64("tableID", tableID),
+			zap.String("tableName", tableName),
+			zap.Int("heapID", task.heapSorterID),
 			zap.Uint64("resolvedTs", task.maxResolvedTs))
 	})
 
@@ -256,9 +256,9 @@ func (h *heapSorter) flush(ctx context.Context, maxResolvedTs uint64) error {
 			failpoint.Inject("sorterDebug", func() {
 				tableID, tableName := util.TableIDFromCtx(ctx)
 				log.Debug("Unified Sorter flushTask finished",
-					zap.Int("heap-id", task.heapSorterID),
-					zap.Int64("table-id", tableID),
-					zap.String("table-name", tableName),
+					zap.Int("heapID", task.heapSorterID),
+					zap.Int64("tableID", tableID),
+					zap.String("tableName", tableName),
 					zap.Uint64("resolvedTs", task.maxResolvedTs),
 					zap.Uint64("data-size", dataSize),
 					zap.Int("size", eventCount))
diff --git a/cdc/sorter/unified/merger.go b/cdc/sorter/unified/merger.go
index 3ab1a2096..b3fe8cab1 100644
--- a/cdc/sorter/unified/merger.go
+++ b/cdc/sorter/unified/merger.go
@@ -259,8 +259,8 @@ func runMerger(ctx context.Context, numSorters int, in <-chan *flushTask, out ch
 			if sortHeap.Len() > 0 {
 				tableID, tableName := util.TableIDFromCtx(ctx)
 				log.Debug("Unified Sorter: start merging",
-					zap.Int64("table-id", tableID),
-					zap.String("table-name", tableName),
+					zap.Int64("tableID", tableID),
+					zap.String("tableName", tableName),
 					zap.Uint64("minResolvedTs", minResolvedTs))
 			}
 		})
@@ -274,7 +274,7 @@ func runMerger(ctx context.Context, numSorters int, in <-chan *flushTask, out ch
 			event := item.entry
 
 			if event.CRTs < task.lastTs {
-				log.Panic("unified sorter: ts regressed in one backEnd, bug?", zap.Uint64("cur-ts", event.CRTs), zap.Uint64("last-ts", task.lastTs))
+				log.Panic("unified sorter: ts regressed in one backEnd, bug?", zap.Uint64("curTs", event.CRTs), zap.Uint64("lastTs", task.lastTs))
 			}
 			task.lastTs = event.CRTs
 
@@ -284,22 +284,22 @@ func runMerger(ctx context.Context, numSorters int, in <-chan *flushTask, out ch
 						item := heap.Pop(sortHeap).(*sortItem)
 						task := item.data.(*flushTask)
 						event := item.entry
-						log.Debug("dump", zap.Reflect("event", event), zap.Int("heap-id", task.heapSorterID))
+						log.Debug("dump", zap.Reflect("event", event), zap.Int("heapID", task.heapSorterID))
 					}
 					log.Panic("unified sorter: output ts regressed, bug?",
 						zap.Int("counter", counter),
 						zap.Uint64("minResolvedTs", minResolvedTs),
-						zap.Int("cur-heap-id", task.heapSorterID),
-						zap.Int("cur-task-id", task.taskID),
-						zap.Uint64("cur-task-resolved", task.maxResolvedTs),
-						zap.Reflect("cur-event", event),
-						zap.Uint64("cur-ts", event.CRTs),
-						zap.Int("last-heap-id", lastTask.heapSorterID),
-						zap.Int("last-task-id", lastTask.taskID),
-						zap.Uint64("last-task-resolved", task.maxResolvedTs),
-						zap.Reflect("last-event", lastEvent),
-						zap.Uint64("last-ts", lastOutputTs),
-						zap.Int("sort-heap-len", sortHeap.Len()))
+						zap.Int("curHeapID", task.heapSorterID),
+						zap.Int("curTaskID", task.taskID),
+						zap.Uint64("curTaskResolved", task.maxResolvedTs),
+						zap.Reflect("curEvent", event),
+						zap.Uint64("curTs", event.CRTs),
+						zap.Int("lastHeapID", lastTask.heapSorterID),
+						zap.Int("lastTaskID", lastTask.taskID),
+						zap.Uint64("lastTask-resolved", task.maxResolvedTs),
+						zap.Reflect("lastEvent", lastEvent),
+						zap.Uint64("lastTs", lastOutputTs),
+						zap.Int("sortHeapLen", sortHeap.Len()))
 				}
 
 				if event.CRTs <= lastOutputResolvedTs {
@@ -367,8 +367,8 @@ func runMerger(ctx context.Context, numSorters int, in <-chan *flushTask, out ch
 				if counter%10 == 0 {
 					tableID, tableName := util.TableIDFromCtx(ctx)
 					log.Debug("Merging progress",
-						zap.Int64("table-id", tableID),
-						zap.String("table-name", tableName),
+						zap.Int64("tableID", tableID),
+						zap.String("tableName", tableName),
 						zap.Int("counter", counter))
 				}
 			})
@@ -387,8 +387,8 @@ func runMerger(ctx context.Context, numSorters int, in <-chan *flushTask, out ch
 			if counter > 0 {
 				tableID, tableName := util.TableIDFromCtx(ctx)
 				log.Debug("Unified Sorter: merging ended",
-					zap.Int64("table-id", tableID),
-					zap.String("table-name", tableName),
+					zap.Int64("tableID", tableID),
+					zap.String("tableName", tableName),
 					zap.Uint64("resolvedTs", minResolvedTs), zap.Int("count", counter))
 			}
 		})
@@ -421,8 +421,8 @@ func runMerger(ctx context.Context, numSorters int, in <-chan *flushTask, out ch
 			if task == nil {
 				tableID, tableName := util.TableIDFromCtx(ctx)
 				log.Debug("Merger input channel closed, exiting",
-					zap.Int64("table-id", tableID),
-					zap.String("table-name", tableName))
+					zap.Int64("tableID", tableID),
+					zap.String("tableName", tableName))
 				return nil
 			}
 
@@ -475,8 +475,8 @@ func runMerger(ctx context.Context, numSorters int, in <-chan *flushTask, out ch
 					}
 				} else if curResolvedTs < lastResolvedTs {
 					log.Panic("resolved-ts regressed in sorter",
-						zap.Uint64("cur-resolved-ts", curResolvedTs),
-						zap.Uint64("last-resolved-ts", lastResolvedTs))
+						zap.Uint64("curResolved-ts", curResolvedTs),
+						zap.Uint64("lastResolved-ts", lastResolvedTs))
 				}
 			}
 		}
diff --git a/errors.toml b/errors.toml
index 043f2eb26..faf9fcf08 100755
--- a/errors.toml
+++ b/errors.toml
@@ -778,7 +778,7 @@ pulsar send message failed
 
 ["CDC:ErrReachMaxTry"]
 error = '''
-reach maximum try: %d
+reach maximum try: %s, error: %s
 '''
 
 ["CDC:ErrReactorFinished"]
diff --git a/metrics/grafana/ticdc.json b/metrics/grafana/ticdc.json
index be35ef317..d353cecb1 100644
--- a/metrics/grafana/ticdc.json
+++ b/metrics/grafana/ticdc.json
@@ -1121,11 +1121,11 @@
           "steppedLine": false,
           "targets": [
             {
-              "expr": "avg(ticdc_processor_min_resolved_table_id{k8s_cluster=\"k8s_cluster\", tidb_cluster=\"$tidb_cluster\", capture=~\"$capture\"}) by (capture)",
+              "expr": "ticdc_processor_min_resolved_table_id{k8s_cluster=\"k8s_cluster\", tidb_cluster=\"$tidb_cluster\", capture=~\"$capture\"})",
               "format": "time_series",
               "hide": false,
               "intervalFactor": 2,
-              "legendFormat": "{{capture}}-table-id",
+              "legendFormat": "{{capture}}-{{changefeed}}-table-id",
               "refId": "A",
               "step": 10
             }
@@ -1220,11 +1220,11 @@
           "steppedLine": false,
           "targets": [
             {
-              "expr": "avg(ticdc_processor_min_checkpoint_table_id{k8s_cluster=\"k8s_cluster\", tidb_cluster=\"$tidb_cluster\", capture=~\"$capture\"}) by (capture)",
+              "expr": "ticdc_processor_min_checkpoint_table_id{k8s_cluster=\"k8s_cluster\", tidb_cluster=\"$tidb_cluster\", capture=~\"$capture\"})",
               "format": "time_series",
               "hide": false,
               "intervalFactor": 2,
-              "legendFormat": "{{capture}}-table-id",
+              "legendFormat": "{{capture}}-{{changefeed}}-table-id",
               "refId": "A",
               "step": 10
             }
@@ -15232,4 +15232,4 @@
   "title": "Test-Cluster-TiCDC",
   "uid": "YiGL8hBZ1",
   "version": 34
-}
+}
\ No newline at end of file
diff --git a/pkg/api/internal/rest/request.go b/pkg/api/internal/rest/request.go
index c4d63c02f..37d663ee3 100644
--- a/pkg/api/internal/rest/request.go
+++ b/pkg/api/internal/rest/request.go
@@ -57,7 +57,7 @@ type Request struct {
 	// retry options
 	backoffBaseDelay time.Duration
 	backoffMaxDelay  time.Duration
-	maxRetries       int64
+	maxRetries       uint64
 
 	// output
 	err  error
@@ -187,8 +187,8 @@ func (r *Request) WithBackoffMaxDelay(delay time.Duration) *Request {
 	return r
 }
 
-// MaxRetries specifies the maximum times a request will retry.
-func (r *Request) WithMaxRetries(maxRetries int64) *Request {
+// WithMaxRetries specifies the maximum times a request will retry.
+func (r *Request) WithMaxRetries(maxRetries uint64) *Request {
 	if r.err != nil {
 		return r
 	}
diff --git a/pkg/applier/redo.go b/pkg/applier/redo.go
index 28c705b84..59de74510 100644
--- a/pkg/applier/redo.go
+++ b/pkg/applier/redo.go
@@ -100,7 +100,7 @@ func (ra *RedoApplier) consumeLogs(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	log.Info("apply redo log starts", zap.Uint64("checkpoint-ts", checkpointTs), zap.Uint64("resolved-ts", resolvedTs))
+	log.Info("apply redo log starts", zap.Uint64("checkpointTs", checkpointTs), zap.Uint64("resolvedTs", resolvedTs))
 
 	// MySQL sink will use the following replication config
 	// - EnableOldValue: default true
diff --git a/pkg/cmd/cli/cli_unsafe.go b/pkg/cmd/cli/cli_unsafe.go
index 354fb3db9..4c0fe3e47 100644
--- a/pkg/cmd/cli/cli_unsafe.go
+++ b/pkg/cmd/cli/cli_unsafe.go
@@ -73,6 +73,7 @@ func newCmdUnsafe(f factory.Factory) *cobra.Command {
 	command.AddCommand(newCmdReset(f, commonOptions))
 	command.AddCommand(newCmdShowMetadata(f))
 	command.AddCommand(newCmdDeleteServiceGcSafepoint(f, commonOptions))
+	command.AddCommand(newCmdResolveLock(f))
 
 	return command
 }
diff --git a/pkg/cmd/cli/cli_unsafe_resolve_lock.go b/pkg/cmd/cli/cli_unsafe_resolve_lock.go
new file mode 100644
index 000000000..70820466a
--- /dev/null
+++ b/pkg/cmd/cli/cli_unsafe_resolve_lock.go
@@ -0,0 +1,113 @@
+// Copyright 2022 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package cli
+
+import (
+	"time"
+
+	"github.com/pingcap/log"
+	"github.com/pingcap/tidb/kv"
+	"github.com/pingcap/tiflow/pkg/cmd/context"
+	"github.com/pingcap/tiflow/pkg/cmd/factory"
+	"github.com/pingcap/tiflow/pkg/txnutil"
+	"github.com/spf13/cobra"
+	"github.com/tikv/client-go/v2/oracle"
+	"github.com/tikv/client-go/v2/tikv"
+)
+
+// unsafeResolveLockOptions defines flags for the `cli unsafe show-metadata` command.
+type unsafeResolveLockOptions struct {
+	kvStorage kv.Storage
+
+	regionID uint64
+	ts       uint64
+}
+
+// newUnsafeResolveLockOptions creates new unsafeResolveLockOptions
+// for the `cli unsafe show-metadata` command.
+func newUnsafeResolveLockOptions() *unsafeResolveLockOptions {
+	return &unsafeResolveLockOptions{}
+}
+
+// complete adapts from the command line args to the data and client required.
+func (o *unsafeResolveLockOptions) complete(f factory.Factory) error {
+	ctx := context.GetDefaultContext()
+
+	pdClient, err := f.PdClient()
+	if err != nil {
+		return err
+	}
+	if o.ts == 0 {
+		ts, logic, err := pdClient.GetTS(ctx)
+		if err != nil {
+			return err
+		}
+		now := oracle.GetTimeFromTS(oracle.ComposeTS(ts, logic))
+		// Try not kill active transaction, we only resolves lock 1 minute ago.
+		o.ts = oracle.GoTimeToTS(now.Add(-time.Minute))
+	}
+
+	o.kvStorage, err = f.KvStorage()
+	return err
+}
+
+// run runs the `cli unsafe show-metadata` command.
+func (o *unsafeResolveLockOptions) run() error {
+	ctx := context.GetDefaultContext()
+
+	conf := &log.Config{Level: "info", File: log.FileLogConfig{}}
+	lg, p, e := log.InitLogger(conf)
+	if e != nil {
+		return e
+	}
+	log.ReplaceGlobals(lg, p)
+	txnResolver := txnutil.NewLockerResolver(o.kvStorage.(tikv.Storage))
+	return txnResolver.Resolve(ctx, o.regionID, o.ts)
+}
+
+// addFlags receives a *cobra.Command reference and binds
+// flags related to template printing to it.
+func (o *unsafeResolveLockOptions) addFlags(cmd *cobra.Command) {
+	if o == nil {
+		return
+	}
+
+	cmd.Flags().Uint64Var(&o.regionID, "region", 0, "Region ID")
+	cmd.Flags().Uint64Var(&o.ts, "ts", 0,
+		"resolve locks before the timestamp, default 1 minute ago from now")
+	_ = cmd.MarkFlagRequired("region")
+}
+
+// newCmdResolveLock creates the `cli unsafe show-metadata` command.
+func newCmdResolveLock(f factory.Factory) *cobra.Command {
+	o := newUnsafeResolveLockOptions()
+
+	command := &cobra.Command{
+		Use:   "resolve-lock",
+		Short: "resolve locks in regions",
+		Args:  cobra.NoArgs,
+		RunE: func(_ *cobra.Command, args []string) error {
+			err := o.complete(f)
+			if err != nil {
+				return err
+			}
+
+			return o.run()
+		},
+	}
+
+	o.addFlags(command)
+
+	return command
+}
diff --git a/pkg/cmd/factory/factory.go b/pkg/cmd/factory/factory.go
index 48eb31be6..b1f07054c 100644
--- a/pkg/cmd/factory/factory.go
+++ b/pkg/cmd/factory/factory.go
@@ -17,6 +17,7 @@ import (
 	"crypto/tls"
 
 	"github.com/pingcap/errors"
+	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tiflow/pkg/cmd/util"
 	"github.com/pingcap/tiflow/pkg/etcd"
 	"github.com/pingcap/tiflow/pkg/security"
@@ -30,6 +31,7 @@ type Factory interface {
 	ClientGetter
 	EtcdClient() (*etcd.CDCEtcdClient, error)
 	PdClient() (pd.Client, error)
+	KvStorage() (kv.Storage, error)
 }
 
 // ClientGetter defines the client getter.
diff --git a/pkg/cmd/factory/factory_impl.go b/pkg/cmd/factory/factory_impl.go
index fc4e9c775..812d9406d 100644
--- a/pkg/cmd/factory/factory_impl.go
+++ b/pkg/cmd/factory/factory_impl.go
@@ -19,6 +19,8 @@ import (
 	"time"
 
 	"github.com/pingcap/errors"
+	tidbkv "github.com/pingcap/tidb/kv"
+	"github.com/pingcap/tiflow/cdc/kv"
 	cmdconetxt "github.com/pingcap/tiflow/pkg/cmd/context"
 	"github.com/pingcap/tiflow/pkg/etcd"
 	"github.com/pingcap/tiflow/pkg/security"
@@ -166,3 +168,14 @@ func (f factoryImpl) PdClient() (pd.Client, error) {
 
 	return pdClient, nil
 }
+
+func (f factoryImpl) KvStorage() (tidbkv.Storage, error) {
+	pdAddr := f.GetPdAddr()
+	credential := f.GetCredential()
+	kvStore, err := kv.CreateTiStore(pdAddr, credential)
+	if err != nil {
+		return nil, errors.Annotatef(err,
+			"fail to open KV storage client, please check pd address \"%s\"", pdAddr)
+	}
+	return kvStore, nil
+}
diff --git a/pkg/config/kvclient.go b/pkg/config/kvclient.go
index 0df0261c3..9b28ef16d 100644
--- a/pkg/config/kvclient.go
+++ b/pkg/config/kvclient.go
@@ -13,6 +13,8 @@
 
 package config
 
+import cerror "github.com/pingcap/tiflow/pkg/errors"
+
 // KVClientConfig represents config for kv client
 type KVClientConfig struct {
 	// how many workers will be used for a single region worker
@@ -21,4 +23,23 @@ type KVClientConfig struct {
 	WorkerPoolSize int `toml:"worker-pool-size" json:"worker-pool-size"`
 	// region incremental scan limit for one table in a single store
 	RegionScanLimit int `toml:"region-scan-limit" json:"region-scan-limit"`
+	// the total retry duration of connecting a region
+	RegionRetryDuration TomlDuration `toml:"region-retry-duration" json:"region-retry-duration"`
+}
+
+// ValidateAndAdjust validates and adjusts the kv client configuration
+func (c *KVClientConfig) ValidateAndAdjust() error {
+	if c.WorkerConcurrent <= 0 {
+		return cerror.ErrInvalidServerOption.GenWithStackByArgs(
+			"region-scan-limit should be at least 1")
+	}
+	if c.RegionScanLimit <= 0 {
+		return cerror.ErrInvalidServerOption.GenWithStackByArgs(
+			"region-scan-limit should be at least 1")
+	}
+	if c.RegionRetryDuration <= 0 {
+		return cerror.ErrInvalidServerOption.GenWithStackByArgs(
+			"region-scan-limit should be positive")
+	}
+	return nil
 }
diff --git a/pkg/config/server_config.go b/pkg/config/server_config.go
index 4873f7409..5b11ec35f 100644
--- a/pkg/config/server_config.go
+++ b/pkg/config/server_config.go
@@ -96,6 +96,9 @@ var defaultServerConfig = &ServerConfig{
 		WorkerConcurrent: 8,
 		WorkerPoolSize:   0, // 0 will use NumCPU() * 2
 		RegionScanLimit:  40,
+		// The default TiKV region election timeout is [10s, 20s],
+		// Use 1 minute to cover region leader missing.
+		RegionRetryDuration: TomlDuration(time.Minute),
 	},
 	Debug: &DebugConfig{
 		EnableTableActor:   false,
@@ -247,11 +250,8 @@ func (c *ServerConfig) ValidateAndAdjust() error {
 	if c.KVClient == nil {
 		c.KVClient = defaultCfg.KVClient
 	}
-	if c.KVClient.WorkerConcurrent <= 0 {
-		return cerror.ErrInvalidServerOption.GenWithStackByArgs("region-scan-limit should be at least 1")
-	}
-	if c.KVClient.RegionScanLimit <= 0 {
-		return cerror.ErrInvalidServerOption.GenWithStackByArgs("region-scan-limit should be at least 1")
+	if err = c.KVClient.ValidateAndAdjust(); err != nil {
+		return errors.Trace(err)
 	}
 
 	if c.Debug == nil {
diff --git a/pkg/errors/errors.go b/pkg/errors/errors.go
index 48efe0588..c272e84a9 100644
--- a/pkg/errors/errors.go
+++ b/pkg/errors/errors.go
@@ -261,7 +261,9 @@ var (
 	ErrFlowControllerEventLargerThanQuota = errors.Normalize("event is larger than the total memory quota, size: %d, quota: %d", errors.RFCCodeText("CDC:ErrFlowControllerEventLargerThanQuota"))
 
 	// retry error
-	ErrReachMaxTry = errors.Normalize("reach maximum try: %d", errors.RFCCodeText("CDC:ErrReachMaxTry"))
+	ErrReachMaxTry = errors.Normalize("reach maximum try: %s, error: %s",
+		errors.RFCCodeText("CDC:ErrReachMaxTry"),
+	)
 
 	// tcp server error
 	ErrTCPServerClosed = errors.Normalize("The TCP server has been closed", errors.RFCCodeText("CDC:ErrTCPServerClosed"))
diff --git a/pkg/etcd/client.go b/pkg/etcd/client.go
index 692e2da1e..4296ac484 100644
--- a/pkg/etcd/client.go
+++ b/pkg/etcd/client.go
@@ -64,7 +64,7 @@ var (
 )
 
 // set to var instead of const for mocking the value to speedup test
-var maxTries int64 = 8
+var maxTries uint64 = 8
 
 // Client is a simple wrapper that adds retry to etcd RPC
 type Client struct {
diff --git a/pkg/orchestrator/reactor_state.go b/pkg/orchestrator/reactor_state.go
index f809761fb..82f6c0dbe 100644
--- a/pkg/orchestrator/reactor_state.go
+++ b/pkg/orchestrator/reactor_state.go
@@ -66,7 +66,7 @@ func (s *GlobalReactorState) Update(key util.EtcdKey, value []byte, _ bool) erro
 	case etcd.CDCKeyTypeCapture:
 		if value == nil {
 			log.Info("remote capture offline",
-				zap.String("capture-id", k.CaptureID),
+				zap.String("captureID", k.CaptureID),
 				zap.Any("info", s.Captures[k.CaptureID]))
 			delete(s.Captures, k.CaptureID)
 			if s.onCaptureRemoved != nil {
@@ -81,7 +81,8 @@ func (s *GlobalReactorState) Update(key util.EtcdKey, value []byte, _ bool) erro
 			return cerrors.ErrUnmarshalFailed.Wrap(err).GenWithStackByArgs()
 		}
 
-		log.Info("remote capture online", zap.String("capture-id", k.CaptureID), zap.Any("info", newCaptureInfo))
+		log.Info("remote capture online",
+			zap.String("captureID", k.CaptureID), zap.Any("info", newCaptureInfo))
 		if s.onCaptureAdded != nil {
 			s.onCaptureAdded(k.CaptureID, newCaptureInfo.AdvertiseAddr)
 		}
diff --git a/pkg/p2p/client.go b/pkg/p2p/client.go
index 4b1d9c667..ba4f1a624 100644
--- a/pkg/p2p/client.go
+++ b/pkg/p2p/client.go
@@ -115,7 +115,7 @@ func (c *MessageClient) Run(
 
 		log.Info("peer message client exited",
 			zap.String("addr", addr),
-			zap.String("capture-id", receiverID),
+			zap.String("captureID", receiverID),
 			zap.Error(ret))
 	}()
 
diff --git a/pkg/p2p/message_router.go b/pkg/p2p/message_router.go
index dd9ac0114..732bc8b99 100644
--- a/pkg/p2p/message_router.go
+++ b/pkg/p2p/message_router.go
@@ -144,7 +144,7 @@ func (m *messageRouterImpl) GetClient(target NodeID) *MessageClient {
 		err := client.Run(ctx, "tcp", addr, target, m.credentials)
 		log.Warn("p2p client exited with error",
 			zap.String("addr", addr),
-			zap.String("target-capture", target),
+			zap.String("targetCapture", target),
 			zap.Error(err))
 
 		if errors.Cause(err) != context.Canceled {
diff --git a/pkg/p2p/server.go b/pkg/p2p/server.go
index a91f03730..c5ff06fc9 100755
--- a/pkg/p2p/server.go
+++ b/pkg/p2p/server.go
@@ -100,7 +100,7 @@ func newCDCPeer(senderID NodeID, epoch int64, sender *streamHandle) *cdcPeer {
 func (p *cdcPeer) abortWithError(ctx context.Context, err error) {
 	if err1 := p.sender.Send(ctx, errorToRPCResponse(err)); err1 != nil {
 		log.Warn("could not send error to peer", zap.Error(err),
-			zap.NamedError("send-err", err1))
+			zap.NamedError("sendErr", err1))
 		return
 	}
 	log.Debug("sent error to peer", zap.Error(err))
@@ -401,10 +401,10 @@ func (m *MessageServer) AddHandler(
 			}).Inc()
 
 			log.Debug("skipping peer message",
-				zap.String("sender-id", sm.SenderId),
+				zap.String("senderID", sm.SenderId),
 				zap.String("topic", topic),
-				zap.Int64("skipped-Seq", entry.Sequence),
-				zap.Int64("last-ack", lastAck))
+				zap.Int64("skippedSeq", entry.Sequence),
+				zap.Int64("lastAck", lastAck))
 			return nil
 		}
 
@@ -533,8 +533,8 @@ func (m *MessageServer) registerPeer(
 	streamMeta := sender.GetStreamMeta()
 
 	log.Info("peer connection received",
-		zap.String("sender-id", streamMeta.SenderId),
-		zap.String("sender-advertise-addr", streamMeta.SenderAdvertisedAddr),
+		zap.String("senderID", streamMeta.SenderId),
+		zap.String("senderAdvertiseAddr", streamMeta.SenderAdvertisedAddr),
 		zap.String("addr", clientIP),
 		zap.Int64("epoch", streamMeta.Epoch))
 
@@ -554,7 +554,7 @@ func (m *MessageServer) registerPeer(
 		// there is an existing peer
 		if peer.Epoch > streamMeta.Epoch {
 			log.Warn("incoming connection is stale",
-				zap.String("sender-id", streamMeta.SenderId),
+				zap.String("senderID", streamMeta.SenderId),
 				zap.String("addr", clientIP),
 				zap.Int64("epoch", streamMeta.Epoch))
 
@@ -568,7 +568,7 @@ func (m *MessageServer) registerPeer(
 			m.peerLock.Unlock()
 		} else {
 			log.Warn("incoming connection is duplicate",
-				zap.String("sender-id", streamMeta.SenderId),
+				zap.String("senderID", streamMeta.SenderId),
 				zap.String("addr", clientIP),
 				zap.Int64("epoch", streamMeta.Epoch))
 
@@ -649,8 +649,8 @@ func (m *MessageServer) SendMessage(stream p2p.CDCPeerToPeer_SendMessageServer)
 			case resp, ok := <-sendCh:
 				if !ok {
 					log.Info("peer stream handle is closed",
-						zap.String("peer-addr", streamHandle.GetStreamMeta().SenderAdvertisedAddr),
-						zap.String("peer-id", streamHandle.GetStreamMeta().SenderId))
+						zap.String("peerAddr", streamHandle.GetStreamMeta().SenderAdvertisedAddr),
+						zap.String("peerID", streamHandle.GetStreamMeta().SenderId))
 					// cancel the stream when sendCh is closed
 					cancel()
 					return nil
@@ -724,7 +724,7 @@ func (m *MessageServer) receive(stream p2p.CDCPeerToPeer_SendMessageServer, stre
 
 		batchSize := len(packet.GetEntries())
 		log.Debug("received packet", zap.String("streamHandle", streamHandle.GetStreamMeta().SenderId),
-			zap.Int("num-entries", batchSize))
+			zap.Int("numEntries", batchSize))
 
 		batchBytes := packet.Size()
 		metricsServerMessageBatchBytesHistogram.Observe(float64(batchBytes))
diff --git a/pkg/pdutil/region_label.go b/pkg/pdutil/region_label.go
index e7a53c4fd..48d9f9593 100644
--- a/pkg/pdutil/region_label.go
+++ b/pkg/pdutil/region_label.go
@@ -55,7 +55,7 @@ const (
 	}`
 )
 
-var defaultMaxRetry int64 = 3
+var defaultMaxRetry uint64 = 3
 
 // pdAPIClient is api client of Placement Driver.
 type pdAPIClient struct {
diff --git a/pkg/pipeline/runner.go b/pkg/pipeline/runner.go
index 3de79a3c6..d77ba0cb5 100644
--- a/pkg/pipeline/runner.go
+++ b/pkg/pipeline/runner.go
@@ -51,7 +51,8 @@ func (r *nodeRunner) run(ctx context.Context) error {
 	defer func() {
 		err := r.node.Destroy(nodeCtx)
 		if err != nil && errors.Cause(err) != stdContext.Canceled {
-			log.Error("found an error when stopping node", zap.String("node name", r.name), zap.Error(err))
+			log.Error("found an error when stopping node",
+				zap.String("node", r.name), zap.Error(err))
 		}
 	}()
 	err := r.node.Init(nodeCtx)
diff --git a/pkg/regionspan/span.go b/pkg/regionspan/span.go
index 827252599..205ed202c 100644
--- a/pkg/regionspan/span.go
+++ b/pkg/regionspan/span.go
@@ -209,7 +209,7 @@ func Intersect(lhs ComparableSpan, rhs ComparableSpan) (span ComparableSpan, err
 // IsSubSpan returns true if the sub span is parents spans
 func IsSubSpan(sub ComparableSpan, parents ...ComparableSpan) bool {
 	if bytes.Compare(sub.Start, sub.End) >= 0 {
-		log.Panic("the sub span is invalid", zap.Reflect("sub span", sub))
+		log.Panic("the sub span is invalid", zap.Reflect("subSpan", sub))
 	}
 	for _, parent := range parents {
 		if StartCompare(parent.Start, sub.Start) <= 0 &&
diff --git a/pkg/retry/options.go b/pkg/retry/options.go
index 724195a3f..942f04025 100644
--- a/pkg/retry/options.go
+++ b/pkg/retry/options.go
@@ -15,14 +15,16 @@ package retry
 
 import (
 	"math"
+	"time"
 )
 
 const (
 	// defaultBackoffBaseInMs is the initial duration, in Millisecond
 	defaultBackoffBaseInMs = 10.0
 	// defaultBackoffCapInMs is the max amount of duration, in Millisecond
-	defaultBackoffCapInMs = 100.0
-	defaultMaxTries       = 3
+	defaultBackoffCapInMs   = 100.0
+	defaultMaxTries         = math.MaxUint64
+	defaultMaxRetryDuration = time.Duration(0)
 )
 
 // Option ...
@@ -32,18 +34,20 @@ type Option func(*retryOptions)
 type IsRetryable func(error) bool
 
 type retryOptions struct {
-	maxTries        int64
-	backoffBaseInMs float64
-	backoffCapInMs  float64
-	isRetryable     IsRetryable
+	totalRetryDuration time.Duration
+	maxTries           uint64
+	backoffBaseInMs    float64
+	backoffCapInMs     float64
+	isRetryable        IsRetryable
 }
 
 func newRetryOptions() *retryOptions {
 	return &retryOptions{
-		maxTries:        defaultMaxTries,
-		backoffBaseInMs: defaultBackoffBaseInMs,
-		backoffCapInMs:  defaultBackoffCapInMs,
-		isRetryable:     func(err error) bool { return true },
+		totalRetryDuration: defaultMaxRetryDuration,
+		maxTries:           defaultMaxTries,
+		backoffBaseInMs:    defaultBackoffBaseInMs,
+		backoffCapInMs:     defaultBackoffCapInMs,
+		isRetryable:        func(err error) bool { return true },
 	}
 }
 
@@ -65,19 +69,20 @@ func WithBackoffMaxDelay(delayInMs int64) Option {
 	}
 }
 
-// WithMaxTries configures maximum tries, if tries <= 0 "defaultMaxTries" will be used
-func WithMaxTries(tries int64) Option {
+// WithMaxTries configures maximum tries, if tries is 0, 1 will be used
+func WithMaxTries(tries uint64) Option {
 	return func(o *retryOptions) {
-		if tries > 0 {
-			o.maxTries = tries
+		if tries == 0 {
+			tries = 1
 		}
+		o.maxTries = tries
 	}
 }
 
-// WithInfiniteTries configures to retry forever (math.MaxInt64 times) till success or got canceled
-func WithInfiniteTries() Option {
+// WithTotalRetryDuratoin configures the total retry duration.
+func WithTotalRetryDuratoin(retryDuration time.Duration) Option {
 	return func(o *retryOptions) {
-		o.maxTries = math.MaxInt64
+		o.totalRetryDuration = retryDuration
 	}
 }
 
diff --git a/pkg/retry/retry_with_opt.go b/pkg/retry/retry_with_opt.go
index c5482ebe1..8f6bc1250 100644
--- a/pkg/retry/retry_with_opt.go
+++ b/pkg/retry/retry_with_opt.go
@@ -17,6 +17,7 @@ import (
 	"context"
 	"math"
 	"math/rand"
+	"strconv"
 	"time"
 
 	"github.com/pingcap/errors"
@@ -26,7 +27,8 @@ import (
 // Operation is the action need to retry
 type Operation func() error
 
-// Do execute the specified function at most maxTries times until it succeeds or got canceled
+// Do execute the specified function.
+// By default, it retries infinitely until it succeeds or got canceled.
 func Do(ctx context.Context, operation Operation, opts ...Option) error {
 	retryOption := setOptions(opts...)
 	return run(ctx, operation, retryOption)
@@ -48,7 +50,8 @@ func run(ctx context.Context, op Operation, retryOption *retryOptions) error {
 	}
 
 	var t *time.Timer
-	try := 0
+	var start time.Time
+	try := uint64(0)
 	backOff := time.Duration(0)
 	for {
 		err := op()
@@ -61,8 +64,17 @@ func run(ctx context.Context, op Operation, retryOption *retryOptions) error {
 		}
 
 		try++
-		if int64(try) >= retryOption.maxTries {
-			return cerror.ErrReachMaxTry.Wrap(err).GenWithStackByArgs(retryOption.maxTries)
+		if try >= retryOption.maxTries {
+			return cerror.ErrReachMaxTry.
+				Wrap(err).GenWithStackByArgs(strconv.Itoa(int(retryOption.maxTries)), err)
+		}
+		if retryOption.totalRetryDuration > 0 {
+			if start.IsZero() {
+				start = time.Now()
+			} else if time.Since(start) > retryOption.totalRetryDuration {
+				return cerror.ErrReachMaxTry.
+					Wrap(err).GenWithStackByArgs(retryOption.totalRetryDuration, err)
+			}
 		}
 
 		backOff = getBackoffInMs(retryOption.backoffBaseInMs, retryOption.backoffCapInMs, float64(try))
diff --git a/pkg/txnutil/lock_resolver.go b/pkg/txnutil/lock_resolver.go
index bead096a0..82b03abb8 100644
--- a/pkg/txnutil/lock_resolver.go
+++ b/pkg/txnutil/lock_resolver.go
@@ -54,6 +54,7 @@ func (r *resolver) Resolve(ctx context.Context, regionID uint64, maxVersion uint
 	})
 
 	bo := tikv.NewGcResolveLockMaxBackoffer(ctx)
+	var lockCount int
 	var loc *tikv.KeyLocation
 	var key []byte
 	flushRegion := func() error {
@@ -105,6 +106,7 @@ func (r *resolver) Resolve(ctx context.Context, regionID uint64, maxVersion uint
 		for i := range locksInfo {
 			locks[i] = txnkv.NewLock(locksInfo[i])
 		}
+		lockCount += len(locksInfo)
 
 		_, err1 := r.kvStorage.GetLockResolver().ResolveLocks(bo, 0, locks)
 		if err1 != nil {
@@ -121,6 +123,9 @@ func (r *resolver) Resolve(ctx context.Context, regionID uint64, maxVersion uint
 		}
 		bo = tikv.NewGcResolveLockMaxBackoffer(ctx)
 	}
-	log.Info("resolve lock successfully", zap.Uint64("regionID", regionID), zap.Uint64("maxVersion", maxVersion))
+	log.Info("resolve lock successfully",
+		zap.Uint64("regionID", regionID),
+		zap.Int("lockCount", lockCount),
+		zap.Uint64("maxVersion", maxVersion))
 	return nil
 }
diff --git a/pkg/util/cancel_monitor.go b/pkg/util/cancel_monitor.go
index 8f64a6f53..fb4953bc9 100644
--- a/pkg/util/cancel_monitor.go
+++ b/pkg/util/cancel_monitor.go
@@ -38,7 +38,9 @@ func MonitorCancelLatency(ctx context.Context, identifier string) func() {
 					return
 				case <-ticker.C:
 					elapsed++
-					log.Warn("MonitorCancelLatency: Cancellation is taking too long", zap.String("identifier", identifier), zap.Int("elapsed seconds", elapsed), zap.Error(ctx.Err()))
+					log.Warn("MonitorCancelLatency: Cancellation is taking too long",
+						zap.String("identifier", identifier),
+						zap.Int("duration", elapsed), zap.Error(ctx.Err()))
 				}
 			}
 		case <-finishedCh:
diff --git a/pkg/workerpool/pool_impl.go b/pkg/workerpool/pool_impl.go
index b8d82f8d5..578532f99 100644
--- a/pkg/workerpool/pool_impl.go
+++ b/pkg/workerpool/pool_impl.go
@@ -413,7 +413,7 @@ func (w *worker) synchronize() {
 			// likely the workerpool has deadlocked, or there is a bug
 			// in the event handlers.
 			logWarn("synchronize is taking too long, report a bug",
-				zap.Duration("elapsed", time.Since(startTime)),
+				zap.Duration("duration", time.Since(startTime)),
 				zap.Stack("stacktrace"))
 		}
 	}
diff --git a/scripts/check-log-style.sh b/scripts/check-log-style.sh
new file mode 100755
index 000000000..89b820da6
--- /dev/null
+++ b/scripts/check-log-style.sh
@@ -0,0 +1,21 @@
+# Copyright 2022 PingCAP, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# zap field name should be camelCase, excepts for idioms and special terms.
+grep -RnE "zap.[A-Z][a-z]+\(\"[0-9A-Za-z]*[-_ ][0-9A-Za-z]*\"(,|\))" cdc tests pkg |
+	grep -vE "user-agent" |
+	grep -vE "https_proxy|http_proxy|no_proxy" |
+	grep -vE "max-message-bytes|max-message-size" |
+	grep -vE "release-version|git-hash|git-branch|go-version" |
+	grep -vE "failpoint-build|utc-build-time" |
+	awk '{ print  } END { if (NR > 0) { exit 1  }  }'


Test patch:
diff --git a/cdc/kv/client_bench_test.go b/cdc/kv/client_bench_test.go
index a6b6647ec..8cc250268 100644
--- a/cdc/kv/client_bench_test.go
+++ b/cdc/kv/client_bench_test.go
@@ -26,6 +26,7 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/store/mockstore/mockcopr"
 	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/pingcap/tiflow/pkg/config"
 	"github.com/pingcap/tiflow/pkg/regionspan"
 	"github.com/pingcap/tiflow/pkg/retry"
 	"github.com/pingcap/tiflow/pkg/security"
@@ -192,11 +193,11 @@ func prepareBenchMultiStore(b *testing.B, storeNum, regionNum int) (
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 1000000)
 	wg.Add(1)
 	go func() {
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, false, lockresolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, lockresolver, isPullInit, eventCh)
 		if errors.Cause(err) != context.Canceled {
 			b.Error(err)
 		}
@@ -283,11 +284,11 @@ func prepareBench(b *testing.B, regionNum int) (
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 1000000)
 	wg.Add(1)
 	go func() {
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("z")}, 100, false, lockresolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("z")}, 100, lockresolver, isPullInit, eventCh)
 		if errors.Cause(err) != context.Canceled {
 			b.Error(err)
 		}
diff --git a/cdc/kv/client_test.go b/cdc/kv/client_test.go
index 057799318..c4566f11d 100644
--- a/cdc/kv/client_test.go
+++ b/cdc/kv/client_test.go
@@ -93,7 +93,7 @@ func (s *clientSuite) TestNewClient(c *check.C) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cli := NewCDCClient(context.Background(), pdClient, nil, grpcPool, regionCache)
+	cli := NewCDCClient(context.Background(), pdClient, nil, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	c.Assert(cli, check.NotNil)
 }
 
@@ -101,11 +101,10 @@ func (s *clientSuite) TestAssembleRowEvent(c *check.C) {
 	defer testleak.AfterTest(c)()
 	defer s.TearDownTest(c)
 	testCases := []struct {
-		regionID       uint64
-		entry          *cdcpb.Event_Row
-		enableOldValue bool
-		expected       model.RegionFeedEvent
-		err            string
+		regionID uint64
+		entry    *cdcpb.Event_Row
+		expected model.RegionFeedEvent
+		err      string
 	}{{
 		regionID: 1,
 		entry: &cdcpb.Event_Row{
@@ -115,7 +114,6 @@ func (s *clientSuite) TestAssembleRowEvent(c *check.C) {
 			Value:    []byte("v1"),
 			OpType:   cdcpb.Event_Row_PUT,
 		},
-		enableOldValue: false,
 		expected: model.RegionFeedEvent{
 			RegionID: 1,
 			Val: &model.RawKVEntry{
@@ -136,7 +134,6 @@ func (s *clientSuite) TestAssembleRowEvent(c *check.C) {
 			Value:    []byte("v2"),
 			OpType:   cdcpb.Event_Row_DELETE,
 		},
-		enableOldValue: false,
 		expected: model.RegionFeedEvent{
 			RegionID: 2,
 			Val: &model.RawKVEntry{
@@ -148,28 +145,6 @@ func (s *clientSuite) TestAssembleRowEvent(c *check.C) {
 				RegionID: 2,
 			},
 		},
-	}, {
-		regionID: 3,
-		entry: &cdcpb.Event_Row{
-			StartTs:  1,
-			CommitTs: 2,
-			Key:      []byte("k2"),
-			Value:    []byte("v2"),
-			OldValue: []byte("ov2"),
-			OpType:   cdcpb.Event_Row_PUT,
-		},
-		enableOldValue: false,
-		expected: model.RegionFeedEvent{
-			RegionID: 3,
-			Val: &model.RawKVEntry{
-				OpType:   model.OpTypePut,
-				StartTs:  1,
-				CRTs:     2,
-				Key:      []byte("k2"),
-				Value:    []byte("v2"),
-				RegionID: 3,
-			},
-		},
 	}, {
 		regionID: 4,
 		entry: &cdcpb.Event_Row{
@@ -180,7 +155,6 @@ func (s *clientSuite) TestAssembleRowEvent(c *check.C) {
 			OldValue: []byte("ov3"),
 			OpType:   cdcpb.Event_Row_PUT,
 		},
-		enableOldValue: true,
 		expected: model.RegionFeedEvent{
 			RegionID: 4,
 			Val: &model.RawKVEntry{
@@ -202,12 +176,12 @@ func (s *clientSuite) TestAssembleRowEvent(c *check.C) {
 			Value:    []byte("v2"),
 			OpType:   cdcpb.Event_Row_UNKNOWN,
 		},
-		enableOldValue: false,
-		err:            "[CDC:ErrUnknownKVEventType]unknown kv optype: UNKNOWN, entry: start_ts:1 commit_ts:2 key:\"k2\" value:\"v2\" ",
+		err: "[CDC:ErrUnknownKVEventType]unknown kv optype: UNKNOWN, entry: start_ts:1 " +
+			"commit_ts:2 key:\"k2\" value:\"v2\" ",
 	}}
 
 	for _, tc := range testCases {
-		event, err := assembleRowEvent(tc.regionID, tc.entry, tc.enableOldValue)
+		event, err := assembleRowEvent(tc.regionID, tc.entry)
 		c.Assert(event, check.DeepEquals, tc.expected)
 		if err != nil {
 			c.Assert(err.Error(), check.Equals, tc.err)
@@ -366,12 +340,12 @@ func (s *clientSuite) TestConnectOfflineTiKV(c *check.C) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(context.Background(), pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(context.Background(), pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 10)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 1, false, lockresolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 1, lockresolver, isPullInit, eventCh)
 		c.Assert(errors.Cause(err), check.Equals, context.Canceled)
 	}()
 
@@ -466,12 +440,12 @@ func (s *clientSuite) TestRecvLargeMessageSize(c *check.C) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 10)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 1, false, lockresolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 1, lockresolver, isPullInit, eventCh)
 		c.Assert(errors.Cause(err), check.Equals, context.Canceled)
 	}()
 
@@ -565,12 +539,12 @@ func (s *clientSuite) TestHandleError(c *check.C) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 10)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("d")}, 100, false, lockresolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("d")}, 100, lockresolver, isPullInit, eventCh)
 		c.Assert(errors.Cause(err), check.Equals, context.Canceled)
 	}()
 
@@ -723,13 +697,13 @@ func (s *clientSuite) TestCompatibilityWithSameConn(c *check.C) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 10)
 	var wg2 sync.WaitGroup
 	wg2.Add(1)
 	go func() {
 		defer wg2.Done()
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, false, lockResolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, lockResolver, isPullInit, eventCh)
 		c.Assert(cerror.ErrVersionIncompatible.Equal(err), check.IsTrue)
 	}()
 
@@ -792,14 +766,14 @@ func (s *clientSuite) TestClusterIDMismatch(c *check.C) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 10)
 
 	var wg2 sync.WaitGroup
 	wg2.Add(1)
 	go func() {
 		defer wg2.Done()
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, false, lockResolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, lockResolver, isPullInit, eventCh)
 		c.Assert(cerror.ErrClusterIDMismatch.Equal(err), check.IsTrue)
 	}()
 
@@ -859,12 +833,12 @@ func (s *clientSuite) testHandleFeedEvent(c *check.C) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 10)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, false, lockresolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, lockresolver, isPullInit, eventCh)
 		c.Assert(errors.Cause(err), check.Equals, context.Canceled)
 	}()
 
@@ -1309,12 +1283,12 @@ func (s *clientSuite) TestStreamSendWithError(c *check.C) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 10)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("c")}, 100, false, lockerResolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("c")}, 100, lockerResolver, isPullInit, eventCh)
 		c.Assert(errors.Cause(err), check.Equals, context.Canceled)
 	}()
 
@@ -1420,12 +1394,12 @@ func (s *clientSuite) testStreamRecvWithError(c *check.C, failpointStr string) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 40)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, false, lockresolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, lockresolver, isPullInit, eventCh)
 		c.Assert(errors.Cause(err), check.Equals, context.Canceled)
 	}()
 
@@ -1550,13 +1524,13 @@ func (s *clientSuite) TestStreamRecvWithErrorAndResolvedGoBack(c *check.C) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 10)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
 		defer close(eventCh)
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, false, lockresolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, lockresolver, isPullInit, eventCh)
 		c.Assert(errors.Cause(err), check.Equals, context.Canceled)
 	}()
 
@@ -1760,13 +1734,13 @@ func (s *clientSuite) TestIncompatibleTiKV(c *check.C) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	// NOTICE: eventCh may block the main logic of EventFeed
 	eventCh := make(chan model.RegionFeedEvent, 128)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, false, lockresolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, lockresolver, isPullInit, eventCh)
 		c.Assert(errors.Cause(err), check.Equals, context.Canceled)
 	}()
 
@@ -1837,13 +1811,13 @@ func (s *clientSuite) TestNoPendingRegionError(c *check.C) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 10)
 
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, false, lockresolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, lockresolver, isPullInit, eventCh)
 		c.Assert(errors.Cause(err), check.Equals, context.Canceled)
 	}()
 
@@ -1915,12 +1889,12 @@ func (s *clientSuite) TestDropStaleRequest(c *check.C) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 10)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, false, lockresolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, lockresolver, isPullInit, eventCh)
 		c.Assert(errors.Cause(err), check.Equals, context.Canceled)
 	}()
 
@@ -2025,12 +1999,12 @@ func (s *clientSuite) TestResolveLock(c *check.C) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 10)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, false, lockresolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, lockresolver, isPullInit, eventCh)
 		c.Assert(errors.Cause(err), check.Equals, context.Canceled)
 	}()
 
@@ -2125,13 +2099,13 @@ func (s *clientSuite) testEventCommitTsFallback(c *check.C, events []*cdcpb.Chan
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 10)
 	var clientWg sync.WaitGroup
 	clientWg.Add(1)
 	go func() {
 		defer clientWg.Done()
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, false, lockresolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, lockresolver, isPullInit, eventCh)
 		c.Assert(err, check.Equals, errUnreachable)
 	}()
 
@@ -2273,12 +2247,12 @@ func (s *clientSuite) testEventAfterFeedStop(c *check.C) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 10)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, false, lockresolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, lockresolver, isPullInit, eventCh)
 		c.Assert(errors.Cause(err), check.Equals, context.Canceled)
 	}()
 
@@ -2453,12 +2427,12 @@ func (s *clientSuite) TestOutOfRegionRangeEvent(c *check.C) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 10)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, false, lockresolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, lockresolver, isPullInit, eventCh)
 		c.Assert(errors.Cause(err), check.Equals, context.Canceled)
 	}()
 
@@ -2668,12 +2642,12 @@ func (s *clientSuite) TestResolveLockNoCandidate(c *check.C) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 10)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, false, lockresolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, lockresolver, isPullInit, eventCh)
 		c.Assert(errors.Cause(err), check.Equals, context.Canceled)
 	}()
 
@@ -2763,12 +2737,12 @@ func (s *clientSuite) TestFailRegionReentrant(c *check.C) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage.(tikv.Storage), grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage.(tikv.Storage), grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 10)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, false, lockresolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, lockresolver, isPullInit, eventCh)
 		c.Assert(errors.Cause(err), check.Equals, context.Canceled)
 	}()
 
@@ -2845,12 +2819,12 @@ func (s *clientSuite) TestClientV1UnlockRangeReentrant(c *check.C) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 10)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("c")}, 100, false, lockresolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("c")}, 100, lockresolver, isPullInit, eventCh)
 		c.Assert(errors.Cause(err), check.Equals, context.Canceled)
 	}()
 
@@ -2912,12 +2886,12 @@ func (s *clientSuite) testClientErrNoPendingRegion(c *check.C) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 10)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("c")}, 100, false, lockresolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("c")}, 100, lockresolver, isPullInit, eventCh)
 		c.Assert(errors.Cause(err), check.Equals, context.Canceled)
 	}()
 
@@ -2990,12 +2964,12 @@ func (s *clientSuite) testKVClientForceReconnect(c *check.C) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 10)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("c")}, 100, false, lockresolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("c")}, 100, lockresolver, isPullInit, eventCh)
 		c.Assert(errors.Cause(err), check.Equals, context.Canceled)
 	}()
 
@@ -3141,12 +3115,12 @@ func (s *clientSuite) TestConcurrentProcessRangeRequest(c *check.C) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 100)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("z")}, 100, false, lockresolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("z")}, 100, lockresolver, isPullInit, eventCh)
 		c.Assert(errors.Cause(err), check.Equals, context.Canceled)
 	}()
 
@@ -3257,12 +3231,12 @@ func (s *clientSuite) TestEvTimeUpdate(c *check.C) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 10)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, false, lockresolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, lockresolver, isPullInit, eventCh)
 		c.Assert(errors.Cause(err), check.Equals, context.Canceled)
 	}()
 
@@ -3377,12 +3351,12 @@ func (s *clientSuite) TestRegionWorkerExitWhenIsIdle(c *check.C) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 10)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, false, lockresolver, isPullInit, eventCh)
+		err := cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")}, 100, lockresolver, isPullInit, eventCh)
 		c.Assert(errors.Cause(err), check.Equals, context.Canceled)
 	}()
 
@@ -3469,14 +3443,14 @@ func (s *clientSuite) TestPrewriteNotMatchError(c *check.C) {
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdClient)
 	defer regionCache.Close()
-	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache)
+	cdcClient := NewCDCClient(ctx, pdClient, kvStorage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 	eventCh := make(chan model.RegionFeedEvent, 10)
 	baseAllocatedID := currentRequestID()
 
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		err = cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("c")}, 100, false, lockResolver, isPullInit, eventCh)
+		err = cdcClient.EventFeed(ctx, regionspan.ComparableSpan{Start: []byte("a"), End: []byte("c")}, 100, lockResolver, isPullInit, eventCh)
 		c.Assert(errors.Cause(err), check.Equals, context.Canceled)
 	}()
 
@@ -3549,14 +3523,13 @@ func (s *clientSuite) TestPrewriteNotMatchError(c *check.C) {
 
 func createFakeEventFeedSession(ctx context.Context) *eventFeedSession {
 	return newEventFeedSession(ctx,
-		&CDCClient{regionLimiters: defaultRegionEventFeedLimiters},
+		&CDCClient{regionLimiters: defaultRegionEventFeedLimiters, config: config.GetDefaultServerConfig().KVClient},
 		nil, /*regionCache*/
 		nil, /*kvStorage*/
 		regionspan.ComparableSpan{Start: []byte("a"), End: []byte("b")},
-		nil,   /*lockResolver*/
-		nil,   /*isPullerInit*/
-		false, /*enableOldValue*/
-		100,   /*startTs*/
+		nil, /*lockResolver*/
+		nil, /*isPullerInit*/
+		100, /*startTs*/
 		nil /*eventCh*/)
 }
 
diff --git a/cdc/kv/matcher_test.go b/cdc/kv/matcher_test.go
index fd9f745e7..8def1b11f 100644
--- a/cdc/kv/matcher_test.go
+++ b/cdc/kv/matcher_test.go
@@ -14,9 +14,12 @@
 package kv
 
 import (
+	"testing"
+
 	"github.com/pingcap/check"
 	"github.com/pingcap/kvproto/pkg/cdcpb"
 	"github.com/pingcap/tiflow/pkg/util/testleak"
+	"github.com/stretchr/testify/require"
 )
 
 type matcherSuite struct{}
@@ -47,7 +50,7 @@ func (s *matcherSuite) TestMatchRow(c *check.C) {
 		StartTs: 1,
 		Key:     []byte("k1"),
 	}
-	ok := matcher.matchRow(commitRow1)
+	ok := matcher.matchRow(commitRow1, true)
 	c.Assert(ok, check.IsFalse)
 	c.Assert(commitRow1, check.DeepEquals, &cdcpb.Event_Row{
 		StartTs: 1,
@@ -60,7 +63,7 @@ func (s *matcherSuite) TestMatchRow(c *check.C) {
 		CommitTs: 3,
 		Key:      []byte("k1"),
 	}
-	ok = matcher.matchRow(commitRow2)
+	ok = matcher.matchRow(commitRow2, true)
 	c.Assert(ok, check.IsTrue)
 	c.Assert(commitRow2, check.DeepEquals, &cdcpb.Event_Row{
 		StartTs:  2,
@@ -92,7 +95,7 @@ func (s *matcherSuite) TestMatchFakePrewrite(c *check.C) {
 		CommitTs: 2,
 		Key:      []byte("k1"),
 	}
-	ok := matcher.matchRow(commitRow1)
+	ok := matcher.matchRow(commitRow1, true)
 	c.Assert(commitRow1, check.DeepEquals, &cdcpb.Event_Row{
 		StartTs:  1,
 		CommitTs: 2,
@@ -103,10 +106,76 @@ func (s *matcherSuite) TestMatchFakePrewrite(c *check.C) {
 	c.Assert(ok, check.IsTrue)
 }
 
+func TestMatchRowUninitialized(t *testing.T) {
+	t.Parallel()
+	matcher := newMatcher()
+
+	// fake prewrite before init.
+	matcher.putPrewriteRow(&cdcpb.Event_Row{
+		StartTs:  1,
+		Key:      []byte("k1"),
+		OldValue: []byte("v4"),
+	})
+	commitRow1 := &cdcpb.Event_Row{
+		StartTs:  1,
+		CommitTs: 2,
+		Key:      []byte("k1"),
+	}
+	ok := matcher.matchRow(commitRow1, false)
+	require.Equal(t, &cdcpb.Event_Row{
+		StartTs:  1,
+		CommitTs: 2,
+		Key:      []byte("k1"),
+	}, commitRow1)
+	require.False(t, ok)
+	matcher.cacheCommitRow(commitRow1)
+
+	// actual prewrite before init.
+	matcher.putPrewriteRow(&cdcpb.Event_Row{
+		StartTs:  1,
+		Key:      []byte("k1"),
+		Value:    []byte("v3"),
+		OldValue: []byte("v4"),
+	})
+
+	// normal prewrite and commit before init.
+	matcher.putPrewriteRow(&cdcpb.Event_Row{
+		StartTs:  2,
+		Key:      []byte("k2"),
+		Value:    []byte("v3"),
+		OldValue: []byte("v4"),
+	})
+	commitRow2 := &cdcpb.Event_Row{
+		StartTs:  2,
+		CommitTs: 3,
+		Key:      []byte("k2"),
+	}
+	ok = matcher.matchRow(commitRow2, false)
+	require.Equal(t, &cdcpb.Event_Row{
+		StartTs:  2,
+		CommitTs: 3,
+		Key:      []byte("k2"),
+		Value:    []byte("v3"),
+		OldValue: []byte("v4"),
+	}, commitRow2)
+	require.True(t, ok)
+
+	// match cached row after init.
+	rows := matcher.matchCachedRow(true)
+	require.Len(t, rows, 1)
+	require.Equal(t, &cdcpb.Event_Row{
+		StartTs:  1,
+		CommitTs: 2,
+		Key:      []byte("k1"),
+		Value:    []byte("v3"),
+		OldValue: []byte("v4"),
+	}, rows[0])
+}
+
 func (s *matcherSuite) TestMatchMatchCachedRow(c *check.C) {
 	defer testleak.AfterTest(c)()
 	matcher := newMatcher()
-	c.Assert(len(matcher.matchCachedRow()), check.Equals, 0)
+	c.Assert(len(matcher.matchCachedRow(true)), check.Equals, 0)
 	matcher.cacheCommitRow(&cdcpb.Event_Row{
 		StartTs:  1,
 		CommitTs: 2,
@@ -122,7 +191,7 @@ func (s *matcherSuite) TestMatchMatchCachedRow(c *check.C) {
 		CommitTs: 5,
 		Key:      []byte("k3"),
 	})
-	c.Assert(len(matcher.matchCachedRow()), check.Equals, 0)
+	c.Assert(len(matcher.matchCachedRow(true)), check.Equals, 0)
 
 	matcher.cacheCommitRow(&cdcpb.Event_Row{
 		StartTs:  1,
@@ -159,7 +228,7 @@ func (s *matcherSuite) TestMatchMatchCachedRow(c *check.C) {
 		OldValue: []byte("ov3"),
 	})
 
-	c.Assert(matcher.matchCachedRow(), check.DeepEquals, []*cdcpb.Event_Row{{
+	c.Assert(matcher.matchCachedRow(true), check.DeepEquals, []*cdcpb.Event_Row{{
 		StartTs:  1,
 		CommitTs: 2,
 		Key:      []byte("k1"),
@@ -173,3 +242,57 @@ func (s *matcherSuite) TestMatchMatchCachedRow(c *check.C) {
 		OldValue: []byte("ov2"),
 	}})
 }
+
+func TestMatchMatchCachedRollbackRow(t *testing.T) {
+	t.Parallel()
+	matcher := newMatcher()
+	matcher.matchCachedRollbackRow(true)
+	matcher.cacheRollbackRow(&cdcpb.Event_Row{
+		StartTs: 1,
+		Key:     []byte("k1"),
+	})
+	matcher.cacheRollbackRow(&cdcpb.Event_Row{
+		StartTs: 3,
+		Key:     []byte("k2"),
+	})
+	matcher.cacheRollbackRow(&cdcpb.Event_Row{
+		StartTs: 4,
+		Key:     []byte("k3"),
+	})
+	matcher.matchCachedRollbackRow(true)
+
+	matcher.cacheRollbackRow(&cdcpb.Event_Row{
+		StartTs: 1,
+		Key:     []byte("k1"),
+	})
+	matcher.cacheRollbackRow(&cdcpb.Event_Row{
+		StartTs: 3,
+		Key:     []byte("k2"),
+	})
+	matcher.cacheRollbackRow(&cdcpb.Event_Row{
+		StartTs: 4,
+		Key:     []byte("k3"),
+	})
+
+	matcher.putPrewriteRow(&cdcpb.Event_Row{
+		StartTs:  1,
+		Key:      []byte("k1"),
+		Value:    []byte("v1"),
+		OldValue: []byte("ov1"),
+	})
+	matcher.putPrewriteRow(&cdcpb.Event_Row{
+		StartTs:  3,
+		Key:      []byte("k2"),
+		Value:    []byte("v2"),
+		OldValue: []byte("ov2"),
+	})
+	matcher.putPrewriteRow(&cdcpb.Event_Row{
+		StartTs:  4,
+		Key:      []byte("k3"),
+		Value:    []byte("v3"),
+		OldValue: []byte("ov3"),
+	})
+
+	matcher.matchCachedRollbackRow(true)
+	require.Empty(t, matcher.unmatchedValue)
+}
diff --git a/cdc/kv/region_worker_test.go b/cdc/kv/region_worker_test.go
index 13eac0d4c..e19d1e2b8 100644
--- a/cdc/kv/region_worker_test.go
+++ b/cdc/kv/region_worker_test.go
@@ -14,13 +14,20 @@
 package kv
 
 import (
+	"context"
 	"math/rand"
 	"runtime"
 	"sync"
+	"testing"
 
 	"github.com/pingcap/check"
+	"github.com/pingcap/kvproto/pkg/cdcpb"
+	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/pkg/config"
+	"github.com/pingcap/tiflow/pkg/regionspan"
 	"github.com/pingcap/tiflow/pkg/util/testleak"
+	"github.com/stretchr/testify/require"
+	"github.com/tikv/client-go/v2/tikv"
 )
 
 type regionWorkerSuite struct{}
@@ -135,3 +142,131 @@ func (s *regionWorkerSuite) TestRegionWorkerPoolSize(c *check.C) {
 	size = getWorkerPoolSize()
 	c.Assert(size, check.Equals, maxWorkerPoolSize)
 }
+
+func TestRegionWokerHandleEventEntryEventOutOfOrder(t *testing.T) {
+	// For UPDATE SQL, its prewrite event has both value and old value.
+	// It is possible that TiDB prewrites multiple times for the same row when
+	// there are other transactions it conflicts with. For this case,
+	// if the value is not "short", only the first prewrite contains the value.
+	//
+	// TiKV may output events for the UPDATE SQL as following:
+	//
+	// TiDB: [Prwrite1]    [Prewrite2]      [Commit]
+	//       v             v                v                                   Time
+	// ---------------------------------------------------------------------------->
+	//         ^            ^    ^           ^     ^       ^     ^          ^     ^
+	// TiKV:   [Scan Start] [Send Prewrite2] [Send Commit] [Send Prewrite1] [Send Init]
+	// TiCDC:                    [Recv Prewrite2]  [Recv Commit] [Recv Prewrite1] [Recv Init]
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	eventCh := make(chan model.RegionFeedEvent, 2)
+	s := createFakeEventFeedSession(ctx)
+	s.eventCh = eventCh
+	span := regionspan.Span{}.Hack()
+	state := newRegionFeedState(newSingleRegionInfo(
+		tikv.RegionVerID{},
+		regionspan.ToComparableSpan(span),
+		0, &tikv.RPCContext{}), 0)
+	state.start()
+	worker := newRegionWorker(s, "")
+	worker.initMetrics(ctx)
+	require.Equal(t, 2, cap(worker.outputCh))
+
+	// Receive prewrite2 with empty value.
+	events := &cdcpb.Event_Entries_{
+		Entries: &cdcpb.Event_Entries{
+			Entries: []*cdcpb.Event_Row{{
+				StartTs:  1,
+				Type:     cdcpb.Event_PREWRITE,
+				OpType:   cdcpb.Event_Row_PUT,
+				Key:      []byte("key"),
+				Value:    nil,
+				OldValue: []byte("oldvalue"),
+			}},
+		},
+	}
+	err := worker.handleEventEntry(ctx, events, state)
+	require.Nil(t, err)
+
+	// Receive commit.
+	events = &cdcpb.Event_Entries_{
+		Entries: &cdcpb.Event_Entries{
+			Entries: []*cdcpb.Event_Row{{
+				StartTs:  1,
+				CommitTs: 2,
+				Type:     cdcpb.Event_COMMIT,
+				OpType:   cdcpb.Event_Row_PUT,
+				Key:      []byte("key"),
+			}},
+		},
+	}
+	err = worker.handleEventEntry(context.Background(), events, state)
+	require.Nil(t, err)
+
+	// Must not output event.
+	var event model.RegionFeedEvent
+	var ok bool
+	select {
+	case event, ok = <-eventCh:
+	default:
+	}
+	require.Falsef(t, ok, "%v", event)
+	require.EqualValuesf(t, model.RegionFeedEvent{}, event, "%v", event)
+
+	// Receive prewrite1 with actual value.
+	events = &cdcpb.Event_Entries_{
+		Entries: &cdcpb.Event_Entries{
+			Entries: []*cdcpb.Event_Row{{
+				StartTs:  1,
+				Type:     cdcpb.Event_PREWRITE,
+				OpType:   cdcpb.Event_Row_PUT,
+				Key:      []byte("key"),
+				Value:    []byte("value"),
+				OldValue: []byte("oldvalue"),
+			}},
+		},
+	}
+	err = worker.handleEventEntry(ctx, events, state)
+	require.Nil(t, err)
+
+	// Must not output event.
+	select {
+	case event, ok = <-eventCh:
+	default:
+	}
+	require.Falsef(t, ok, "%v", event)
+	require.EqualValuesf(t, model.RegionFeedEvent{}, event, "%v", event)
+
+	// Receive prewrite1 with actual value.
+	events = &cdcpb.Event_Entries_{
+		Entries: &cdcpb.Event_Entries{
+			Entries: []*cdcpb.Event_Row{
+				{
+					Type: cdcpb.Event_INITIALIZED,
+				},
+			},
+		},
+	}
+	err = worker.handleEventEntry(ctx, events, state)
+	require.Nil(t, err)
+
+	// Must output event.
+	select {
+	case event, ok = <-eventCh:
+	default:
+	}
+	require.Truef(t, ok, "%v", event)
+	require.EqualValuesf(t, model.RegionFeedEvent{
+		RegionID: 0,
+		Val: &model.RawKVEntry{
+			OpType:   model.OpTypePut,
+			Key:      []byte("key"),
+			Value:    []byte("value"),
+			StartTs:  1,
+			CRTs:     2,
+			RegionID: 0,
+			OldValue: []byte("oldvalue"),
+		},
+	}, event, "%v", event)
+}
diff --git a/cdc/kv/testing.go b/cdc/kv/testing.go
index 6520f2559..97c72be60 100644
--- a/cdc/kv/testing.go
+++ b/cdc/kv/testing.go
@@ -26,6 +26,7 @@ import (
 	"github.com/pingcap/tidb/store"
 	"github.com/pingcap/tidb/store/driver"
 	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/pingcap/tiflow/pkg/config"
 	"github.com/pingcap/tiflow/pkg/regionspan"
 	"github.com/pingcap/tiflow/pkg/security"
 	"github.com/pingcap/tiflow/pkg/txnutil"
@@ -152,14 +153,14 @@ func TestSplit(t require.TestingT, pdCli pd.Client, storage tikv.Storage, kvStor
 	grpcPool := NewGrpcPoolImpl(ctx, &security.Credential{})
 	regionCache := tikv.NewRegionCache(pdCli)
 
-	cli := NewCDCClient(context.Background(), pdCli, storage, grpcPool, regionCache)
+	cli := NewCDCClient(context.Background(), pdCli, storage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 
 	startTS := mustGetTimestamp(t, storage)
 
 	lockresolver := txnutil.NewLockerResolver(storage)
 	isPullInit := &mockPullerInit{}
 	go func() {
-		err := cli.EventFeed(ctx, regionspan.ComparableSpan{Start: nil, End: nil}, startTS, false, lockresolver, isPullInit, eventCh)
+		err := cli.EventFeed(ctx, regionspan.ComparableSpan{Start: nil, End: nil}, startTS, lockresolver, isPullInit, eventCh)
 		require.Equal(t, err, context.Canceled)
 	}()
 
@@ -242,13 +243,13 @@ func TestGetKVSimple(t require.TestingT, pdCli pd.Client, storage tikv.Storage,
 
 	grpcPool := NewGrpcPoolImpl(ctx, &security.Credential{})
 	regionCache := tikv.NewRegionCache(pdCli)
-	cli := NewCDCClient(context.Background(), pdCli, storage, grpcPool, regionCache)
+	cli := NewCDCClient(context.Background(), pdCli, storage, grpcPool, regionCache, config.GetDefaultServerConfig().KVClient)
 
 	startTS := mustGetTimestamp(t, storage)
 	lockresolver := txnutil.NewLockerResolver(storage)
 	isPullInit := &mockPullerInit{}
 	go func() {
-		err := cli.EventFeed(ctx, regionspan.ComparableSpan{Start: nil, End: nil}, startTS, false, lockresolver, isPullInit, checker.eventCh)
+		err := cli.EventFeed(ctx, regionspan.ComparableSpan{Start: nil, End: nil}, startTS, lockresolver, isPullInit, checker.eventCh)
 		require.Equal(t, err, context.Canceled)
 	}()
 
@@ -270,7 +271,7 @@ func TestGetKVSimple(t require.TestingT, pdCli pd.Client, storage tikv.Storage,
 		if i == 1 {
 			checker = newEventChecker(t)
 			go func() {
-				err := cli.EventFeed(ctx, regionspan.ComparableSpan{Start: nil, End: nil}, startTS, false, lockresolver, isPullInit, checker.eventCh)
+				err := cli.EventFeed(ctx, regionspan.ComparableSpan{Start: nil, End: nil}, startTS, lockresolver, isPullInit, checker.eventCh)
 				require.Equal(t, err, context.Canceled)
 			}()
 		}
diff --git a/cdc/model/kv_test.go b/cdc/model/kv_test.go
index ae5fcfd01..a7fdb18f3 100644
--- a/cdc/model/kv_test.go
+++ b/cdc/model/kv_test.go
@@ -55,6 +55,8 @@ func TestRawKVEntry(t *testing.T) {
 		Value:   []byte("345"),
 	}
 
-	require.Equal(t, "OpType: 1, Key: 123, Value: 345, StartTs: 100, CRTs: 101, RegionID: 0", raw.String())
+	require.Equal(t,
+		"OpType: 1, Key: 123, Value: 345, OldValue: , StartTs: 100, CRTs: 101, RegionID: 0",
+		raw.String())
 	require.Equal(t, int64(6), raw.ApproximateDataSize())
 }
diff --git a/cdc/puller/puller_test.go b/cdc/puller/puller_test.go
index def434506..6d4fc1869 100644
--- a/cdc/puller/puller_test.go
+++ b/cdc/puller/puller_test.go
@@ -25,6 +25,7 @@ import (
 	"github.com/pingcap/tidb/store/mockstore"
 	"github.com/pingcap/tiflow/cdc/kv"
 	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/pingcap/tiflow/pkg/config"
 	cerrors "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/regionspan"
 	"github.com/pingcap/tiflow/pkg/retry"
@@ -63,6 +64,7 @@ func newMockCDCKVClient(
 	kvStorage tikv.Storage,
 	grpcPool kv.GrpcPool,
 	regionCache *tikv.RegionCache,
+	cfg *config.KVClientConfig,
 ) kv.CDCKVClient {
 	return &mockCDCKVClient{
 		expectations: make(chan model.RegionFeedEvent, 1024),
@@ -73,7 +75,6 @@ func (mc *mockCDCKVClient) EventFeed(
 	ctx context.Context,
 	span regionspan.ComparableSpan,
 	ts uint64,
-	enableOldValue bool,
 	lockResolver txnutil.LockResolver,
 	isPullerInit kv.PullerInitialization,
 	eventCh chan<- model.RegionFeedEvent,
@@ -116,7 +117,6 @@ func (s *pullerSuite) newPullerForTest(
 	ctx, cancel := context.WithCancel(context.Background())
 	store, err := mockstore.NewMockStore()
 	c.Assert(err, check.IsNil)
-	enableOldValue := true
 	backupNewCDCKVClient := kv.NewCDCKVClient
 	kv.NewCDCKVClient = newMockCDCKVClient
 	defer func() {
@@ -127,7 +127,7 @@ func (s *pullerSuite) newPullerForTest(
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdCli)
 	defer regionCache.Close()
-	plr := NewPuller(ctx, pdCli, grpcPool, regionCache, store, checkpointTs, spans, enableOldValue)
+	plr := NewPuller(ctx, pdCli, grpcPool, regionCache, store, checkpointTs, spans, config.GetDefaultServerConfig().KVClient)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
diff --git a/cdc/scheduler/schedule_dispatcher_test.go b/cdc/scheduler/schedule_dispatcher_test.go
index b699c8b6b..c5fa7e394 100644
--- a/cdc/scheduler/schedule_dispatcher_test.go
+++ b/cdc/scheduler/schedule_dispatcher_test.go
@@ -66,10 +66,10 @@ func (m *mockScheduleDispatcherCommunicator) DispatchTable(
 ) (done bool, err error) {
 	if !m.isBenchmark {
 		log.Info("dispatch table called",
-			zap.String("changefeed-id", changeFeedID),
+			zap.String("changefeed", changeFeedID),
 			zap.Int64("table-id", tableID),
-			zap.String("capture-id", captureID),
-			zap.Bool("is-delete", isDelete),
+			zap.String("captureID", captureID),
+			zap.Bool("isDelete", isDelete),
 			zap.String("epoch", epoch))
 		if !isDelete {
 			m.addTableRecords[captureID] = append(m.addTableRecords[captureID], tableID)
diff --git a/cdc/sink/mysql_test.go b/cdc/sink/mysql_test.go
index ee1e61bf0..ae3de13c0 100644
--- a/cdc/sink/mysql_test.go
+++ b/cdc/sink/mysql_test.go
@@ -882,7 +882,7 @@ func TestExecDMLRollbackErrRetryable(t *testing.T) {
 		// normal db
 		db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherEqual))
 		require.Nil(t, err)
-		for i := 0; i < defaultDMLMaxRetryTime; i++ {
+		for i := 0; i < int(defaultDMLMaxRetry); i++ {
 			mock.ExpectBegin()
 			mock.ExpectExec("REPLACE INTO `s1`.`t1`(`a`) VALUES (?),(?)").
 				WithArgs(1, 2).
@@ -894,8 +894,11 @@ func TestExecDMLRollbackErrRetryable(t *testing.T) {
 	}
 	backupGetDBConn := GetDBConnImpl
 	GetDBConnImpl = mockGetDBConnErrDatabaseNotExists
+	backupMaxRetry := defaultDMLMaxRetry
+	defaultDMLMaxRetry = 2
 	defer func() {
 		GetDBConnImpl = backupGetDBConn
+		defaultDMLMaxRetry = backupMaxRetry
 	}()
 
 	ctx, cancel := context.WithCancel(context.Background())
diff --git a/pkg/cmd/server/server_test.go b/pkg/cmd/server/server_test.go
index 915064aab..5b866278b 100644
--- a/pkg/cmd/server/server_test.go
+++ b/pkg/cmd/server/server_test.go
@@ -170,9 +170,10 @@ func TestParseCfg(t *testing.T) {
 		},
 		PerTableMemoryQuota: 10 * 1024 * 1024, // 10M
 		KVClient: &config.KVClientConfig{
-			WorkerConcurrent: 8,
-			WorkerPoolSize:   0,
-			RegionScanLimit:  40,
+			WorkerConcurrent:    8,
+			WorkerPoolSize:      0,
+			RegionScanLimit:     40,
+			RegionRetryDuration: config.TomlDuration(time.Minute),
 		},
 		Debug: &config.DebugConfig{
 			EnableTableActor: false,
@@ -240,6 +241,9 @@ num-concurrent-worker = 4
 num-workerpool-goroutine = 5
 sort-dir = "/tmp/just_a_test"
 
+[kv-client]
+region-retry-duration = "3s"
+
 [debug]
 enable-db-sorter = false
 [debug.db]
@@ -309,9 +313,10 @@ server-worker-pool-size = 16
 		Security:            &config.SecurityConfig{},
 		PerTableMemoryQuota: 10 * 1024 * 1024, // 10M
 		KVClient: &config.KVClientConfig{
-			WorkerConcurrent: 8,
-			WorkerPoolSize:   0,
-			RegionScanLimit:  40,
+			WorkerConcurrent:    8,
+			WorkerPoolSize:      0,
+			RegionScanLimit:     40,
+			RegionRetryDuration: config.TomlDuration(3 * time.Second),
 		},
 		Debug: &config.DebugConfig{
 			EnableTableActor: false,
@@ -446,9 +451,10 @@ cert-allowed-cn = ["dd","ee"]
 		},
 		PerTableMemoryQuota: 10 * 1024 * 1024, // 10M
 		KVClient: &config.KVClientConfig{
-			WorkerConcurrent: 8,
-			WorkerPoolSize:   0,
-			RegionScanLimit:  40,
+			WorkerConcurrent:    8,
+			WorkerPoolSize:      0,
+			RegionScanLimit:     40,
+			RegionRetryDuration: config.TomlDuration(time.Minute),
 		},
 		Debug: &config.DebugConfig{
 			EnableTableActor: false,
diff --git a/pkg/config/config_test_data.go b/pkg/config/config_test_data.go
index 2e6206bc0..e8b40781b 100644
--- a/pkg/config/config_test_data.go
+++ b/pkg/config/config_test_data.go
@@ -105,7 +105,8 @@ const (
   "kv-client": {
     "worker-concurrent": 8,
     "worker-pool-size": 0,
-    "region-scan-limit": 40
+    "region-scan-limit": 40,
+    "region-retry-duration": 60000000000
   },
   "debug": {
     "enable-table-actor": false,
diff --git a/pkg/config/server_config_test.go b/pkg/config/server_config_test.go
index b164defe6..96478cff8 100644
--- a/pkg/config/server_config_test.go
+++ b/pkg/config/server_config_test.go
@@ -15,6 +15,7 @@ package config
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 )
@@ -93,3 +94,14 @@ func TestDBConfigValidateAndAdjust(t *testing.T) {
 	conf.CleanupSpeedLimit = 0
 	require.Error(t, conf.ValidateAndAdjust())
 }
+
+func TestKVClientConfigValidateAndAdjust(t *testing.T) {
+	t.Parallel()
+	conf := GetDefaultServerConfig().Clone().KVClient
+
+	require.Nil(t, conf.ValidateAndAdjust())
+	conf.RegionRetryDuration = TomlDuration(time.Second)
+	require.Nil(t, conf.ValidateAndAdjust())
+	conf.RegionRetryDuration = -TomlDuration(time.Second)
+	require.Error(t, conf.ValidateAndAdjust())
+}
diff --git a/pkg/orchestrator/etcd_worker_test.go b/pkg/orchestrator/etcd_worker_test.go
index b493815b0..dc8c0ecb4 100644
--- a/pkg/orchestrator/etcd_worker_test.go
+++ b/pkg/orchestrator/etcd_worker_test.go
@@ -78,7 +78,8 @@ func (s *simpleReactor) Tick(_ context.Context, state ReactorState) (nextState R
 			}
 		}
 		if sum != expectedSum {
-			log.Panic("state is inconsistent", zap.Int("expected-sum", sum), zap.Int("actual-sum", s.state.sum))
+			log.Panic("state is inconsistent",
+				zap.Int("expectedSum", sum), zap.Int("actualSum", s.state.sum))
 		}
 
 		s.state.SetSum(sum)
diff --git a/pkg/retry/retry_test.go b/pkg/retry/retry_test.go
index c7f9d5d8b..2080e7cb3 100644
--- a/pkg/retry/retry_test.go
+++ b/pkg/retry/retry_test.go
@@ -97,7 +97,7 @@ func TestDoCancelInfiniteRetry(t *testing.T) {
 		return errors.New("test")
 	}
 
-	err := Do(ctx, f, WithInfiniteTries(), WithBackoffBaseDelay(2), WithBackoffMaxDelay(10))
+	err := Do(ctx, f, WithBackoffBaseDelay(2), WithBackoffMaxDelay(10))
 	require.Equal(t, errors.Cause(err), context.DeadlineExceeded)
 	require.GreaterOrEqual(t, callCount, 1, "tries: %d", callCount)
 	require.Less(t, callCount, math.MaxInt64)
@@ -114,7 +114,7 @@ func TestDoCancelAtBeginning(t *testing.T) {
 		return errors.New("test")
 	}
 
-	err := Do(ctx, f, WithInfiniteTries(), WithBackoffBaseDelay(2), WithBackoffMaxDelay(10))
+	err := Do(ctx, f, WithBackoffBaseDelay(2), WithBackoffMaxDelay(10))
 	require.Equal(t, errors.Cause(err), context.Canceled)
 	require.Equal(t, callCount, 0, "tries:%d", callCount)
 }
@@ -147,16 +147,58 @@ func TestDoCornerCases(t *testing.T) {
 	require.Regexp(t, "test", errors.Cause(err))
 	require.Equal(t, callCount, 2)
 
-	var i int64
-	for i = -10; i < 10; i++ {
+	var i uint64
+	for i = 0; i < 10; i++ {
 		callCount = 0
-		err = Do(context.Background(), f, WithBackoffBaseDelay(i), WithBackoffMaxDelay(i), WithMaxTries(i))
+		err = Do(context.Background(), f,
+			WithBackoffBaseDelay(int64(i)), WithBackoffMaxDelay(int64(i)), WithMaxTries(i))
 		require.Regexp(t, "test", errors.Cause(err))
 		require.Regexp(t, ".*CDC:ErrReachMaxTry.*", err)
-		if i > 0 {
-			require.Equal(t, int64(callCount), i)
+		if i == 0 {
+			require.Equal(t, 1, callCount)
 		} else {
-			require.Equal(t, callCount, defaultMaxTries)
+			require.Equal(t, int(i), callCount)
 		}
 	}
 }
+
+func TestTotalRetryDuration(t *testing.T) {
+	t.Parallel()
+
+	f := func() error {
+		return errors.New("test")
+	}
+
+	start := time.Now()
+	err := Do(
+		context.Background(), f,
+		WithBackoffBaseDelay(math.MinInt64),
+		WithTotalRetryDuratoin(time.Second),
+	)
+	require.Regexp(t, "test", errors.Cause(err))
+	require.LessOrEqual(t, 1, int(math.Round(time.Since(start).Seconds())))
+
+	start = time.Now()
+	err = Do(
+		context.Background(), f,
+		WithBackoffBaseDelay(math.MinInt64),
+		WithTotalRetryDuratoin(2*time.Second),
+	)
+	require.Regexp(t, "test", errors.Cause(err))
+	require.LessOrEqual(t, 2, int(math.Round(time.Since(start).Seconds())))
+}
+
+func TestRetryError(t *testing.T) {
+	t.Parallel()
+
+	f := func() error {
+		return errors.New("some error info")
+	}
+
+	err := Do(
+		context.Background(), f, WithBackoffBaseDelay(math.MinInt64), WithMaxTries(2),
+	)
+	require.Regexp(t, "some error info", errors.Cause(err))
+	require.Regexp(t, ".*some error info.*", err.Error())
+	require.Regexp(t, ".*CDC:ErrReachMaxTry.*", err.Error())
+}
diff --git a/pkg/txnutil/gc/gc_service_test.go b/pkg/txnutil/gc/gc_service_test.go
index d32214a8b..a42111305 100644
--- a/pkg/txnutil/gc/gc_service_test.go
+++ b/pkg/txnutil/gc/gc_service_test.go
@@ -62,7 +62,8 @@ func (s *gcServiceSuite) TestCheckSafetyOfStartTs(c *check.C) {
 	s.pdCli.retryCount = 0
 	err = EnsureChangefeedStartTsSafety(ctx, s.pdCli, "changefeed2", TTL, 65)
 	c.Assert(err, check.NotNil)
-	c.Assert(err.Error(), check.Equals, "[CDC:ErrReachMaxTry]reach maximum try: 9: not pd leader")
+	c.Assert(err.Error(), check.Equals,
+		"[CDC:ErrReachMaxTry]reach maximum try: 9, error: not pd leader: not pd leader")
 
 	s.pdCli.retryThreshold = 3
 	s.pdCli.retryCount = 0
diff --git a/tests/integration_tests/cli/run.sh b/tests/integration_tests/cli/run.sh
index 6d42258bf..f32adbfb0 100644
--- a/tests/integration_tests/cli/run.sh
+++ b/tests/integration_tests/cli/run.sh
@@ -158,6 +158,10 @@ EOF
 	# Smoke test unsafe commands
 	echo "y" | run_cdc_cli unsafe delete-service-gc-safepoint
 	run_cdc_cli unsafe reset --no-confirm
+	REGION_ID=$(pd-ctl -u=$pd_addr region | jq '.regions[0].id')
+	TS=$(cdc cli tso query --pd=$pd_addr)
+	run_cdc_cli unsafe resolve-lock --region=$REGION_ID
+	run_cdc_cli unsafe resolve-lock --region=$REGION_ID --ts=$TS
 
 	# Smoke test change log level
 	curl -X POST -d '"warn"' http://127.0.0.1:8300/api/v1/log
diff --git a/tests/integration_tests/move_table/main.go b/tests/integration_tests/move_table/main.go
index 0ced3eb2c..0dfcb5694 100644
--- a/tests/integration_tests/move_table/main.go
+++ b/tests/integration_tests/move_table/main.go
@@ -244,9 +244,9 @@ func (c *cluster) refreshInfo(ctx context.Context) error {
 		}
 
 		log.Debug("retrieved processor details",
-			zap.String("changefeed-id", changefeed),
-			zap.String("capture-id", capture.ID),
-			zap.Any("processor-detail", processorDetails))
+			zap.String("changefeed", changefeed),
+			zap.String("captureID", capture.ID),
+			zap.Any("processorDetail", processorDetails))
 		for _, tableID := range processorDetails.Tables {
 			c.captures[capture.ID] = append(c.captures[capture.ID], &tableInfo{
 				ID:         tableID,
diff --git a/tests/mq_protocol_tests/framework/sql_batch_op.go b/tests/mq_protocol_tests/framework/sql_batch_op.go
index 3ba8f0dc0..e05c9072c 100644
--- a/tests/mq_protocol_tests/framework/sql_batch_op.go
+++ b/tests/mq_protocol_tests/framework/sql_batch_op.go
@@ -105,7 +105,7 @@ func (s *sqlAllAwaiter) poll(ctx context.Context) (bool, error) {
 		}
 	}
 
-	log.Debug("poll finished", zap.Int("total-retrieved", len(s.retrievedValues)))
+	log.Debug("poll finished", zap.Int("totalRetrieved", len(s.retrievedValues)))
 
 	if len(s.data) == len(s.retrievedValues) {
 		return true, nil
diff --git a/tests/mq_protocol_tests/framework/sql_helper.go b/tests/mq_protocol_tests/framework/sql_helper.go
index 69a087110..d05e8b4dd 100644
--- a/tests/mq_protocol_tests/framework/sql_helper.go
+++ b/tests/mq_protocol_tests/framework/sql_helper.go
@@ -400,7 +400,7 @@ func (s *sqlRequest) poll(ctx context.Context) (bool, error) {
 		if res == nil {
 			return true, nil
 		}
-		log.Debug("Delete not successful yet", zap.Reflect("where condition", s.getWhereCondition()))
+		log.Debug("Delete not successful yet", zap.Reflect("where", s.getWhereCondition()))
 		return false, nil
 	}
 	return true, nil
diff --git a/tests/utils/cdc_state_checker/state.go b/tests/utils/cdc_state_checker/state.go
index 313f24d41..77396a56a 100644
--- a/tests/utils/cdc_state_checker/state.go
+++ b/tests/utils/cdc_state_checker/state.go
@@ -53,12 +53,12 @@ func newCDCReactorState() *cdcReactorState {
 func (s *cdcReactorState) Update(key util.EtcdKey, value []byte, isInit bool) error {
 	if key.String() == etcd.CaptureOwnerKey {
 		if value == nil {
-			log.Info("Owner lost", zap.String("old-owner", s.Owner))
+			log.Info("Owner lost", zap.String("oldOwner", s.Owner))
 			return nil
 		}
 
-		log.Info("Owner updated", zap.String("old-owner", s.Owner),
-			zap.ByteString("new-owner", value))
+		log.Info("Owner updated", zap.String("oldOwner", s.Owner),
+			zap.ByteString("newOwner", value))
 		s.Owner = string(value)
 		return nil
 	}
@@ -69,7 +69,7 @@ func (s *cdcReactorState) Update(key util.EtcdKey, value []byte, isInit bool) er
 		if value == nil {
 			log.Info("Capture deleted",
 				zap.String("captureID", captureID),
-				zap.Reflect("old-capture", s.Captures[captureID]))
+				zap.Reflect("oldCapture", s.Captures[captureID]))
 
 			delete(s.Captures, captureID)
 			return nil
@@ -84,12 +84,12 @@ func (s *cdcReactorState) Update(key util.EtcdKey, value []byte, isInit bool) er
 		if oldCaptureInfo, ok := s.Captures[captureID]; ok {
 			log.Info("Capture updated",
 				zap.String("captureID", captureID),
-				zap.Reflect("old-capture", oldCaptureInfo),
-				zap.Reflect("new-capture", newCaptureInfo))
+				zap.Reflect("oldCapture", oldCaptureInfo),
+				zap.Reflect("newCapture", newCaptureInfo))
 		} else {
 			log.Info("Capture added",
 				zap.String("captureID", captureID),
-				zap.Reflect("new-capture", newCaptureInfo))
+				zap.Reflect("newCapture", newCaptureInfo))
 		}
 
 		s.Captures[captureID] = &newCaptureInfo
@@ -101,8 +101,8 @@ func (s *cdcReactorState) Update(key util.EtcdKey, value []byte, isInit bool) er
 
 		if value == nil {
 			log.Info("Changefeed deleted",
-				zap.String("changefeedID", changefeedID),
-				zap.Reflect("old-changefeed", s.ChangefeedStatuses))
+				zap.String("changefeed", changefeedID),
+				zap.Reflect("oldChangefeed", s.ChangefeedStatuses))
 
 			delete(s.ChangefeedStatuses, changefeedID)
 			return nil
@@ -116,13 +116,13 @@ func (s *cdcReactorState) Update(key util.EtcdKey, value []byte, isInit bool) er
 
 		if oldChangefeedInfo, ok := s.ChangefeedStatuses[changefeedID]; ok {
 			log.Info("Changefeed updated",
-				zap.String("changefeedID", changefeedID),
-				zap.Reflect("old-changefeed", oldChangefeedInfo),
-				zap.Reflect("new-changefeed", newChangefeedStatus))
+				zap.String("changefeed", changefeedID),
+				zap.Reflect("oldChangefeed", oldChangefeedInfo),
+				zap.Reflect("newChangefeed", newChangefeedStatus))
 		} else {
 			log.Info("Changefeed added",
-				zap.String("changefeedID", changefeedID),
-				zap.Reflect("new-changefeed", newChangefeedStatus))
+				zap.String("changefeed", changefeedID),
+				zap.Reflect("newChangefeed", newChangefeedStatus))
 		}
 
 		s.ChangefeedStatuses[changefeedID] = &newChangefeedStatus
@@ -137,8 +137,8 @@ func (s *cdcReactorState) Update(key util.EtcdKey, value []byte, isInit bool) er
 		if value == nil {
 			log.Info("Position deleted",
 				zap.String("captureID", captureID),
-				zap.String("changefeedID", changefeedID),
-				zap.Reflect("old-position", s.TaskPositions[changefeedID][captureID]))
+				zap.String("changefeed", changefeedID),
+				zap.Reflect("oldPosition", s.TaskPositions[changefeedID][captureID]))
 
 			delete(s.TaskPositions[changefeedID], captureID)
 			if len(s.TaskPositions[changefeedID]) == 0 {
@@ -161,14 +161,14 @@ func (s *cdcReactorState) Update(key util.EtcdKey, value []byte, isInit bool) er
 		if position, ok := s.TaskPositions[changefeedID][captureID]; ok {
 			log.Info("Position updated",
 				zap.String("captureID", captureID),
-				zap.String("changefeedID", changefeedID),
-				zap.Reflect("old-position", position),
-				zap.Reflect("new-position", newTaskPosition))
+				zap.String("changefeed", changefeedID),
+				zap.Reflect("oldPosition", position),
+				zap.Reflect("newPosition", newTaskPosition))
 		} else {
 			log.Info("Position created",
 				zap.String("captureID", captureID),
-				zap.String("changefeedID", changefeedID),
-				zap.Reflect("new-position", newTaskPosition))
+				zap.String("changefeed", changefeedID),
+				zap.Reflect("newPosition", newTaskPosition))
 		}
 
 		s.TaskPositions[changefeedID][captureID] = &newTaskPosition
@@ -183,8 +183,8 @@ func (s *cdcReactorState) Update(key util.EtcdKey, value []byte, isInit bool) er
 		if value == nil {
 			log.Info("Status deleted",
 				zap.String("captureID", captureID),
-				zap.String("changefeedID", changefeedID),
-				zap.Reflect("old-status", s.TaskStatuses[changefeedID][captureID]))
+				zap.String("changefeed", changefeedID),
+				zap.Reflect("oldStatus", s.TaskStatuses[changefeedID][captureID]))
 
 			delete(s.TaskStatuses[changefeedID], captureID)
 			if len(s.TaskStatuses[changefeedID]) == 0 {
@@ -207,14 +207,14 @@ func (s *cdcReactorState) Update(key util.EtcdKey, value []byte, isInit bool) er
 		if status, ok := s.TaskStatuses[changefeedID][captureID]; ok {
 			log.Info("Status updated",
 				zap.String("captureID", captureID),
-				zap.String("changefeedID", changefeedID),
-				zap.Reflect("old-status", status),
-				zap.Reflect("new-status", newTaskStatus))
+				zap.String("changefeed", changefeedID),
+				zap.Reflect("oldStatus", status),
+				zap.Reflect("newStatus", newTaskStatus))
 		} else {
 			log.Info("Status updated",
 				zap.String("captureID", captureID),
-				zap.String("changefeedID", changefeedID),
-				zap.Reflect("new-status", newTaskStatus))
+				zap.String("changefeed", changefeedID),
+				zap.Reflect("newStatus", newTaskStatus))
 		}
 
 		s.TaskStatuses[changefeedID][captureID] = &newTaskStatus
diff --git a/tests/utils/many_sorters_test/many_sorters.go b/tests/utils/many_sorters_test/many_sorters.go
index e6f21e4dc..e6cb6bcfa 100644
--- a/tests/utils/many_sorters_test/many_sorters.go
+++ b/tests/utils/many_sorters_test/many_sorters.go
@@ -122,7 +122,7 @@ func main() {
 				}
 
 				if ev.CRTs == uint64(((*numEvents)<<5)+1) {
-					log.Info("Sorter finished", zap.Int("sorter-id", finalI))
+					log.Info("Sorter finished", zap.Int("sorterID", finalI))
 					if atomic.AddInt32(&finishCount, 1) == int32(*numSorters) {
 						log.Info("Many Sorters test finished, cancelling all goroutines")
 						cancel()