Merge branch 'release-7.1' into cherry-pick-6983-to-release-7.1

pkg/core/region.go

@@ -625,6 +625,11 @@ func (r *RegionInfo) GetReplicationStatus() *replication_modepb.RegionReplicatio
 	return r.replicationStatus
 }
 
+// IsFlashbackChanged returns true if flashback changes.
+func (r *RegionInfo) IsFlashbackChanged(l *RegionInfo) bool {
+	return r.meta.FlashbackStartTs != l.meta.FlashbackStartTs || r.meta.IsInFlashback != l.meta.IsInFlashback
+}
+
 // IsFromHeartbeat returns whether the region info is from the region heartbeat.
 func (r *RegionInfo) IsFromHeartbeat() bool {
 	return r.fromHeartbeat
@@ -751,6 +756,14 @@ func GenerateRegionGuideFunc(enableLog bool) RegionGuideFunc {
 				(region.GetReplicationStatus().GetState() != origin.GetReplicationStatus().GetState() ||
 					region.GetReplicationStatus().GetStateId() != origin.GetReplicationStatus().GetStateId()) {
 				saveCache = true
+				return
+			}
+			// Do not save to kv, because 1) flashback will be eventually set to
+			// false, 2) flashback changes almost all regions in a cluster.
+			// Saving kv may downgrade PD performance when there are many regions.
+			if region.IsFlashbackChanged(origin) {
+				saveCache = true
+				return
 			}
 		}
 		return

pkg/core/region_option.go

@@ -183,6 +183,14 @@ func WithDecConfVer() RegionCreateOption {
 	}
 }
 
+// WithFlashback set region flashback states.
+func WithFlashback(isInFlashback bool, flashbackTS uint64) RegionCreateOption {
+	return func(region *RegionInfo) {
+		region.meta.FlashbackStartTs = flashbackTS
+		region.meta.IsInFlashback = isInFlashback
+	}
+}
+
 // SetCPUUsage sets the CPU usage of the region.
 func SetCPUUsage(v uint64) RegionCreateOption {
 	return func(region *RegionInfo) {

server/cluster/cluster.go

@@ -1025,7 +1025,7 @@ func (c *RaftCluster) processRegionHeartbeat(region *core.RegionInfo) error {
 	c.coordinator.CheckTransferWitnessLeader(region)
 
 	hasRegionStats := c.regionStats != nil
-	// Save to storage if meta is updated.
+	// Save to storage if meta is updated, except for flashback.
 	// Save to cache if meta or leader is updated, or contains any down/pending peer.
 	// Mark isNew if the region in cache does not have leader.
 	isNew, saveKV, saveCache, needSync := regionGuide(region, origin)

server/cluster/cluster_test.go

@@ -841,6 +841,16 @@ func TestRegionHeartbeat(t *testing.T) {
 		regions[i] = region
 		re.NoError(cluster.processRegionHeartbeat(region))
 		checkRegions(re, cluster.core, regions[:i+1])
+
+		// Flashback
+		region = region.Clone(core.WithFlashback(true, 1))
+		regions[i] = region
+		re.NoError(cluster.processRegionHeartbeat(region))
+		checkRegions(re, cluster.core, regions[:i+1])
+		region = region.Clone(core.WithFlashback(false, 0))
+		regions[i] = region
+		re.NoError(cluster.processRegionHeartbeat(region))
+		checkRegions(re, cluster.core, regions[:i+1])
 	}
 
 	regionCounts := make(map[uint64]int)