VReplication: Improve handling of vplayer stalls

go/vt/vttablet/tabletmanager/vreplication/framework_test.go

@@ -32,8 +32,6 @@ import (
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/proto"
 
-	"vitess.io/vitess/go/vt/sqlparser"
-
 	_flag "vitess.io/vitess/go/internal/flag"
 	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/mysql/replication"
@@ -46,6 +44,7 @@ import (
 	"vitess.io/vitess/go/vt/log"
 	"vitess.io/vitess/go/vt/servenv"
 	"vitess.io/vitess/go/vt/sidecardb"
+	"vitess.io/vitess/go/vt/sqlparser"
 	"vitess.io/vitess/go/vt/topo"
 	"vitess.io/vitess/go/vt/vttablet"
 	"vitess.io/vitess/go/vt/vttablet/queryservice"
@@ -74,6 +73,7 @@ var (
 	testForeignKeyQueries    = false
 	testSetForeignKeyQueries = false
 	doNotLogDBQueries        = false
+	recvTimeout              = 5 * time.Second
 )
 
 type LogExpectation struct {
@@ -492,14 +492,14 @@ func (dbc *realDBClient) ExecuteFetch(query string, maxrows int) (*sqltypes.Resu
 	return qr, err
 }
 
-func (dc *realDBClient) ExecuteFetchMulti(query string, maxrows int) ([]*sqltypes.Result, error) {
+func (dbc *realDBClient) ExecuteFetchMulti(query string, maxrows int) ([]*sqltypes.Result, error) {
 	queries, err := sqlparser.NewTestParser().SplitStatementToPieces(query)
 	if err != nil {
 		return nil, err
 	}
 	results := make([]*sqltypes.Result, 0, len(queries))
 	for _, query := range queries {
-		qr, err := dc.ExecuteFetch(query, maxrows)
+		qr, err := dbc.ExecuteFetch(query, maxrows)
 		if err != nil {
 			return nil, err
 		}
@@ -518,7 +518,7 @@ func expectDeleteQueries(t *testing.T) {
 		"/delete from _vt.vreplication",
 		"/delete from _vt.copy_state",
 		"/delete from _vt.post_copy_action",
-	))
+	), recvTimeout)
 }
 
 func deleteAllVReplicationStreams(t *testing.T) {
@@ -635,7 +635,7 @@ func expectDBClientQueries(t *testing.T, expectations qh.ExpectationSequence, sk
 				))
 			}
 		case <-time.After(5 * time.Second):
-			t.Fatalf("no query received")
+			require.FailNow(t, "no query received")
 			failed = true
 		}
 	}
