Repository: pingcap/tiflow (go)

Issue:
TiKV initial scan task not cancel/termicated when changefeed paused or failed
### What did you do?

1. 07:51 - 07:55 Create 4 CDC clusters (each CDC cluster with 32C CDC nodes), for each CDC cluster create one changefeed,  each changefeed replicating ~40k tables (overall ~160k tables)
2. 07:58 Pause changefeed and update config 
3. 08:12 resume all changefeeds
4. 09:38 Pause all changefeed
5. 10:33 delete changefeed (Delete changefeeds from TiDB cloud UI)
6. 10:07 - 10:33 manually kill and restart all TiKV

### What did you expect to see?

- When task paused, previous pending initial scan task should be cancelled, Otherwise this might block following changefeed initilization.

### What did you see instead?

After step 1, one changefeed initilization finished (all tables added the changefeed)
After step 3, For the initilized changefeed, table count is 0 for 1.4h+ before it was paused.
After step 4, initial scan pending tasks is decreasing, but it is very slow.
After step 6, the pending tasks disappeared.

![image](https://github.com/user-attachments/assets/f55c340a-4b03-410f-ac61-0c23836ba3bb)
![image](https://github.com/user-attachments/assets/7a8a404c-8c94-4107-a740-0610fec418d8)

### Versions of the cluster

v6.5.9

Pull request description:
sourcemanager(ticdc): close all pullers asyncrhonsly to prevent block too long
<!--
Thank you for contributing to TiFlow! 
Please read MD's [CONTRIBUTING](https://github.com/pingcap/tiflow/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.
-->

### What problem does this PR solve?
<!--
Please create an issue first to describe the problem.

There MUST be one line starting with "Issue Number:  " and 
linking the relevant issues via the "close" or "ref".

For more info, check https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue.
 -->

Issue Number: close #11638 

### What is changed and how it works?

* close all pullers asyncronsly

Before this PR, all pullers close in order, one by one, each takes around 0.03 ~ 0.05s, if there are a lot of tables, it would takes a long time.

The following log shows that close 1000 tables takes 50 seconds.
```
["All pullers have been closed"] [namespace=default] [changefeed=blackhole-test-2] [cost=51.169607168s]
```

After the PR, the time cost is depend on the slowest table puller, around 0 ~ 2 seconds.
```
[manager.go:179] ["All pullers have been closed"] [namespace=default] [changefeed=blackhole-test-2] [cost=2.137702035s]
```

### Check List <!--REMOVE the items that are not applicable-->

#### Tests <!-- At least one of them must be included. -->

 - Unit test
 - Integration test

#### Questions <!-- Authors should answer these questions and reviewers should consider these questions. -->

##### Will it cause performance regression or break compatibility?

##### Do you need to update user documentation, design documentation or monitoring documentation?

### Release note <!-- bugfixes or new features need a release note -->

```release-note
fix the puller resource not closed in time
```


Patch:
diff --git a/OWNERS b/OWNERS
index 6dfab384e..f12d5ff1a 100644
--- a/OWNERS
+++ b/OWNERS
@@ -29,6 +29,7 @@ approvers:
   - sdojjy
   - suzaku
   - wk989898
+  - wlwilliamx
   - YuJuncen
 reviewers:
   # - migration-reviewers
diff --git a/cdc/api/v2/api.go b/cdc/api/v2/api.go
index 40779b58d..f0243992d 100644
--- a/cdc/api/v2/api.go
+++ b/cdc/api/v2/api.go
@@ -86,7 +86,7 @@ func RegisterOpenAPIV2Routes(router *gin.Engine, api OpenAPIV2) {
 
 	// owner apis
 	ownerGroup := v2.Group("/owner")
-	unsafeGroup.Use(middleware.ForwardToOwnerMiddleware(api.capture))
+	ownerGroup.Use(middleware.ForwardToOwnerMiddleware(api.capture))
 	ownerGroup.POST("/resign", api.resignOwner)
 
 	// common APIs
diff --git a/cdc/api/v2/changefeed.go b/cdc/api/v2/changefeed.go
index 4465d0053..a5d2d45c3 100644
--- a/cdc/api/v2/changefeed.go
+++ b/cdc/api/v2/changefeed.go
@@ -24,7 +24,7 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
-	tidbkv "github.com/pingcap/tidb/kv"
+	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tiflow/cdc/api"
 	"github.com/pingcap/tiflow/cdc/capture"
 	"github.com/pingcap/tiflow/cdc/model"
@@ -57,30 +57,34 @@ func (h *OpenAPIV2) createChangefeed(c *gin.Context) {
 		_ = c.Error(cerror.WrapError(cerror.ErrAPIInvalidParam, err))
 		return
 	}
+	var pdClient pd.Client
+	var kvStorage kv.Storage
+	// if PDAddrs is empty, use the default pdClient
 	if len(cfg.PDAddrs) == 0 {
 		up, err := getCaptureDefaultUpstream(h.capture)
 		if err != nil {
 			_ = c.Error(err)
 			return
 		}
-		cfg.PDConfig = getUpstreamPDConfig(up)
-	}
-	credential := cfg.PDConfig.toCredential()
-
-	timeoutCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
-	defer cancel()
-	pdClient, err := h.helpers.getPDClient(timeoutCtx, cfg.PDAddrs, credential)
-	if err != nil {
-		_ = c.Error(cerror.WrapError(cerror.ErrAPIGetPDClientFailed, err))
-		return
-	}
-	defer pdClient.Close()
-
-	// verify tables todo: del kvstore
-	kvStorage, err := h.helpers.createTiStore(cfg.PDAddrs, credential)
-	if err != nil {
-		_ = c.Error(cerror.WrapError(cerror.ErrNewStore, err))
-		return
+		pdClient = up.PDClient
+		kvStorage = up.KVStorage
+	} else {
+		credential := cfg.PDConfig.toCredential()
+		timeoutCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
+		defer cancel()
+		var err error
+		pdClient, err = h.helpers.getPDClient(timeoutCtx, cfg.PDAddrs, credential)
+		if err != nil {
+			_ = c.Error(cerror.WrapError(cerror.ErrAPIGetPDClientFailed, err))
+			return
+		}
+		defer pdClient.Close()
+		// verify tables todo: del kvstore
+		kvStorage, err = h.helpers.createTiStore(cfg.PDAddrs, credential)
+		if err != nil {
+			_ = c.Error(cerror.WrapError(cerror.ErrNewStore, err))
+			return
+		}
 	}
 	// We should not close kvStorage since all kvStorage in cdc is the same one.
 	// defer kvStorage.Close()
@@ -245,21 +249,25 @@ func (h *OpenAPIV2) verifyTable(c *gin.Context) {
 		_ = c.Error(cerror.WrapError(cerror.ErrAPIInvalidParam, err))
 		return
 	}
+	var kvStore kv.Storage
+	// if PDAddrs is empty, use the default upstream
 	if len(cfg.PDAddrs) == 0 {
 		up, err := getCaptureDefaultUpstream(h.capture)
 		if err != nil {
 			_ = c.Error(err)
 			return
 		}
-		cfg.PDConfig = getUpstreamPDConfig(up)
+		kvStore = up.KVStorage
+	} else {
+		credential := cfg.PDConfig.toCredential()
+		var err error
+		kvStore, err = h.helpers.createTiStore(cfg.PDAddrs, credential)
+		if err != nil {
+			_ = c.Error(errors.Trace(err))
+			return
+		}
 	}
-	credential := cfg.PDConfig.toCredential()
 
-	kvStore, err := h.helpers.createTiStore(cfg.PDAddrs, credential)
-	if err != nil {
-		_ = c.Error(err)
-		return
-	}
 	replicaCfg := cfg.ReplicaConfig.ToInternalReplicaConfig()
 	ineligibleTables, eligibleTables, err := h.helpers.
 		getVerfiedTables(replicaCfg, kvStore, cfg.StartTs)
@@ -353,7 +361,7 @@ func (h *OpenAPIV2) updateChangefeed(c *gin.Context) {
 		return
 	}
 
-	var storage tidbkv.Storage
+	var storage kv.Storage
 	// if PDAddrs is not empty, use it to create a new kvstore
 	// Note: upManager is nil in some unit test cases
 	if len(updateCfConfig.PDAddrs) != 0 || upManager == nil {
@@ -789,48 +797,55 @@ func (h *OpenAPIV2) synced(c *gin.Context) {
 		cfg.ReplicaConfig.SyncedStatus.CheckpointInterval = status.CheckpointInterval
 		cfg.ReplicaConfig.SyncedStatus.SyncedCheckInterval = status.SyncedCheckInterval
 	}
+	if c.Request.Body != nil && c.Request.ContentLength > 0 {
+		if err := c.BindJSON(cfg); err != nil {
+			_ = c.Error(cerror.WrapError(cerror.ErrAPIInvalidParam, err))
+			return
+		}
+	}
 
 	// try to get pd client to get pd time, and determine synced status based on the pd time
+	var pdClient pd.Client
 	if len(cfg.PDAddrs) == 0 {
 		up, err := getCaptureDefaultUpstream(h.capture)
 		if err != nil {
 			_ = c.Error(err)
 			return
 		}
-		cfg.PDConfig = getUpstreamPDConfig(up)
-	}
-	credential := cfg.PDConfig.toCredential()
-
-	timeoutCtx, cancel := context.WithTimeout(ctx, timeout)
-	defer cancel()
+		pdClient = up.PDClient
+	} else {
+		credential := cfg.PDConfig.toCredential()
+		timeoutCtx, cancel := context.WithTimeout(ctx, timeout)
+		defer cancel()
 
-	pdClient, err := h.helpers.getPDClient(timeoutCtx, cfg.PDAddrs, credential)
-	if err != nil {
-		// case 1. we can't get pd client, pd may be unavailable.
-		//         if pullerResolvedTs - checkpointTs > checkpointInterval, data is not synced
-		//         otherwise, if pd is unavailable, we decide data whether is synced based on
-		//         the time difference between current time and lastSyncedTs.
-		var message string
-		if (oracle.ExtractPhysical(status.PullerResolvedTs) - oracle.ExtractPhysical(status.CheckpointTs)) >
-			cfg.ReplicaConfig.SyncedStatus.CheckpointInterval*1000 {
-			message = fmt.Sprintf("%s. Besides the data is not finish syncing", err.Error())
-		} else {
-			message = fmt.Sprintf("%s. You should check the pd status first. If pd status is normal, means we don't finish sync data. "+
-				"If pd is offline, please check whether we satisfy the condition that "+
-				"the time difference from lastSyncedTs to the current time from the time zone of pd is greater than %v secs. "+
-				"If it's satisfied, means the data syncing is totally finished", err, cfg.ReplicaConfig.SyncedStatus.SyncedCheckInterval)
+		pdClient, err = h.helpers.getPDClient(timeoutCtx, cfg.PDAddrs, credential)
+		if err != nil {
+			// case 1. we can't get pd client, pd may be unavailable.
+			//         if pullerResolvedTs - checkpointTs > checkpointInterval, data is not synced
+			//         otherwise, if pd is unavailable, we decide data whether is synced based on
+			//         the time difference between current time and lastSyncedTs.
+			var message string
+			if (oracle.ExtractPhysical(status.PullerResolvedTs) - oracle.ExtractPhysical(status.CheckpointTs)) >
+				cfg.ReplicaConfig.SyncedStatus.CheckpointInterval*1000 {
+				message = fmt.Sprintf("%s. Besides the data is not finish syncing", err.Error())
+			} else {
+				message = fmt.Sprintf("%s. You should check the pd status first. If pd status is normal, means we don't finish sync data. "+
+					"If pd is offline, please check whether we satisfy the condition that "+
+					"the time difference from lastSyncedTs to the current time from the time zone of pd is greater than %v secs. "+
+					"If it's satisfied, means the data syncing is totally finished", err, cfg.ReplicaConfig.SyncedStatus.SyncedCheckInterval)
+			}
+			c.JSON(http.StatusOK, SyncedStatus{
+				Synced:           false,
+				SinkCheckpointTs: model.JSONTime(oracle.GetTimeFromTS(status.CheckpointTs)),
+				PullerResolvedTs: model.JSONTime(oracle.GetTimeFromTS(status.PullerResolvedTs)),
+				LastSyncedTs:     model.JSONTime(oracle.GetTimeFromTS(status.LastSyncedTs)),
+				NowTs:            model.JSONTime(time.Unix(0, 0)),
+				Info:             message,
+			})
+			return
 		}
-		c.JSON(http.StatusOK, SyncedStatus{
-			Synced:           false,
-			SinkCheckpointTs: model.JSONTime(oracle.GetTimeFromTS(status.CheckpointTs)),
-			PullerResolvedTs: model.JSONTime(oracle.GetTimeFromTS(status.PullerResolvedTs)),
-			LastSyncedTs:     model.JSONTime(oracle.GetTimeFromTS(status.LastSyncedTs)),
-			NowTs:            model.JSONTime(time.Unix(0, 0)),
-			Info:             message,
-		})
-		return
+		defer pdClient.Close()
 	}
-	defer pdClient.Close()
 	// get time from pd
 	physicalNow, _, _ := pdClient.GetTS(ctx)
 
@@ -950,12 +965,3 @@ func getCaptureDefaultUpstream(cp capture.Capture) (*upstream.Upstream, error) {
 	}
 	return up, nil
 }
-
-func getUpstreamPDConfig(up *upstream.Upstream) PDConfig {
-	return PDConfig{
-		PDAddrs:  up.PdEndpoints,
-		KeyPath:  up.SecurityConfig.KeyPath,
-		CAPath:   up.SecurityConfig.CAPath,
-		CertPath: up.SecurityConfig.CertPath,
-	}
-}
diff --git a/cdc/kv/client.go b/cdc/kv/client.go
index 33e64b8ae..119425bbb 100644
--- a/cdc/kv/client.go
+++ b/cdc/kv/client.go
@@ -107,6 +107,7 @@ var (
 	metricFeedDuplicateRequestCounter = eventFeedErrorCounter.WithLabelValues("DuplicateRequest")
 	metricFeedUnknownErrorCounter     = eventFeedErrorCounter.WithLabelValues("Unknown")
 	metricFeedRPCCtxUnavailable       = eventFeedErrorCounter.WithLabelValues("RPCCtxUnavailable")
+	metricGetStoreErr                 = eventFeedErrorCounter.WithLabelValues("GetStoreErr")
 	metricStoreSendRequestErr         = eventFeedErrorCounter.WithLabelValues("SendRequestToStore")
 	metricConnectToStoreErr           = eventFeedErrorCounter.WithLabelValues("ConnectToStore")
 )
@@ -670,8 +671,17 @@ func (s *eventFeedSession) requestRegionToStore(
 					time.Sleep(delay)
 				}
 				bo := tikv.NewBackoffer(ctx, tikvRequestMaxBackoff)
-				s.client.regionCache.OnSendFail(bo, rpcCtx, regionScheduleReload, err)
-				errInfo := newRegionErrorInfo(sri, &connectToStoreErr{})
+				var regionErr error
+				var scheduleReload bool
+				if cerror.Is(err, cerror.ErrGetAllStoresFailed) {
+					regionErr = &getStoreErr{}
+					scheduleReload = true
+				} else {
+					regionErr = &connectToStoreErr{}
+					scheduleReload = regionScheduleReload
+				}
+				s.client.regionCache.OnSendFail(bo, rpcCtx, scheduleReload, err)
+				errInfo := newRegionErrorInfo(sri, regionErr)
 				s.onRegionFail(ctx, errInfo)
 				continue
 			}
@@ -946,6 +956,10 @@ func (s *eventFeedSession) handleError(ctx context.Context, errInfo regionErrorI
 		metricConnectToStoreErr.Inc()
 	case *sendRequestToStoreErr:
 		metricStoreSendRequestErr.Inc()
+	case *getStoreErr:
+		metricGetStoreErr.Inc()
+		s.scheduleDivideRegionAndRequest(ctx, errInfo.span)
+		return nil
 	default:
 		//[TODO] Move all OnSendFail logic here
 		// We expect some unknown error to trigger RegionCache recheck its store state and change leader to peer to
@@ -1502,3 +1516,7 @@ func (e *connectToStoreErr) Error() string { return "connect to store error" }
 type sendRequestToStoreErr struct{}
 
 func (e *sendRequestToStoreErr) Error() string { return "send request to store error" }
+
+type getStoreErr struct{}
+
+func (e *getStoreErr) Error() string { return "get store error" }
diff --git a/cdc/owner/ddl_manager.go b/cdc/owner/ddl_manager.go
index 83b2347bc..5cc9320a1 100644
--- a/cdc/owner/ddl_manager.go
+++ b/cdc/owner/ddl_manager.go
@@ -210,6 +210,8 @@ func (m *ddlManager) tick(
 		}
 
 		if job != nil && job.BinlogInfo != nil {
+			// Note: do not change the key words in the log, it is used to search the
+			// FinishTS of the DDL job. Some integration tests and users depend on it.
 			log.Info("handle a ddl job",
 				zap.String("namespace", m.changfeedID.Namespace),
 				zap.String("ID", m.changfeedID.ID),
diff --git a/cdc/processor/sourcemanager/manager.go b/cdc/processor/sourcemanager/manager.go
index 6306a612c..6d8df5697 100644
--- a/cdc/processor/sourcemanager/manager.go
+++ b/cdc/processor/sourcemanager/manager.go
@@ -165,10 +165,17 @@ func (m *SourceManager) Close() error {
 		zap.String("namespace", m.changefeedID.Namespace),
 		zap.String("changefeed", m.changefeedID.ID))
 	start := time.Now()
+
+	var wg sync.WaitGroup
 	m.pullers.Range(func(key, value interface{}) bool {
-		value.(*pullerwrapper.Wrapper).Close()
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			value.(*pullerwrapper.Wrapper).Close()
+		}()
 		return true
 	})
+	wg.Wait()
 	log.Info("All pullers have been closed",
 		zap.String("namespace", m.changefeedID.Namespace),
 		zap.String("changefeed", m.changefeedID.ID),
diff --git a/cdc/puller/ddl_puller.go b/cdc/puller/ddl_puller.go
index 98a438c84..4e7fd9405 100644
--- a/cdc/puller/ddl_puller.go
+++ b/cdc/puller/ddl_puller.go
@@ -74,7 +74,6 @@ type ddlJobPullerImpl struct {
 	kvStorage     tidbkv.Storage
 	schemaStorage entry.SchemaStorage
 	resolvedTs    uint64
-	schemaVersion int64
 	filter        filter.Filter
 	// ddlJobsTable is initialized when receive the first concurrent DDL job.
 	// It holds the info of table `tidb_ddl_jobs` of upstream TiDB.
@@ -363,8 +362,7 @@ func (p *ddlJobPullerImpl) handleJob(job *timodel.Job) (skip bool, err error) {
 			errors.Trace(err), job.Query, job.StartTS, job.StartTS)
 	}
 
-	if job.BinlogInfo.FinishedTS <= p.getResolvedTs() ||
-		job.BinlogInfo.SchemaVersion <= p.schemaVersion {
+	if job.BinlogInfo.FinishedTS <= p.getResolvedTs() {
 		log.Info("ddl job finishedTs less than puller resolvedTs,"+
 			"discard the ddl job",
 			zap.Uint64("jobFinishedTS", job.BinlogInfo.FinishedTS),
@@ -454,7 +452,6 @@ func (p *ddlJobPullerImpl) handleJob(job *timodel.Job) (skip bool, err error) {
 	}
 
 	p.setResolvedTs(job.BinlogInfo.FinishedTS)
-	p.schemaVersion = job.BinlogInfo.SchemaVersion
 
 	return p.checkIneligibleTableDDL(snap, job)
 }
diff --git a/cdc/redo/writer/memory/encoding_worker.go b/cdc/redo/writer/memory/encoding_worker.go
index 8930e7969..b7057fbb8 100644
--- a/cdc/redo/writer/memory/encoding_worker.go
+++ b/cdc/redo/writer/memory/encoding_worker.go
@@ -130,7 +130,7 @@ func (e *encodingWorkerGroup) Run(ctx context.Context) (err error) {
 			zap.String("namespace", e.changefeed.Namespace),
 			zap.String("changefeed", e.changefeed.ID),
 			zap.Error(err))
-		if err != nil && errors.Cause(err) != context.Canceled {
+		if err != nil {
 			e.closed <- err
 		}
 		close(e.closed)
diff --git a/cdc/sink/codec/avro/avro.go b/cdc/sink/codec/avro/avro.go
index ca1ddc612..e75f27a0a 100644
--- a/cdc/sink/codec/avro/avro.go
+++ b/cdc/sink/codec/avro/avro.go
@@ -401,8 +401,10 @@ func rowToAvroSchema(
 			}
 		} else {
 			if col.Flag.IsNullable() {
+				// the string literal "null" must be coerced to a `nil`
+				// see https://github.com/linkedin/goavro/blob/5ec5a5ee7ec82e16e6e2b438d610e1cab2588393/record.go#L109-L114
 				// https://stackoverflow.com/questions/22938124/avro-field-default-values
-				if defaultValue == nil {
+				if defaultValue == nil || defaultValue == "null" {
 					field["type"] = []interface{}{"null", avroType}
 				} else {
 					field["type"] = []interface{}{avroType, "null"}
diff --git a/cdc/sink/codec/canal/canal_json_encoder.go b/cdc/sink/codec/canal/canal_json_encoder.go
index 6235d0cff..a94c8bdc3 100644
--- a/cdc/sink/codec/canal/canal_json_encoder.go
+++ b/cdc/sink/codec/canal/canal_json_encoder.go
@@ -349,7 +349,8 @@ func (c *JSONBatchEncoder) AppendRowChangedEvent(
 	if err != nil {
 		return errors.Trace(err)
 	}
-	if len(c.config.Terminator) > 0 {
+
+	if c.config.IsStorageScheme && len(c.config.Terminator) > 0 {
 		value = append(value, c.config.Terminator...)
 	}
 
diff --git a/cdc/sink/codec/common/config.go b/cdc/sink/codec/common/config.go
index 2ce0c98d1..d974b393e 100644
--- a/cdc/sink/codec/common/config.go
+++ b/cdc/sink/codec/common/config.go
@@ -21,6 +21,7 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
+	"github.com/pingcap/tiflow/pkg/sink"
 	"go.uber.org/zap"
 )
 
@@ -47,6 +48,7 @@ type Config struct {
 	AvroBigintUnsignedHandlingMode string
 
 	// for sinking to cloud storage
+	IsStorageScheme      bool
 	Delimiter            string
 	Quote                string
 	NullString           string
@@ -162,6 +164,7 @@ func (c *Config) Apply(sinkURI *url.URL, config *config.ReplicaConfig) error {
 
 	}
 
+	c.IsStorageScheme = sink.IsStorageScheme(sinkURI.Scheme)
 	c.DeleteOnlyHandleKeyColumns = !config.EnableOldValue
 	return nil
 }
diff --git a/cdc/sink/codec/encoder_group.go b/cdc/sink/codec/encoder_group.go
index 603d629f1..5b6def80f 100644
--- a/cdc/sink/codec/encoder_group.go
+++ b/cdc/sink/codec/encoder_group.go
@@ -15,6 +15,7 @@ package codec
 
 import (
 	"context"
+	"runtime"
 	"strconv"
 	"sync/atomic"
 	"time"
@@ -61,7 +62,11 @@ func NewEncoderGroup(builder EncoderBuilder, count int, changefeedID model.Chang
 	if count <= 0 {
 		count = defaultEncoderGroupSize
 	}
-
+	limitCount := runtime.GOMAXPROCS(0) * 10
+	if count > limitCount {
+		count = limitCount
+		log.Warn("limit count to avoid crash", zap.Int("count", count), zap.Any("limitCount", limitCount))
+	}
 	inputCh := make([]chan *future, count)
 	for i := 0; i < count; i++ {
 		inputCh[i] = make(chan *future, defaultInputChanSize)
diff --git a/pkg/etcd/client.go b/pkg/etcd/client.go
index e63f77579..d00521793 100644
--- a/pkg/etcd/client.go
+++ b/pkg/etcd/client.go
@@ -358,6 +358,7 @@ func newClient(tlsConfig *tls.Config, grpcDialOption grpc.DialOption, endpoints
 		TLS:                  tlsConfig,
 		LogConfig:            &logConfig,
 		DialTimeout:          defaultEtcdClientTimeout,
+		AutoSyncInterval:     30 * time.Second,
 		DialKeepAliveTime:    defaultDialKeepAliveTime,
 		DialKeepAliveTimeout: defaultDialKeepAliveTimeout,
 		DialOptions: []grpc.DialOption{
diff --git a/pkg/filter/sql_event_filter.go b/pkg/filter/sql_event_filter.go
index 39235eb05..7e929878d 100644
--- a/pkg/filter/sql_event_filter.go
+++ b/pkg/filter/sql_event_filter.go
@@ -166,6 +166,18 @@ func (f *sqlEventFilter) shouldSkipDDL(ddl *model.DDLEvent) (bool, error) {
 	if err != nil {
 		return false, err
 	}
+	if evenType == bf.RenameTable {
+		if ddl.PreTableInfo == nil {
+			log.Warn("sql event filter doesn't find old table info when the event type is `rename table`",
+				zap.String("schema", schema),
+				zap.String("table", table),
+				zap.Stringer("type", ddl.Type),
+				zap.String("query", ddl.Query))
+			return true, cerror.ErrTableIneligible.GenWithStackByArgs(ddl.TableInfo.TableName)
+		}
+		schema = ddl.PreTableInfo.TableName.Schema
+		table = ddl.PreTableInfo.TableName.Table
+	}
 	if evenType == bf.NullEvent {
 		log.Warn("sql event filter unsupported ddl type, do nothing",
 			zap.String("type", ddl.Type.String()),
diff --git a/pkg/redo/config.go b/pkg/redo/config.go
index 8865e9f4f..4b7821b10 100644
--- a/pkg/redo/config.go
+++ b/pkg/redo/config.go
@@ -190,7 +190,7 @@ func IsBlackholeStorage(scheme string) bool {
 
 // InitExternalStorage init an external storage.
 var InitExternalStorage = func(ctx context.Context, uri url.URL) (storage.ExternalStorage, error) {
-	s, err := util.GetExternalStorageWithTimeout(ctx, uri.String(), DefaultTimeout)
+	s, err := util.GetExternalStorageWithDefaultTimeout(ctx, uri.String())
 	if err != nil {
 		return nil, errors.WrapError(errors.ErrStorageInitialize, err,
 			fmt.Sprintf("can't init external storage for %s", uri.String()))
diff --git a/pkg/util/external_storage.go b/pkg/util/external_storage.go
index ec65c380c..2f7f45bc5 100644
--- a/pkg/util/external_storage.go
+++ b/pkg/util/external_storage.go
@@ -34,6 +34,8 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
+const defaultTimeout = 5 * time.Minute
+
 // GetExternalStorageFromURI creates a new storage.ExternalStorage from a uri.
 func GetExternalStorageFromURI(
 	ctx context.Context, uri string,
@@ -41,18 +43,18 @@ func GetExternalStorageFromURI(
 	return GetExternalStorage(ctx, uri, nil, DefaultS3Retryer())
 }
 
-// GetExternalStorageWithTimeout creates a new storage.ExternalStorage from a uri
+// GetExternalStorageWithDefaultTimeout creates a new storage.ExternalStorage from a uri
 // without retry. It is the caller's responsibility to set timeout to the context.
-func GetExternalStorageWithTimeout(
-	ctx context.Context, uri string, timeout time.Duration,
-) (storage.ExternalStorage, error) {
-	ctx, cancel := context.WithTimeout(ctx, timeout)
+func GetExternalStorageWithDefaultTimeout(ctx context.Context, uri string) (storage.ExternalStorage, error) {
+	ctx, cancel := context.WithTimeout(ctx, defaultTimeout)
 	defer cancel()
-	s, err := GetExternalStorage(ctx, uri, nil, nil)
+	// total retry time is [1<<7, 1<<8] = [128, 256] + 30*6 = [308, 436] seconds
+	r := NewS3Retryer(7, 1*time.Second, 2*time.Second)
+	s, err := GetExternalStorage(ctx, uri, nil, r)
 
 	return &extStorageWithTimeout{
 		ExternalStorage: s,
-		timeout:         timeout,
+		timeout:         defaultTimeout,
 	}, err
 }
 
@@ -139,6 +141,17 @@ func DefaultS3Retryer() request.Retryer {
 	}
 }
 
+// NewS3Retryer creates a new s3 retryer.
+func NewS3Retryer(maxRetries int, minRetryDelay, minThrottleDelay time.Duration) request.Retryer {
+	return retryerWithLog{
+		DefaultRetryer: client.DefaultRetryer{
+			NumMaxRetries:    maxRetries,
+			MinRetryDelay:    minRetryDelay,
+			MinThrottleDelay: minThrottleDelay,
+		},
+	}
+}
+
 type extStorageWithTimeout struct {
 	storage.ExternalStorage
 	timeout time.Duration


Test patch:
diff --git a/cdc/api/v2/changefeed_test.go b/cdc/api/v2/changefeed_test.go
index 3e34bb507..bd69beb39 100644
--- a/cdc/api/v2/changefeed_test.go
+++ b/cdc/api/v2/changefeed_test.go
@@ -99,7 +99,7 @@ func TestCreateChangefeed(t *testing.T) {
 	}{
 		ID:      changeFeedID.ID,
 		SinkURI: blackholeSink,
-		PDAddrs: []string{},
+		PDAddrs: []string{"http://127.0.0.1:2379"}, // arbitrary pd address to trigger create new pd client
 	}
 	body, err := json.Marshal(&cfConfig)
 	require.Nil(t, err)
@@ -624,6 +624,8 @@ func TestVerifyTable(t *testing.T) {
 
 	// case 2: kv create failed
 	updateCfg := getDefaultVerifyTableConfig()
+	// arbitrary pd address to trigger create new pd client
+	updateCfg.PDAddrs = []string{"http://127.0.0.1:2379"}
 	body, err := json.Marshal(&updateCfg)
 	require.Nil(t, err)
 	helpers.EXPECT().
@@ -998,6 +1000,10 @@ func TestChangefeedSynced(t *testing.T) {
 	statusProvider.err = nil
 	statusProvider.changefeedInfo = cfInfo
 	{
+		cfg := getDefaultVerifyTableConfig()
+		// arbitrary pd address to trigger create new pd client
+		cfg.PDAddrs = []string{"http://127.0.0.1:2379"}
+		body, _ := json.Marshal(&cfg)
 		helpers.EXPECT().getPDClient(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, cerrors.ErrAPIGetPDClientFailed).Times(1)
 		// case3: pd is offline，resolvedTs - checkpointTs > 15s
 		statusProvider.changeFeedSyncedStatus = &model.ChangeFeedSyncedStatusForAPI{
@@ -1010,7 +1016,7 @@ func TestChangefeedSynced(t *testing.T) {
 			context.Background(),
 			syncedInfo.method,
 			fmt.Sprintf(syncedInfo.url, validID),
-			nil,
+			bytes.NewReader(body),
 		)
 		router.ServeHTTP(w, req)
 		require.Equal(t, http.StatusOK, w.Code)
@@ -1023,6 +1029,10 @@ func TestChangefeedSynced(t *testing.T) {
 	}
 
 	{
+		cfg := getDefaultVerifyTableConfig()
+		// arbitrary pd address to trigger create new pd client
+		cfg.PDAddrs = []string{"http://127.0.0.1:2379"}
+		body, _ := json.Marshal(&cfg)
 		helpers.EXPECT().getPDClient(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, cerrors.ErrAPIGetPDClientFailed).Times(1)
 		// case4: pd is offline，resolvedTs - checkpointTs < 15s
 		statusProvider.changeFeedSyncedStatus = &model.ChangeFeedSyncedStatusForAPI{
@@ -1035,7 +1045,7 @@ func TestChangefeedSynced(t *testing.T) {
 			context.Background(),
 			syncedInfo.method,
 			fmt.Sprintf(syncedInfo.url, validID),
-			nil,
+			bytes.NewReader(body),
 		)
 		router.ServeHTTP(w, req)
 		require.Equal(t, http.StatusOK, w.Code)
@@ -1054,6 +1064,10 @@ func TestChangefeedSynced(t *testing.T) {
 	pdClient.logicTime = 1000
 	pdClient.timestamp = 1701153217279
 	{
+		cfg := getDefaultVerifyTableConfig()
+		// arbitrary pd address to trigger create new pd client
+		cfg.PDAddrs = []string{"http://127.0.0.1:2379"}
+		body, _ := json.Marshal(&cfg)
 		// case5: pdTs - lastSyncedTs > 5min, pdTs - checkpointTs < 15s
 		statusProvider.changeFeedSyncedStatus = &model.ChangeFeedSyncedStatusForAPI{
 			CheckpointTs:     1701153217209 << 18,
@@ -1065,7 +1079,7 @@ func TestChangefeedSynced(t *testing.T) {
 			context.Background(),
 			syncedInfo.method,
 			fmt.Sprintf(syncedInfo.url, validID),
-			nil,
+			bytes.NewReader(body),
 		)
 		router.ServeHTTP(w, req)
 		require.Equal(t, http.StatusOK, w.Code)
@@ -1077,6 +1091,10 @@ func TestChangefeedSynced(t *testing.T) {
 	}
 
 	{
+		cfg := getDefaultVerifyTableConfig()
+		// arbitrary pd address to trigger create new pd client
+		cfg.PDAddrs = []string{"http://127.0.0.1:2379"}
+		body, _ := json.Marshal(&cfg)
 		// case6: pdTs - lastSyncedTs > 5min, pdTs - checkpointTs > 15s, resolvedTs - checkpointTs < 15s
 		statusProvider.changeFeedSyncedStatus = &model.ChangeFeedSyncedStatusForAPI{
 			CheckpointTs:     1701153201279 << 18,
@@ -1088,7 +1106,7 @@ func TestChangefeedSynced(t *testing.T) {
 			context.Background(),
 			syncedInfo.method,
 			fmt.Sprintf(syncedInfo.url, validID),
-			nil,
+			bytes.NewReader(body),
 		)
 		router.ServeHTTP(w, req)
 		require.Equal(t, http.StatusOK, w.Code)
@@ -1105,6 +1123,10 @@ func TestChangefeedSynced(t *testing.T) {
 	}
 
 	{
+		cfg := getDefaultVerifyTableConfig()
+		// arbitrary pd address to trigger create new pd client
+		cfg.PDAddrs = []string{"http://127.0.0.1:2379"}
+		body, _ := json.Marshal(&cfg)
 		// case7: pdTs - lastSyncedTs > 5min, pdTs - checkpointTs > 15s, resolvedTs - checkpointTs > 15s
 		statusProvider.changeFeedSyncedStatus = &model.ChangeFeedSyncedStatusForAPI{
 			CheckpointTs:     1701153201279 << 18,
@@ -1116,7 +1138,7 @@ func TestChangefeedSynced(t *testing.T) {
 			context.Background(),
 			syncedInfo.method,
 			fmt.Sprintf(syncedInfo.url, validID),
-			nil,
+			bytes.NewReader(body),
 		)
 		router.ServeHTTP(w, req)
 		require.Equal(t, http.StatusOK, w.Code)
@@ -1128,6 +1150,10 @@ func TestChangefeedSynced(t *testing.T) {
 	}
 
 	{
+		cfg := getDefaultVerifyTableConfig()
+		// arbitrary pd address to trigger create new pd client
+		cfg.PDAddrs = []string{"http://127.0.0.1:2379"}
+		body, _ := json.Marshal(&cfg)
 		// case8: pdTs - lastSyncedTs < 5min
 		statusProvider.changeFeedSyncedStatus = &model.ChangeFeedSyncedStatusForAPI{
 			CheckpointTs:     1701153217279 << 18,
@@ -1139,7 +1165,7 @@ func TestChangefeedSynced(t *testing.T) {
 			context.Background(),
 			syncedInfo.method,
 			fmt.Sprintf(syncedInfo.url, validID),
-			nil,
+			bytes.NewReader(body),
 		)
 		router.ServeHTTP(w, req)
 		require.Equal(t, http.StatusOK, w.Code)
diff --git a/cdc/redo/writer/memory/mem_log_writer_test.go b/cdc/redo/writer/memory/mem_log_writer_test.go
index d3b72541f..e8a21aea4 100644
--- a/cdc/redo/writer/memory/mem_log_writer_test.go
+++ b/cdc/redo/writer/memory/mem_log_writer_test.go
@@ -22,6 +22,7 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/redo/writer"
+	"github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/redo"
 	"github.com/pingcap/tiflow/pkg/util"
 	"github.com/stretchr/testify/require"
@@ -94,10 +95,45 @@ func testWriteEvents(t *testing.T, events []writer.RedoEvent) {
 	})
 	require.NoError(t, err)
 
+	require.ErrorIs(t, lw.Close(), context.Canceled)
+	// duplicate close should return the same error
 	require.ErrorIs(t, lw.Close(), context.Canceled)
 
-	err = lw.WriteEvents(ctx, events...)
-	require.NoError(t, err)
-	err = lw.FlushLog(ctx)
-	require.NoError(t, err)
+	functions := map[string]func(error){
+		"WriteEvents": func(expected error) {
+			if expected == nil {
+				err := lw.WriteEvents(ctx, events...)
+				require.NoError(t, err)
+			} else {
+				require.Eventually(
+					t, func() bool {
+						err := lw.WriteEvents(ctx, events...)
+						return errors.Is(errors.Cause(err), expected)
+					}, time.Second*2, time.Microsecond*10,
+				)
+			}
+		},
+		"FlushLog": func(expected error) {
+			if expected == nil {
+				err := lw.FlushLog(ctx)
+				require.NoError(t, err)
+			} else {
+				require.Eventually(
+					t, func() bool {
+						err := lw.WriteEvents(ctx, events...)
+						return errors.Is(errors.Cause(err), expected)
+					}, time.Second*2, time.Microsecond*10,
+				)
+			}
+		},
+	}
+	firstCall := true
+	for _, f := range functions {
+		if firstCall {
+			firstCall = false
+			f(context.Canceled)
+		} else {
+			f(nil)
+		}
+	}
 }
diff --git a/pkg/filter/sql_event_filter_test.go b/pkg/filter/sql_event_filter_test.go
index a564e29d7..ec5d59521 100644
--- a/pkg/filter/sql_event_filter_test.go
+++ b/pkg/filter/sql_event_filter_test.go
@@ -18,6 +18,7 @@ import (
 
 	"github.com/pingcap/errors"
 	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
+	timodel "github.com/pingcap/tidb/parser/model"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
@@ -27,10 +28,13 @@ import (
 func TestShouldSkipDDL(t *testing.T) {
 	t.Parallel()
 	type innerCase struct {
-		schema string
-		table  string
-		query  string
-		skip   bool
+		schema    string
+		table     string
+		preSchema string
+		preTable  string
+		query     string
+		ddlType   timodel.ActionType
+		skip      bool
 	}
 
 	type testCase struct {
@@ -51,28 +55,32 @@ func TestShouldSkipDDL(t *testing.T) {
 			},
 			cases: []innerCase{
 				{
-					schema: "test",
-					table:  "t1",
-					query:  "alter table t1 modify column age int",
-					skip:   true,
+					schema:  "test",
+					table:   "t1",
+					query:   "alter table t1 modify column age int",
+					ddlType: timodel.ActionModifyColumn,
+					skip:    true,
 				},
 				{
-					schema: "test",
-					table:  "t1",
-					query:  "create table t1(id int primary key)",
-					skip:   true,
+					schema:  "test",
+					table:   "t1",
+					query:   "create table t1(id int primary key)",
+					ddlType: timodel.ActionCreateTable,
+					skip:    true,
 				},
 				{
-					schema: "test",
-					table:  "t2", // table name not match
-					query:  "alter table t2 modify column age int",
-					skip:   false,
+					schema:  "test",
+					table:   "t2", // table name not match
+					query:   "alter table t2 modify column age int",
+					ddlType: timodel.ActionModifyColumn,
+					skip:    false,
 				},
 				{
-					schema: "test2", // schema name not match
-					table:  "t1",
-					query:  "alter table t1 modify column age int",
-					skip:   false,
+					schema:  "test2", // schema name not match
+					table:   "t1",
+					query:   "alter table t1 modify column age int",
+					ddlType: timodel.ActionModifyColumn,
+					skip:    false,
 				},
 			},
 		},
@@ -87,28 +95,32 @@ func TestShouldSkipDDL(t *testing.T) {
 			},
 			cases: []innerCase{
 				{
-					schema: "test",
-					table:  "t1",
-					query:  "alter table t1 modify column age int",
-					skip:   false,
+					schema:  "test",
+					table:   "t1",
+					query:   "alter table t1 modify column age int",
+					ddlType: timodel.ActionModifyColumn,
+					skip:    false,
 				},
 				{
-					schema: "test",
-					table:  "t1",
-					query:  "alter table t1 drop column age",
-					skip:   false,
+					schema:  "test",
+					table:   "t1",
+					query:   "alter table t1 drop column age",
+					ddlType: timodel.ActionModifyColumn,
+					skip:    false,
 				},
 				{
-					schema: "test2",
-					table:  "t1",
-					query:  "drop database test2",
-					skip:   true,
+					schema:  "test2",
+					table:   "t1",
+					query:   "drop database test2",
+					ddlType: timodel.ActionDropSchema,
+					skip:    true,
 				},
 				{
-					schema: "test3",
-					table:  "t1",
-					query:  "drop index i3 on t1",
-					skip:   false,
+					schema:  "test3",
+					table:   "t1",
+					query:   "drop index i3 on t1",
+					ddlType: timodel.ActionDropIndex,
+					skip:    false,
 				},
 			},
 		},
@@ -123,27 +135,31 @@ func TestShouldSkipDDL(t *testing.T) {
 			},
 			cases: []innerCase{
 				{
-					schema: "test",
-					table:  "t1",
-					query:  "ALTER TABLE t1 MODIFY COLUMN age int(11) NOT NULL",
-					skip:   true,
+					schema:  "test",
+					table:   "t1",
+					query:   "ALTER TABLE t1 MODIFY COLUMN age int(11) NOT NULL",
+					ddlType: timodel.ActionModifyColumn,
+					skip:    true,
 				},
 				{
-					schema: "test",
-					table:  "t1",
-					query:  "ALTER TABLE t1 DROP COLUMN age",
-					skip:   true,
+					schema:  "test",
+					table:   "t1",
+					query:   "ALTER TABLE t1 DROP COLUMN age",
+					ddlType: timodel.ActionDropColumn,
+					skip:    true,
 				},
 				{ // no table name
-					schema: "test2",
-					query:  "DROP DATABASE test",
-					skip:   true,
+					schema:  "test2",
+					query:   "DROP DATABASE test",
+					ddlType: timodel.ActionDropSchema,
+					skip:    true,
 				},
 				{
-					schema: "test3",
-					table:  "t1",
-					query:  "Drop Index i1 on test3.t1",
-					skip:   false,
+					schema:  "test3",
+					table:   "t1",
+					query:   "Drop Index i1 on test3.t1",
+					ddlType: timodel.ActionDropIndex,
+					skip:    false,
 				},
 			},
 		},
@@ -168,6 +184,47 @@ func TestShouldSkipDDL(t *testing.T) {
 				},
 			},
 		},
+		// rename table
+		{
+			cfg: &config.FilterConfig{
+				Rules: []string{"*.t1", "*.t2"},
+				EventFilters: []*config.EventFilterRule{
+					{
+						Matcher:     []string{"*.t1"},
+						IgnoreEvent: []bf.EventType{bf.AllDDL},
+					},
+				},
+			},
+			cases: []innerCase{
+				{
+					schema:    "test",
+					table:     "t2",
+					preSchema: "test",
+					preTable:  "t1",
+					query:     "RENAME TABLE t1 to t2",
+					ddlType:   timodel.ActionRenameTable,
+					skip:      true,
+				},
+				{
+					schema:    "test",
+					table:     "t3",
+					preSchema: "test",
+					preTable:  "t1",
+					query:     "RENAME TABLE t1 to t3",
+					ddlType:   timodel.ActionRenameTable,
+					skip:      true,
+				},
+				{
+					schema:    "test",
+					table:     "t1",
+					preSchema: "test",
+					preTable:  "t2",
+					query:     "RENAME TABLE t2 to t1",
+					ddlType:   timodel.ActionRenameTable,
+					skip:      false,
+				},
+			},
+		},
 	}
 
 	for _, tc := range testCases {
@@ -181,7 +238,14 @@ func TestShouldSkipDDL(t *testing.T) {
 						Table:  c.table,
 					},
 				},
+				PreTableInfo: &model.TableInfo{
+					TableName: model.TableName{
+						Schema: c.preSchema,
+						Table:  c.preTable,
+					},
+				},
 				Query: c.query,
+				Type:  c.ddlType,
 			}
 			skip, err := f.shouldSkipDDL(ddl)
 			require.NoError(t, err)
diff --git a/tests/integration_tests/availability/owner.sh b/tests/integration_tests/availability/owner.sh
index b78c0ecd7..82eb2ad69 100755
--- a/tests/integration_tests/availability/owner.sh
+++ b/tests/integration_tests/availability/owner.sh
@@ -13,6 +13,7 @@ function test_owner_ha() {
 	test_owner_retryable_error
 	test_gap_between_watch_capture
 	test_delete_owner_key
+	test_resign_owner
 }
 # test_kill_owner starts two captures and kill the owner
 # we expect the live capture will be elected as the new
@@ -219,7 +220,7 @@ function test_delete_owner_key() {
 	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY --addr "127.0.0.1:8301" --logsuffix test_gap_between_watch_capture.server2
 	ensure $MAX_RETRIES "$CDC_BINARY cli capture list 2>&1 | grep -v \"$owner_id\" | grep id"
 	capture_pid=$(ps -C $CDC_BINARY -o pid= | awk '{print $1}' | grep -v "$owner_pid")
-	capture_id=$($CDC_BINARY cli capture list 2>&1 | awk -F '"' '/id/{print $4}' | grep -v "$owner_id")
+	capture_id=$($CDC_BINARY cli capture list 2>&1 | awk -F '"' '/\"id/{print $4}' | grep -v "$owner_id")
 	echo "capture_id:" $capture_id
 
 	etcdctl del $owner_key
@@ -242,3 +243,30 @@ function test_delete_owner_key() {
 	echo "delete_owner_key pass"
 	cleanup_process $CDC_BINARY
 }
+
+# test_resign_owner resign the owner by sending
+# the resign owner v2 API
+# We expect when the owner is resigned, the new owner will be elected
+function test_resign_owner() {
+	echo "run test case test_resign_owner"
+	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY --addr "127.0.0.1:8300" --logsuffix test_resign_owner.server1
+	# ensure the server become the owner
+	ensure $MAX_RETRIES "$CDC_BINARY cli capture list 2>&1 | grep '\"is-owner\": true'"
+	owner_pid=$(ps -C $CDC_BINARY -o pid= | awk '{print $1}')
+	owner_id=$($CDC_BINARY cli capture list 2>&1 | awk -F '"' '/\"id/{print $4}')
+	echo "owner pid:" $owner_pid
+	echo "owner id" $owner_id
+
+	# run another server
+	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY --addr "127.0.0.1:8301" --logsuffix test_resign_owner.server2
+	ensure $MAX_RETRIES "$CDC_BINARY cli capture list 2>&1 | grep -v \"$owner_id\" | grep id"
+	capture_id=$($CDC_BINARY cli capture list 2>&1 | awk -F '"' '/\"id/{print $4}' | grep -v "$owner_id")
+	echo "capture_id:" $capture_id
+
+	# resign the owner
+	curl -X POST http://127.0.0.1:8301/api/v2/owner/resign
+	# check that the new owner is elected
+	ensure $MAX_RETRIES "$CDC_BINARY cli capture list --server 'http://127.0.0.1:8301' 2>&1 |grep $capture_id -A1 | grep '\"is-owner\": true'"
+	echo "test_resign_owner: pass"
+	cleanup_process $CDC_BINARY
+}
diff --git a/tests/integration_tests/bank/case.go b/tests/integration_tests/bank/case.go
index bd1a5794d..72adf2845 100644
--- a/tests/integration_tests/bank/case.go
+++ b/tests/integration_tests/bank/case.go
@@ -14,10 +14,16 @@
 package main
 
 import (
+	"bufio"
 	"context"
 	"database/sql"
 	"fmt"
+	"io"
 	"math/rand"
+	"os"
+	"path/filepath"
+	"regexp"
+	"strconv"
 	"strings"
 	"sync/atomic"
 	"time"
@@ -619,10 +625,11 @@ func getDownStreamSyncedEndTs(ctx context.Context, db *sql.DB, tableName string)
 		case <-ctx.Done():
 			log.Error("get downstream sync end ts failed due to timeout", zap.String("table", tableName), zap.Error(ctx.Err()))
 			return "", ctx.Err()
-		case <-time.After(2 * time.Second):
-			result, ok := tryGetEndTs(db, tableName)
+		case <-time.After(15 * time.Second):
+			// result, ok := tryGetEndTs(db, tidbAPIEndpoint, tableName)
+			result, ok := tryGetEndTsFromLog(db, tableName)
 			if ok {
-				return result, nil
+				return strconv.Itoa(int(result)), nil
 			}
 		}
 	}
@@ -642,3 +649,65 @@ func tryGetEndTs(db *sql.DB, tableName string) (result string, ok bool) {
 
 	return endTime, true
 }
+
+func tryGetEndTsFromLog(_ *sql.DB, tableName string) (result uint64, ok bool) {
+	log.Info("try parse finishedTs from ticdc log", zap.String("tableName", tableName))
+
+	logFilePath := "/tmp/tidb_cdc_test/bank"
+	cdcLogFiles := make([]string, 0)
+	// walk all file with cdc prefix
+	err := filepath.WalkDir(logFilePath, func(path string, d os.DirEntry, err error) error {
+		if err != nil {
+			return err
+		}
+		if !d.IsDir() {
+			if strings.Contains(d.Name(), "down") && strings.Contains(d.Name(), "cdc") && strings.Contains(d.Name(), "log") {
+				cdcLogFiles = append(cdcLogFiles, path)
+				fmt.Println(path)
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		log.Error("Failed to walk dir: %v", zap.Error(err))
+	}
+	log.Info("total files", zap.Any("file", cdcLogFiles))
+
+	logRegex := regexp.MustCompile(`handle a ddl job`)
+	tableNameRegex := regexp.MustCompile(tableName + "`")
+	timeStampRegex := regexp.MustCompile(`finishedTs=([0-9]+)`)
+	for _, f := range cdcLogFiles {
+		file, err := os.Open(f)
+		if err != nil {
+			log.Error("Failed to open file: %v", zap.Error(err))
+		}
+		defer file.Close()
+
+		reader := bufio.NewReader(file)
+		for {
+			bs, _, err := reader.ReadLine()
+			if err != nil {
+				if err != io.EOF {
+					fmt.Printf("Error reading file: %v\n", err)
+				}
+				return 0, false
+			}
+			line := string(bs)
+			if !logRegex.MatchString(line) || !tableNameRegex.MatchString(line) {
+				continue
+			}
+
+			matches := timeStampRegex.FindStringSubmatch(line)
+			if len(matches) > 1 {
+				fmt.Println("found first match line, Match Result: ", matches[1], ", line: ", line)
+				// convert to uint64
+				result, err := strconv.ParseUint(matches[1], 10, 64)
+				if err != nil {
+					log.Error("Failed to parse uint64: %v", zap.Error(err))
+				}
+				return result, true
+			}
+		}
+	}
+	return 0, false
+}
diff --git a/tests/integration_tests/bank/run.sh b/tests/integration_tests/bank/run.sh
index d883e16dc..069d5946a 100644
--- a/tests/integration_tests/bank/run.sh
+++ b/tests/integration_tests/bank/run.sh
@@ -24,8 +24,10 @@ function prepare() {
 	run_sql "CREATE DATABASE bank" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 
 	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
-
 	run_cdc_cli changefeed create --sink-uri="mysql://root@${DOWN_TIDB_HOST}:${DOWN_TIDB_PORT}/"
+
+	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY --addr "127.0.0.1:8400" --pd "http://${DOWN_PD_HOST}:${DOWN_PD_PORT}" --logsuffix "down"
+	run_cdc_cli changefeed create --sink-uri="blackhole://" -c "changefeed-for-find-finished-ts" --server "http://127.0.0.1:8400"
 }
 
 trap stop_tidb_cluster EXIT
diff --git a/tests/integration_tests/event_filter/conf/cf.toml b/tests/integration_tests/event_filter/conf/cf.toml
index d3a36188b..4597fe3e4 100644
--- a/tests/integration_tests/event_filter/conf/cf.toml
+++ b/tests/integration_tests/event_filter/conf/cf.toml
@@ -13,3 +13,7 @@ ignore-event = ["truncate table"]
 [[filter.event-filters]]
 matcher = ["event_filter.t_alter"]
 ignore-event = ["alter table"]
+
+[[filter.event-filters]]
+matcher = ["event_filter.t_name*"]
+ignore-event = ["rename table"]
\ No newline at end of file
diff --git a/tests/integration_tests/event_filter/conf/diff_config.toml b/tests/integration_tests/event_filter/conf/diff_config.toml
index fff11e2ee..ae516cbae 100644
--- a/tests/integration_tests/event_filter/conf/diff_config.toml
+++ b/tests/integration_tests/event_filter/conf/diff_config.toml
@@ -13,7 +13,7 @@ source-instances = ["tidb0"]
 
 target-instance = "mysql1"
 
-target-check-tables = ["event_filter.t_normal", "event_filter.t_truncate", "event_filter.t_alter"]
+target-check-tables = ["event_filter.t_normal", "event_filter.t_truncate", "event_filter.t_alter", "event_filter.t_name*", "event_filter.t_rename*"]
 
 [data-sources]
 [data-sources.tidb0]
diff --git a/tests/integration_tests/event_filter/data/test.sql b/tests/integration_tests/event_filter/data/test.sql
index 6af0efed2..5ffd396ae 100644
--- a/tests/integration_tests/event_filter/data/test.sql
+++ b/tests/integration_tests/event_filter/data/test.sql
@@ -89,4 +89,25 @@ CREATE TABLE t_alter
 ) ENGINE = InnoDB
   DEFAULT CHARSET = utf8
   COLLATE = utf8_bin;
-  
\ No newline at end of file
+  
+
+CREATE TABLE t_name (
+                    id INT,
+                    name varchar(128),
+                    PRIMARY KEY (id)
+);
+CREATE TABLE t_name1 (
+                    id INT,
+                    name varchar(128),
+                    PRIMARY KEY (id)
+);
+CREATE TABLE t_name2 (
+                    id INT,
+                    name varchar(128),
+                    PRIMARY KEY (id)
+);
+CREATE TABLE t_name3 (
+                    id INT,
+                    name varchar(128),
+                    PRIMARY KEY (id)
+);
\ No newline at end of file
diff --git a/tests/integration_tests/event_filter/data/test_rename.sql b/tests/integration_tests/event_filter/data/test_rename.sql
new file mode 100644
index 000000000..01df4e517
--- /dev/null
+++ b/tests/integration_tests/event_filter/data/test_rename.sql
@@ -0,0 +1,27 @@
+USE `event_filter`;
+
+RENAME TABLE t_name TO t_rename;
+
+INSERT INTO t_rename
+VALUES (1, 'guagua');
+
+INSERT INTO t_rename
+VALUES (2, 'huahua');
+
+INSERT INTO t_rename
+VALUES (3, 'xigua');
+
+INSERT INTO t_rename
+VALUES (4, 'yuko');
+
+-- rename tables
+RENAME TABLE t_name1 TO t_rename1, t_name2 TO t_rename2, t_name3 TO t_rename3;
+
+INSERT INTO t_rename1
+VALUES (1, 'guagua');
+
+INSERT INTO t_rename2
+VALUES (2, 'huahua');
+
+INSERT INTO t_rename3
+VALUES (3, 'xigua');
diff --git a/tests/integration_tests/event_filter/run.sh b/tests/integration_tests/event_filter/run.sh
index 6e9385052..ff4dbce92 100644
--- a/tests/integration_tests/event_filter/run.sh
+++ b/tests/integration_tests/event_filter/run.sh
@@ -37,6 +37,10 @@ function run() {
 	check_table_exists "event_filter.t_normal" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 	check_table_exists "event_filter.t_truncate" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 	check_table_exists "event_filter.t_alter" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	check_table_exists "event_filter.t_name" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	check_table_exists "event_filter.t_name1" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	check_table_exists "event_filter.t_name2" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	check_table_exists "event_filter.t_name3" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 
 	# check those rows that are not filtered are synced to downstream
 	run_sql "select count(1) from event_filter.t1;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
@@ -54,6 +58,11 @@ function run() {
 	run_sql_file $CUR/data/test_truncate.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 	run_sql "ALTER TABLE event_filter.t_alter MODIFY t_bigint BIGINT;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 	run_sql_file $CUR/data/test_alter.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	run_sql "RENAME TABLE event_filter.t_name TO event_filter.t_rename;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	run_sql "RENAME TABLE event_filter.t_name1 TO event_filter.t_rename1;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	run_sql "RENAME TABLE event_filter.t_name2 TO event_filter.t_rename2;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	run_sql "RENAME TABLE event_filter.t_name3 TO event_filter.t_rename3;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	run_sql_file $CUR/data/test_rename.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 	run_sql "create table event_filter.finish_mark(id int primary key);" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 	check_table_exists "event_filter.finish_mark" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 
diff --git a/tests/integration_tests/run_group.sh b/tests/integration_tests/run_group.sh
index e45d88251..da042c89e 100755
--- a/tests/integration_tests/run_group.sh
+++ b/tests/integration_tests/run_group.sh
@@ -10,7 +10,7 @@ group=$2
 # Other tests that only support mysql: batch_update_to_no_batch ddl_reentrant
 # changefeed_fast_fail changefeed_resume_with_checkpoint_ts sequence
 # multi_cdc_cluster capture_suicide_while_balance_table
-mysql_only="bdr_mode capture_suicide_while_balance_table syncpoint server_config_compatibility changefeed_dup_error_restart"
+mysql_only="bdr_mode capture_suicide_while_balance_table syncpoint server_config_compatibility changefeed_dup_error_restart safe_mode"
 mysql_only_http="http_api http_api_tls api_v2"
 mysql_only_consistent_replicate="consistent_replicate_ddl consistent_replicate_gbk consistent_replicate_nfs consistent_replicate_storage_file consistent_replicate_storage_file_large_value consistent_replicate_storage_s3 consistent_partition_table"
 
diff --git a/tests/integration_tests/safe_mode/conf/diff_config.toml b/tests/integration_tests/safe_mode/conf/diff_config.toml
new file mode 100644
index 000000000..7bf5d3ffe
--- /dev/null
+++ b/tests/integration_tests/safe_mode/conf/diff_config.toml
@@ -0,0 +1,29 @@
+# diff Configuration.
+
+check-thread-count = 4
+
+export-fix-sql = true
+
+check-struct-only = false
+
+[task]
+    output-dir = "/tmp/tidb_cdc_test/safe_mode/sync_diff/output"
+
+    source-instances = ["mysql1"]
+
+    target-instance = "tidb0"
+
+    target-check-tables = ["safe_mode.t"]
+
+[data-sources]
+[data-sources.mysql1]
+    host = "127.0.0.1"
+    port = 4000
+    user = "root"
+    password = ""
+
+[data-sources.tidb0]
+    host = "127.0.0.1"
+    port = 3306
+    user = "root"
+    password = ""
diff --git a/tests/integration_tests/safe_mode/data/create_table.sql b/tests/integration_tests/safe_mode/data/create_table.sql
new file mode 100644
index 000000000..9ed09d75a
--- /dev/null
+++ b/tests/integration_tests/safe_mode/data/create_table.sql
@@ -0,0 +1,5 @@
+drop database if exists `safe_mode`;
+create database `safe_mode`;
+use `safe_mode`;
+
+create table t(id int key, a varchar(200));
diff --git a/tests/integration_tests/safe_mode/data/insert.sql b/tests/integration_tests/safe_mode/data/insert.sql
new file mode 100644
index 000000000..2cd440031
--- /dev/null
+++ b/tests/integration_tests/safe_mode/data/insert.sql
@@ -0,0 +1,3 @@
+use `safe_mode`;
+insert into t values(1, "hello");
+insert into t values(2, "world");
diff --git a/tests/integration_tests/safe_mode/data/update.sql b/tests/integration_tests/safe_mode/data/update.sql
new file mode 100644
index 000000000..97963570c
--- /dev/null
+++ b/tests/integration_tests/safe_mode/data/update.sql
@@ -0,0 +1,4 @@
+use `safe_mode`;
+-- update non key column
+update t set a = "hello2" where id = 1;
+update t set a = "world2" where id = 2;
diff --git a/tests/integration_tests/safe_mode/run.sh b/tests/integration_tests/safe_mode/run.sh
new file mode 100755
index 000000000..eac8e3c83
--- /dev/null
+++ b/tests/integration_tests/safe_mode/run.sh
@@ -0,0 +1,51 @@
+#!/bin/bash
+
+set -eu
+
+CUR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+source $CUR/../_utils/test_prepare
+WORK_DIR=$OUT_DIR/$TEST_NAME
+CDC_BINARY=cdc.test
+SINK_TYPE=$1
+
+function run() {
+	if [ "$SINK_TYPE" != "mysql" ]; then
+		return
+	fi
+
+	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
+
+	start_tidb_cluster --workdir $WORK_DIR
+
+	cd $WORK_DIR
+
+	run_sql_file $CUR/data/create_table.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	run_sql_file $CUR/data/create_table.sql ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+
+	# insert data into upstream but not downstream
+	run_sql_file $CUR/data/insert.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+
+	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
+
+	case $SINK_TYPE in
+	*) SINK_URI="mysql://normal:123456@127.0.0.1:3306/?safe-mode=true" ;;
+	esac
+	run_cdc_cli changefeed create --sink-uri="$SINK_URI"
+
+	# test update sql can be split into delete + replace at all times in safe mode
+	# otherwise the update sql will have no effect on the downstream and the downstream will have no data.
+	sleep 10
+	run_sql_file $CUR/data/update.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+
+	run_sql "CREATE TABLE safe_mode.finish_mark (a int primary key);"
+	sleep 30
+	check_table_exists "safe_mode.finish_mark" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 60
+	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
+
+	cleanup_process $CDC_BINARY
+}
+
+trap stop_tidb_cluster EXIT
+run $*
+check_logs $WORK_DIR
+echo "[$(date)] <<<<<< run test case $TEST_NAME success! >>>>>>"