[fix #193] fix cdc lose data by add resolved ts interval

cdc/cdc/kv/region_worker.go

@@ -730,31 +730,39 @@ func (w *regionWorker) handleResolvedTs(
 		return nil
 	}
 	regionID := state.sri.verID.GetID()
+
+	safeResolvedTs := GetSafeResolvedTs(resolvedTs)
+	if safeResolvedTs == 0 {
+		log.Warn("The resolvedTs is smaller than the ResolvedTsSafeInterval",
+			zap.Uint64("resolvedTs", resolvedTs),
+			zap.Uint64("regionID", regionID))
+		return nil
+	}
 	// Send resolved ts update in non blocking way, since we can re-query real
 	// resolved ts from region state even if resolved ts update is discarded.
 	// NOTICE: We send any regionTsInfo to resolveLock thread to give us a chance to trigger resolveLock logic
 	// (1) if it is a fallback resolvedTs event, it will be discarded and accumulate penalty on the progress;
 	// (2) if it is a normal one, update rtsManager and check sinceLastResolvedTs
 	select {
-	case w.rtsUpdateCh <- &regionTsInfo{regionID: regionID, ts: newResolvedTsItem(resolvedTs)}:
+	case w.rtsUpdateCh <- &regionTsInfo{regionID: regionID, ts: newResolvedTsItem(safeResolvedTs)}:
 	default:
 	}
 
-	if resolvedTs < state.lastResolvedTs {
+	if safeResolvedTs < state.lastResolvedTs {
 		log.Warn("The resolvedTs is fallen back in kvclient",
 			zap.String("Event Type", "RESOLVED"),
-			zap.Uint64("resolvedTs", resolvedTs),
+			zap.Uint64("safeResolvedTs", safeResolvedTs),
 			zap.Uint64("lastResolvedTs", state.lastResolvedTs),
 			zap.Uint64("regionID", regionID))
 		return nil
 	}
-	state.lastResolvedTs = resolvedTs
+	state.lastResolvedTs = safeResolvedTs
 	// emit a checkpointTs
 	revent := model.RegionFeedEvent{
 		RegionID: regionID,
 		Resolved: &model.ResolvedSpan{
 			Span:       state.sri.span,
-			ResolvedTs: resolvedTs,
+			ResolvedTs: safeResolvedTs,
 		},
 	}
 
@@ -828,3 +836,18 @@ func RunWorkerPool(ctx context.Context) error {
 	})
 	return errg.Wait()
 }
+
+func GetSafeResolvedTs(resolvedTs uint64) uint64 {
+	cfg := config.GetGlobalServerConfig().KVClient
+
+	logicalTs := oracle.ExtractLogical(resolvedTs)
+	time := oracle.GetTimeFromTS(resolvedTs)
+
+	safeTime := time.Add(-cfg.ResolvedTsSafeInterval)
+	physicalTs := oracle.GetPhysical(safeTime)
+	if physicalTs < 0 {
+		return 0
+	}
+
+	return oracle.ComposeTS(physicalTs, logicalTs)
+}

cdc/pkg/config/kvclient.go

@@ -13,6 +13,8 @@
 
 package config
 
+import "time"
+
 // KVClientConfig represents config for kv client
 type KVClientConfig struct {
 	// how many workers will be used for a single region worker
@@ -21,4 +23,6 @@ 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 safe interval to move forward resolved ts
+	ResolvedTsSafeInterval time.Duration `toml:"reoslvedts-safe-interval" json:"reoslvedts-safe-interval"`
 }

cdc/pkg/config/server_config.go

@@ -93,9 +93,10 @@ var defaultServerConfig = &ServerConfig{
 	Security:              &SecurityConfig{},
 	PerKeySpanMemoryQuota: 10 * 1024 * 1024, // 10MB
 	KVClient: &KVClientConfig{
-		WorkerConcurrent: 8,
-		WorkerPoolSize:   0, // 0 will use NumCPU() * 2
-		RegionScanLimit:  40,
+		WorkerConcurrent:       8,
+		WorkerPoolSize:         0, // 0 will use NumCPU() * 2
+		RegionScanLimit:        40,
+		ResolvedTsSafeInterval: 3 * time.Second,
 	},
 	Debug: &DebugConfig{
 		EnableKeySpanActor: false,