This is an automated cherry-pick of pingcap#3277

Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io>

cdc/capture/capture.go

@@ -268,7 +268,8 @@ func (c *Capture) runEtcdWorker(ctx cdcContext.Context, reactor orchestrator.Rea
 	if err != nil {
 		return errors.Trace(err)
 	}
-	if err := etcdWorker.Run(ctx, c.session, timerInterval); err != nil {
+	captureAddr := c.info.AdvertiseAddr
+	if err := etcdWorker.Run(ctx, c.session, timerInterval, captureAddr); err != nil {
 		// We check ttl of lease instead of check `session.Done`, because
 		// `session.Done` is only notified when etcd client establish a
 		// new keepalive request, there could be a time window as long as

cdc/metrics.go

@@ -22,7 +22,17 @@ import (
 	"github.com/pingcap/ticdc/cdc/puller"
 	"github.com/pingcap/ticdc/cdc/puller/sorter"
 	"github.com/pingcap/ticdc/cdc/sink"
+<<<<<<< HEAD
 	"github.com/pingcap/ticdc/pkg/config"
+=======
+	"github.com/pingcap/ticdc/cdc/sorter"
+	"github.com/pingcap/ticdc/cdc/sorter/leveldb"
+	"github.com/pingcap/ticdc/cdc/sorter/memory"
+	"github.com/pingcap/ticdc/cdc/sorter/unified"
+	"github.com/pingcap/ticdc/pkg/actor"
+	"github.com/pingcap/ticdc/pkg/etcd"
+	"github.com/pingcap/ticdc/pkg/orchestrator"
+>>>>>>> 2569abaa3 (etcd_worker: batch etcd patch (#3277))
 	"github.com/prometheus/client_golang/prometheus"
 )
 
@@ -36,6 +46,17 @@ func init() {
 	puller.InitMetrics(registry)
 	sink.InitMetrics(registry)
 	entry.InitMetrics(registry)
+<<<<<<< HEAD
+=======
+	processor.InitMetrics(registry)
+	tablepipeline.InitMetrics(registry)
+	owner.InitMetrics(registry)
+	etcd.InitMetrics(registry)
+	initServerMetrics(registry)
+	actor.InitMetrics(registry)
+	orchestrator.InitMetrics(registry)
+	// Sorter metrics
+>>>>>>> 2569abaa3 (etcd_worker: batch etcd patch (#3277))
 	sorter.InitMetrics(registry)
 	if config.NewReplicaImpl {
 		processor.InitMetrics(registry)

cdc/model/reactor_state.go

@@ -100,7 +100,12 @@ func (s *GlobalReactorState) Update(key util.EtcdKey, value []byte, _ bool) erro
 }
 
 // GetPatches implements the ReactorState interface
+<<<<<<< HEAD:cdc/model/reactor_state.go
 func (s *GlobalReactorState) GetPatches() [][]orchestrator.DataPatch {
+=======
+// Every []DataPatch slice in [][]DataPatch slice is the patches of a ChangefeedReactorState
+func (s *GlobalReactorState) GetPatches() [][]DataPatch {
+>>>>>>> 2569abaa3 (etcd_worker: batch etcd patch (#3277)):pkg/orchestrator/reactor_state.go
 	pendingPatches := s.pendingPatches
 	for _, changefeedState := range s.Changefeeds {
 		pendingPatches = append(pendingPatches, changefeedState.getPatches())

errors.toml

@@ -196,6 +196,11 @@ error = '''
 the etcd txn should be aborted and retried immediately
 '''
 
+["CDC:ErrEtcdTxnSizeExceed"]
+error = '''
+patch size of a single changefeed exceed etcd txn max size
+'''
+
 ["CDC:ErrEventFeedAborted"]
 error = '''
 single event feed aborted