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

cdc/cdc/kv/client_test.go

@@ -410,14 +410,14 @@ func (s *clientSuite) TestConnectOfflineTiKV(c *check.C) {
 	case <-time.After(time.Second):
 		c.Fatalf("reconnection not succeed in 1 second")
 	}
-	checkEvent(event, 1)
+	checkEvent(event, GetSafeResolvedTs(1))
 
 	select {
 	case event = <-eventCh:
 	case <-time.After(time.Second):
 		c.Fatalf("reconnection not succeed in 1 second")
 	}
-	checkEvent(event, ver.Ver)
+	checkEvent(event, GetSafeResolvedTs(ver.Ver))
 
 	// check gRPC connection active counter is updated correctly
 	bucket, ok := grpcPool.bucketConns[invalidStore]

cdc/cdc/kv/region_worker.go

@@ -730,31 +730,36 @@ func (w *regionWorker) handleResolvedTs(
 		return nil
 	}
 	regionID := state.sri.verID.GetID()
+
+	// In TiKV, hen a leader transfer occurs, the old leader may send the last
+	// resolved ts, which may be larger than the new leader appends key to ts.
+	// So we fallback the resolved ts to a safe interval to make sure it's correct.
+	safeResolvedTs := GetSafeResolvedTs(resolvedTs)
 	// 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 +833,21 @@ func RunWorkerPool(ctx context.Context) error {
 	})
 	return errg.Wait()
 }
+
+func GetSafeResolvedTs(resolvedTs uint64) uint64 {
+	cfg := config.GetGlobalServerConfig().KVClient
+
+	logicalTs := oracle.ExtractLogical(resolvedTs)
+	physicalTime := oracle.GetTimeFromTS(resolvedTs)
+
+	safeTime := physicalTime.Add(-cfg.ResolvedTsSafeInterval)
+	physicalTs := oracle.GetPhysical(safeTime)
+
+	if physicalTs < 0 {
+		log.Warn("The resolvedTs is smaller than the ResolvedTsSafeInterval",
+			zap.Uint64("resolvedTs", resolvedTs))
+		return resolvedTs
+	}
+
+	return oracle.ComposeTS(physicalTs, logicalTs)
+}

cdc/cdc/kv/region_worker_test.go

@@ -135,3 +135,30 @@ func (s *regionWorkerSuite) TestRegionWorkerPoolSize(c *check.C) {
 	size = getWorkerPoolSize()
 	c.Assert(size, check.Equals, maxWorkerPoolSize)
 }
+
+func (s *regionWorkerSuite) TestGetSafeResolvedTs(c *check.C) {
+	defer testleak.AfterTest(c)()
+
+	testCases := []struct {
+		resolvedTs uint64
+		expected   uint64
+	}{
+		{
+			resolvedTs: 10,
+			expected:   10,
+		},
+		{
+			resolvedTs: (1<<18)*3*1000 + 1,
+			expected:   1,
+		},
+		{
+			resolvedTs: (1<<18)*4*1000 + 1,
+			expected:   (1<<18)*1000 + 1,
+		},
+	}
+
+	for _, testCase := range testCases {
+		safeResolvedTs := GetSafeResolvedTs(testCase.resolvedTs)
+		c.Assert(testCase.expected, check.Equals, safeResolvedTs)
+	}
+}

cdc/pkg/cmd/server/server_test.go

@@ -170,9 +170,10 @@ func TestParseCfg(t *testing.T) {
 		},
 		PerKeySpanMemoryQuota: 10 * 1024 * 1024, // 10M
 		KVClient: &config.KVClientConfig{
-			WorkerConcurrent: 8,
-			WorkerPoolSize:   0,
-			RegionScanLimit:  40,
+			WorkerConcurrent:       8,
+			WorkerPoolSize:         0,
+			RegionScanLimit:        40,
+			ResolvedTsSafeInterval: 3 * time.Second,
 		},
 		Debug: &config.DebugConfig{
 			EnableKeySpanActor: false,
@@ -299,9 +300,10 @@ server-worker-pool-size = 16
 		Security:              &config.SecurityConfig{},
 		PerKeySpanMemoryQuota: 10 * 1024 * 1024, // 10M
 		KVClient: &config.KVClientConfig{
-			WorkerConcurrent: 8,
-			WorkerPoolSize:   0,
-			RegionScanLimit:  40,
+			WorkerConcurrent:       8,
+			WorkerPoolSize:         0,
+			RegionScanLimit:        40,
+			ResolvedTsSafeInterval: 3 * time.Second,
 		},
 		Debug: &config.DebugConfig{
 			EnableKeySpanActor: false,
@@ -427,9 +429,10 @@ cert-allowed-cn = ["dd","ee"]
 		},
 		PerKeySpanMemoryQuota: 10 * 1024 * 1024, // 10M
 		KVClient: &config.KVClientConfig{
-			WorkerConcurrent: 8,
-			WorkerPoolSize:   0,
-			RegionScanLimit:  40,
+			WorkerConcurrent:       8,
+			WorkerPoolSize:         0,
+			RegionScanLimit:        40,
+			ResolvedTsSafeInterval: 3 * time.Second,
 		},
 		Debug: &config.DebugConfig{
 			EnableKeySpanActor: false,

cdc/pkg/config/config_test_data.go

@@ -81,6 +81,7 @@ const (
     "worker-concurrent": 8,
     "worker-pool-size": 0,
     "region-scan-limit": 40
+	"resolvedts-safe-interval": 3000000000,
   },
   "debug": {
     "enable-keyspan-actor": false,

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:"reoslved-ts-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,