Make the frequency at which heartbeats update the _vt.vreplication table configurable

go/vt/vttablet/tabletmanager/vreplication/vplayer.go

@@ -58,6 +58,9 @@ type vplayer struct {
 	lastTimestampNs int64
 	// timeOffsetNs keeps track of the clock difference with respect to source tablet.
 	timeOffsetNs int64
+	// numAccumulatedHeartbeats keeps track of how many heartbeats have been received since we updated the time_updated column of _vt.vreplication
+	numAccumulatedHeartbeats int
+
 	// canAcceptStmtEvents is set to true if the current player can accept events in statement mode. Only true for filters that are match all.
 	canAcceptStmtEvents bool
 
@@ -227,6 +230,7 @@ func (vp *vplayer) applyRowEvent(ctx context.Context, rowEvent *binlogdatapb.Row
 }
 
 func (vp *vplayer) updatePos(ts int64) (posReached bool, err error) {
+	vp.numAccumulatedHeartbeats = 0
 	update := binlogplayer.GenerateUpdatePos(vp.vr.id, vp.pos, time.Now().Unix(), ts)
 	if _, err := vp.vr.dbClient.Execute(update); err != nil {
 		return false, fmt.Errorf("error %v updating position", err)
@@ -246,9 +250,7 @@ func (vp *vplayer) updatePos(ts int64) (posReached bool, err error) {
 	return posReached, nil
 }
 
-func (vp *vplayer) recordHeartbeat() (err error) {
-	tm := time.Now().Unix()
-	vp.vr.stats.RecordHeartbeat(tm)
+func (vp *vplayer) updateCurrentTime(tm int64) error {
 	update, err := binlogplayer.GenerateUpdateTime(vp.vr.id, tm)
 	if err != nil {
 		return err
@@ -259,6 +261,20 @@ func (vp *vplayer) recordHeartbeat() (err error) {
 	return nil
 }
 
+func (vp *vplayer) mustUpdateCurrentTime() bool {
+	return vp.numAccumulatedHeartbeats >= *vreplicationHeartbeatUpdateInterval
+}
+
+func (vp *vplayer) recordHeartbeat() error {
+	tm := time.Now().Unix()
+	vp.vr.stats.RecordHeartbeat(tm)
+	if !vp.mustUpdateCurrentTime() {
+		return nil
+	}
+	vp.numAccumulatedHeartbeats = 0
+	return vp.updateCurrentTime(tm)
+}
+
 // applyEvents is the main thread that applies the events. It has the following use
 // cases to take into account:
 // * Normal transaction that has row mutations. In this case, the transaction
@@ -606,11 +622,13 @@ func (vp *vplayer) applyEvent(ctx context.Context, event *binlogdatapb.VEvent, m
 		return io.EOF
 	case binlogdatapb.VEventType_HEARTBEAT:
 		if !vp.vr.dbClient.InTransaction {
+			vp.numAccumulatedHeartbeats++
 			err := vp.recordHeartbeat()
 			if err != nil {
 				return err
 			}
 		}
 	}
+
 	return nil
 }

go/vt/vttablet/tabletmanager/vreplication/vplayer_flaky_test.go

@@ -38,6 +38,34 @@ import (
 	binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
 )
 
+func TestHeartbeatFrequencyFlag(t *testing.T) {
+	origVReplicationHeartbeatUpdateInterval := *vreplicationHeartbeatUpdateInterval
+	defer func() {
+		*vreplicationHeartbeatUpdateInterval = origVReplicationHeartbeatUpdateInterval
+	}()
+
+	stats := binlogplayer.NewStats()
+	vp := &vplayer{vr: &vreplicator{dbClient: newVDBClient(realDBClientFactory(), stats), stats: stats}}
+
+	t.Run("default frequency", func(t *testing.T) {
+		vp.numAccumulatedHeartbeats = 0
+		require.False(t, vp.mustUpdateCurrentTime())
+		vp.numAccumulatedHeartbeats = 1
+		require.True(t, vp.mustUpdateCurrentTime())
+	})
+
+	*vreplicationHeartbeatUpdateInterval = 4
+
+	t.Run("custom frequency", func(t *testing.T) {
+		vp.numAccumulatedHeartbeats = 0
+		require.False(t, vp.mustUpdateCurrentTime())
+		vp.numAccumulatedHeartbeats = 3
+		require.False(t, vp.mustUpdateCurrentTime())
+		vp.numAccumulatedHeartbeats = 4
+		require.True(t, vp.mustUpdateCurrentTime())
+	})
+}
+
 func TestVReplicationTimeUpdated(t *testing.T) {
 	ctx := context.Background()
 	defer deleteTablet(addTablet(100))

go/vt/vttablet/tabletmanager/vreplication/vreplicator.go

@@ -41,12 +41,20 @@ var (
 	// idleTimeout is set to slightly above 1s, compared to heartbeatTime
 	// set by VStreamer at slightly below 1s. This minimizes conflicts
 	// between the two timeouts.
-	idleTimeout         = 1100 * time.Millisecond
+	idleTimeout = 1100 * time.Millisecond
+
 	dbLockRetryDelay    = 1 * time.Second
 	relayLogMaxSize     = flag.Int("relay_log_max_size", 250000, "Maximum buffer size (in bytes) for VReplication target buffering. If single rows are larger than this, a single row is buffered at a time.")
 	relayLogMaxItems    = flag.Int("relay_log_max_items", 5000, "Maximum number of rows for VReplication target buffering.")
 	copyTimeout         = 1 * time.Hour
 	replicaLagTolerance = 10 * time.Second
+
+	// vreplicationHeartbeatUpdateInterval determines how often the time_updated column is updated if there are no real events on the source and the source
+	// vstream is only sending heartbeats for this long. Keep this low if you expect high QPS and are monitoring this column to alert about potential
+	// outages. Keep this high if
+	// 		you have too many streams the extra write qps or cpu load due to these updates are unacceptable
+	//		you have too many streams and/or a large source field (lot of participating tables) which generates unacceptable increase in your binlog size
+	vreplicationHeartbeatUpdateInterval = flag.Int("vreplication_heartbeat_update_interval", 1, "Frequency (in seconds) at which the time_updated column of a vreplication stream when idling")
 )
 
 // vreplicator provides the core logic to start vreplication streams