@@ -656,7 +656,7 @@ func expectDBClientQueries(t *testing.T, expectations qh.ExpectationSequence, sk
 
 // expectNontxQueries disregards transactional statements like begin and commit.
 // It also disregards updates to _vt.vreplication.
-func expectNontxQueries(t *testing.T, expectations qh.ExpectationSequence) {
+func expectNontxQueries(t *testing.T, expectations qh.ExpectationSequence, recvTimeout time.Duration) {
 	t.Helper()
 	if doNotLogDBQueries {
 		return
@@ -684,8 +684,8 @@ func expectNontxQueries(t *testing.T, expectations qh.ExpectationSequence) {
 				"query:%q\nmessage:%s\nexpectation:%s\nmatched:%t\nerror:%v\nhistory:%s",
 				got, result.Message, result.Expectation, result.Matched, result.Error, validator.History(),
 			))
-		case <-time.After(5 * time.Second):
-			t.Fatalf("no query received")
+		case <-time.After(recvTimeout):
+			require.FailNow(t, "no query received")
 			failed = true
 		}
 	}

go/vt/vttablet/tabletmanager/vreplication/vcopier_test.go

@@ -182,7 +182,7 @@ func testPlayerCopyCharPK(t *testing.T) {
 		`/insert into _vt.copy_state \(lastpk, vrepl_id, table_name\) values \('fields:{name:\\"idc\\" type:BINARY charset:63 flags:20611} rows:{lengths:2 values:\\"c\\\\x00\\"}'.*`,
 		"/delete cs, pca from _vt.copy_state as cs left join _vt.post_copy_action as pca on cs.vrepl_id=pca.vrepl_id and cs.table_name=pca.table_name.*dst",
 		"/update _vt.vreplication set state='Running",
-	))
+	), recvTimeout)
 
 	expectData(t, "dst", [][]string{
 		{"a\000", "3"},
@@ -304,7 +304,7 @@ func testPlayerCopyVarcharPKCaseInsensitive(t *testing.T) {
 	}).Then(qh.Immediately(
 		"/delete cs, pca from _vt.copy_state as cs left join _vt.post_copy_action as pca on cs.vrepl_id=pca.vrepl_id and cs.table_name=pca.table_name.*dst",
 		"/update _vt.vreplication set state='Running'",
-	)))
+	)), recvTimeout)
 
 	expectData(t, "dst", [][]string{
 		{"a", "1"},
@@ -415,7 +415,7 @@ func testPlayerCopyVarcharCompositePKCaseSensitiveCollation(t *testing.T) {
 		// Wrap-up.
 		"/delete cs, pca from _vt.copy_state as cs left join _vt.post_copy_action as pca on cs.vrepl_id=pca.vrepl_id and cs.table_name=pca.table_name.*dst",
 		"/update _vt.vreplication set state='Running'",
-	))
+	), recvTimeout)
 
 	expectData(t, "dst", [][]string{
 		{"1", "B", "B", "3"},
@@ -790,7 +790,7 @@ func testPlayerCopyBigTable(t *testing.T) {
 		// Copy is done. Go into running state.
 		// All tables copied. Final catch up followed by Running state.
 		"/update _vt.vreplication set state='Running'",
-	)))
+	)), recvTimeout)
 
 	expectData(t, "dst", [][]string{
 		{"1", "aaa"},
@@ -918,7 +918,7 @@ func testPlayerCopyWildcardRule(t *testing.T) {
 		"/delete cs, pca from _vt.copy_state as cs left join _vt.post_copy_action as pca on cs.vrepl_id=pca.vrepl_id and cs.table_name=pca.table_name.*src",
 		// Copy is done. Go into running state.
 		"/update _vt.vreplication set state='Running'",
-	)))
+	)), recvTimeout)
 
 	expectData(t, "src", [][]string{
 		{"1", "aaa"},
@@ -1078,7 +1078,7 @@ func testPlayerCopyTableContinuation(t *testing.T) {
 	)).Then(qh.Immediately(
 		"/delete cs, pca from _vt.copy_state as cs left join _vt.post_copy_action as pca on cs.vrepl_id=pca.vrepl_id and cs.table_name=pca.table_name.*not_copied",
 		"/update _vt.vreplication set state='Running'",
-	)))
+	)), recvTimeout)
 
 	expectData(t, "dst1", [][]string{
 		{"1", "insert in"},
@@ -1188,7 +1188,7 @@ func testPlayerCopyWildcardTableContinuation(t *testing.T) {
 		`/insert into _vt.copy_state .*`,
 		"/delete cs, pca from _vt.copy_state as cs left join _vt.post_copy_action as pca on cs.vrepl_id=pca.vrepl_id and cs.table_name=pca.table_name.*dst",
 		"/update _vt.vreplication set state='Running'",
-	)))
+	)), recvTimeout)
 
 	expectData(t, "dst", [][]string{
 		{"2", "copied"},
@@ -1279,7 +1279,7 @@ func TestPlayerCopyWildcardTableContinuationWithOptimizeInserts(t *testing.T) {
 		`/insert into _vt.copy_state .*`,
 		"/delete cs, pca from _vt.copy_state as cs left join _vt.post_copy_action as pca on cs.vrepl_id=pca.vrepl_id and cs.table_name=pca.table_name.*dst",
 		"/update _vt.vreplication set state='Running'",
-	))
+	), recvTimeout)
 	expectData(t, "dst", [][]string{
 		{"2", "copied"},
 		{"3", "uncopied"},
@@ -1659,7 +1659,7 @@ func testPlayerCopyTablesWithGeneratedColumn(t *testing.T) {
 		// copy of dst2 is done: delete from copy_state.
 		"/delete cs, pca from _vt.copy_state as cs left join _vt.post_copy_action as pca on cs.vrepl_id=pca.vrepl_id and cs.table_name=pca.table_name.*dst2",
 		"/update _vt.vreplication set state",
-	))
+	), recvTimeout)
 
 	expectData(t, "dst1", [][]string{
 		{"1", "aaa", "1aaa", "aaa1", "10"},
@@ -1826,7 +1826,7 @@ func testCopyInvisibleColumns(t *testing.T) {
 		// copy of dst1 is done: delete from copy_state.
 		"/delete cs, pca from _vt.copy_state as cs left join _vt.post_copy_action as pca on cs.vrepl_id=pca.vrepl_id and cs.table_name=pca.table_name.*dst1",
 		"/update _vt.vreplication set state='Running'",
-	))
+	), recvTimeout)
 
 	expectData(t, "dst1", [][]string{
 		{"1", "10"},

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

@@ -30,12 +30,18 @@ import (
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/vt/binlog/binlogplayer"
 	"vitess.io/vitess/go/vt/log"
+	"vitess.io/vitess/go/vt/vterrors"
 	"vitess.io/vitess/go/vt/vttablet"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/throttle/throttlerapp"
 
 	binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
 )
 
+var (
+	// The error to return when we have detected a stall in the vplayer.
+	ErrVPlayerStalled = fmt.Errorf("progress stalled; vplayer was unable to replicate the transaction in a timely manner; examine the target mysqld instance health and the replicated queries' EXPLAIN output to see why queries are taking unusually long")
+)
+
 // vplayer replays binlog events by pulling them from a vstreamer.
 type vplayer struct {
 	vr        *vreplicator
@@ -148,7 +154,7 @@ func newVPlayer(vr *vreplicator, settings binlogplayer.VRSettings, copyState map
 			return nil, vr.dbClient.AddQueryToTrxBatch(sql) // Should become part of the trx batch
 		}
 		commitFunc = func() error {
-			return vr.dbClient.CommitTrxQueryBatch() // Commit the current trx batch
+			return vr.dbClient.CommitTrxQueryBatch()
 		}
 		vr.dbClient.maxBatchSize = maxAllowedPacket
 	}
@@ -367,12 +373,13 @@ func (vp *vplayer) applyRowEvent(ctx context.Context, rowEvent *binlogdatapb.Row
 	return nil
 }
 
+// updatePos should get called at a minimum of vreplicationMinimumHeartbeatUpdateInterval.
 func (vp *vplayer) updatePos(ctx context.Context, ts int64) (posReached bool, err error) {
-	vp.numAccumulatedHeartbeats = 0
 	update := binlogplayer.GenerateUpdatePos(vp.vr.id, vp.pos, time.Now().Unix(), ts, vp.vr.stats.CopyRowCount.Get(), vreplicationStoreCompressedGTID)
 	if _, err := vp.query(ctx, update); err != nil {
 		return false, fmt.Errorf("error %v updating position", err)
 	}
+	vp.numAccumulatedHeartbeats = 0
 	vp.unsavedEvent = nil
 	vp.timeLastSaved = time.Now()
 	vp.vr.stats.SetLastPosition(vp.pos)
@@ -399,8 +406,16 @@ func (vp *vplayer) recordHeartbeat() error {
 	if !vp.mustUpdateHeartbeat() {
 		return nil
 	}
+	if err := vp.vr.updateHeartbeatTime(tm); err != nil {
+		return vterrors.Wrapf(ErrVPlayerStalled, "failed to record heartbeat: %v", err)
+	}
+	// Only reset the pending heartbeat count if the update was successful.
+	// Otherwise the vplayer may not actually be making progress and nobody
+	// is aware of it -- resulting in the com_binlog_dump connection on the
+	// source that is managed by the binlog_player getting closed by mysqld
+	// when the source_net_timeout is hit.
 	vp.numAccumulatedHeartbeats = 0
-	return vp.vr.updateHeartbeatTime(tm)
+	return nil
 }
 
 // applyEvents is the main thread that applies the events. It has the following use
@@ -438,7 +453,7 @@ func (vp *vplayer) recordHeartbeat() error {
 // current position to be saved.
 //
 // In order to handle the above use cases, we use an implicit transaction scheme:
-// A BEGIN does not really start a transaction. Ony a ROW event does. With this
+// A BEGIN does not really start a transaction. Only a ROW event does. With this
 // approach, no transaction gets started if an empty one arrives. If a we receive
 // a commit, and a we are not in a transaction, we infer that the transaction was
 // empty, and remember it as an unsaved event. The next GTID event will reset the
@@ -497,6 +512,7 @@ func (vp *vplayer) applyEvents(ctx context.Context, relay *relayLog) error {
 				return nil
 			}
 		}
+
 		for i, events := range items {
 			for j, event := range events {
 				if event.Timestamp != 0 {
@@ -526,7 +542,17 @@ func (vp *vplayer) applyEvents(ctx context.Context, relay *relayLog) error {
 				if err := vp.applyEvent(ctx, event, mustSave); err != nil {
 					if err != io.EOF {
 						vp.vr.stats.ErrorCounts.Add([]string{"Apply"}, 1)
-						log.Errorf("Error applying event: %s", err.Error())
+						var table, tableLogMsg string
+						switch {
+						case event.GetFieldEvent() != nil:
+							table = event.GetFieldEvent().TableName
+						case event.GetRowEvent() != nil:
+							table = event.GetRowEvent().TableName
+						}
+						if table != "" {
+							tableLogMsg = fmt.Sprintf(" for table %s", table)
+						}
+						log.Errorf("Error applying event%s: %s", tableLogMsg, err.Error())
 					}
 					return err
 				}
@@ -635,7 +661,8 @@ func (vp *vplayer) applyEvent(ctx context.Context, event *binlogdatapb.VEvent, m
 			log.Infof("Error applying row event: %s", err.Error())
 			return err
 		}
-		//Row event is logged AFTER RowChanges are applied so as to calculate the total elapsed time for the Row event
+		// Row event is logged AFTER RowChanges are applied so as to calculate the total elapsed
+		// time for the Row event.
 		stats.Send(fmt.Sprintf("%v", event.RowEvent))
 	case binlogdatapb.VEventType_OTHER:
 		if vp.vr.dbClient.InTransaction {