Etcd (ticdc, dm):add AutoSyncInterval for Etcd client

cdc/api/v2/api_helpers.go

@@ -483,10 +483,11 @@ func (h APIV2HelpersImpl) getEtcdClient(
 	logConfig.Level = zap.NewAtomicLevelAt(zapcore.ErrorLevel)
 	return clientv3.New(
 		clientv3.Config{
-			Endpoints:   pdAddrs,
-			TLS:         tlsCfg,
-			LogConfig:   logConfig,
-			DialTimeout: 5 * time.Second,
+			Endpoints:        pdAddrs,
+			TLS:              tlsCfg,
+			LogConfig:        logConfig,
+			DialTimeout:      5 * time.Second,
+			AutoSyncInterval: 3 * time.Second,
 			DialOptions: []grpc.DialOption{
 				grpcTLSOption,
 				grpc.WithBlock(),

cdc/server/server.go

@@ -147,10 +147,11 @@ func (s *server) prepare(ctx context.Context) error {
 	// then cause the new owner cannot be elected immediately after the old owner offline.
 	// see https://github.com/etcd-io/etcd/blob/525d53bd41/client/v3/concurrency/election.go#L98
 	etcdCli, err := clientv3.New(clientv3.Config{
-		Endpoints:   s.pdEndpoints,
-		TLS:         tlsConfig,
-		LogConfig:   &logConfig,
-		DialTimeout: 5 * time.Second,
+		Endpoints:        s.pdEndpoints,
+		TLS:              tlsConfig,
+		LogConfig:        &logConfig,
+		DialTimeout:      5 * time.Second,
+		AutoSyncInterval: 3 * time.Second,
 		DialOptions: []grpc.DialOption{
 			grpcTLSOption,
 			grpc.WithBlock(),

dm/pkg/etcdutil/etcdutil.go

@@ -60,9 +60,10 @@ var etcdDefaultTxnStrategy = retry.FiniteRetryStrategy{}
 // CreateClient creates an etcd client with some default config items.
 func CreateClient(endpoints []string, tlsCfg *tls.Config) (*clientv3.Client, error) {
 	return clientv3.New(clientv3.Config{
-		Endpoints:   endpoints,
-		DialTimeout: DefaultDialTimeout,
-		TLS:         tlsCfg,
+		Endpoints:        endpoints,
+		DialTimeout:      DefaultDialTimeout,
+		AutoSyncInterval: 3 * time.Second,
+		TLS:              tlsCfg,
 	})
 }
 

pkg/cmd/factory/factory_impl.go

@@ -110,11 +110,12 @@ func (f *factoryImpl) EtcdClient() (*etcd.CDCEtcdClientImpl, error) {
 	pdEndpoints := strings.Split(pdAddr, ",")
 
 	etcdClient, err := clientv3.New(clientv3.Config{
-		Context:     ctx,
-		Endpoints:   pdEndpoints,
-		TLS:         tlsConfig,
-		LogConfig:   &logConfig,
-		DialTimeout: 30 * time.Second,
+		Context:          ctx,
+		Endpoints:        pdEndpoints,
+		TLS:              tlsConfig,
+		LogConfig:        &logConfig,
+		DialTimeout:      30 * time.Second,
+		AutoSyncInterval: 3 * time.Second,
 		// TODO(hi-rustin): add gRPC metrics to Options.
 		// See also: https://github.com/pingcap/tiflow/pull/2341#discussion_r673018537.
 		DialOptions: []grpc.DialOption{