Merge branch 'release-4.0' into cherry-pick-3540-to-release-4.0

pingcap · Nov 23, 2021 · 4016739 · 4016739
2 parents c8337bd + 248a2c0
commit 4016739
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 4 deletions.
diff --git a/cdc/owner/gc_manager.go b/cdc/owner/gc_manager.go
@@ -81,9 +81,11 @@ func (m *gcManager) updateGCSafePoint(ctx cdcContext.Context, state *model.Globa
 		default:
 			continue
 		}
-		checkpointTs := cfState.Info.GetCheckpointTs(cfState.Status)
-		if minCheckpointTs > checkpointTs {
-			minCheckpointTs = checkpointTs
+		// When the changefeed starts up, CDC will do a snapshot read at (checkpoint-ts - 1) from TiKV,
+		// so (checkpoint - 1) should be an upper bound for the GC safepoint.
+		gcSafepointUpperBound := cfState.Info.GetCheckpointTs(cfState.Status) - 1
+		if minCheckpointTs > gcSafepointUpperBound {
+			minCheckpointTs = gcSafepointUpperBound
 		}
 	}
 	m.lastUpdatedTime = time.Now()

diff --git a/cdc/owner/gc_manager_test.go b/cdc/owner/gc_manager_test.go
@@ -117,7 +117,7 @@ func (s *gcManagerSuite) TestUpdateGCSafePoint(c *check.C) {
 	mockPDClient.updateServiceGCSafePointFunc = func(ctx context.Context, serviceID string, ttl int64, safePoint uint64) (uint64, error) {
 		c.Assert(serviceID, check.Equals, cdcServiceSafePointID)
 		c.Assert(ttl, check.Equals, gcManager.gcTTL)
-		c.Assert(safePoint, check.Equals, uint64(20))
+		c.Assert(safePoint, check.Equals, uint64(19))
 		return 0, nil
 	}
 	err = gcManager.updateGCSafePoint(ctx, state)

diff --git a/tests/gc_safepoint/run.sh b/tests/gc_safepoint/run.sh
@@ -22,6 +22,12 @@ function clear_gc_worker_safepoint() {
 	ETCDCTL_API=3 etcdctl --endpoints=$pd_addr del /pd/$pd_cluster_id/gc/safe_point/service/ticdc
 }
 
+function clear_gc_worker_safepoint() {
+	pd_addr=$1
+	pd_cluster_id=$2
+	ETCDCTL_API=3 etcdctl --endpoints=$pd_addr del /pd/$pd_cluster_id/gc/safe_point/service/ticdc
+}
+
 function check_safepoint_cleared() {
 	pd_addr=$1
 	pd_cluster_id=$2