CherryPick(#15962): Improve WaitForPos errors, don't include Result s…

…truct in message (#5239) * cherry pick of 15962 * Fix conflicts Signed-off-by: Matt Lord <mattalord@gmail.com> --------- Signed-off-by: Matt Lord <mattalord@gmail.com> Co-authored-by: Matt Lord <mattalord@gmail.com>
vitessio · May 23, 2024 · 815495a · 815495a
1 parent dcd1a3c
commit 815495a
Show file tree

Hide file tree

Showing 3 changed files with 18 additions and 4 deletions.
diff --git a/go/vt/vttablet/endtoend/vstreamer_test.go b/go/vt/vttablet/endtoend/vstreamer_test.go
@@ -22,6 +22,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -65,6 +66,7 @@ func TestSchemaVersioning(t *testing.T) {
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
+	wg := sync.WaitGroup{}
 	tsv.EnableHistorian(true)
 	tsv.SetTracking(true)
 	time.Sleep(100 * time.Millisecond) // wait for _vt tables to be created
@@ -155,7 +157,9 @@ func TestSchemaVersioning(t *testing.T) {
 		}
 		return nil
 	}
+	wg.Add(1)
 	go func() {
+		defer wg.Done()
 		defer close(eventCh)
 		req := &binlogdatapb.VStreamRequest{Target: target, Position: "current", TableLastPKs: nil, Filter: filter}
 		if err := tsv.VStream(ctx, req, send); err != nil {
@@ -186,6 +190,7 @@ func TestSchemaVersioning(t *testing.T) {
 	}
 	runCases(ctx, t, cases, eventCh)
 	cancel()
+	wg.Wait()
 
 	log.Infof("\n\n\n=============================================== PAST EVENTS WITH TRACK VERSIONS START HERE ======================\n\n\n")
 	ctx, cancel = context.WithCancel(context.Background())
@@ -214,7 +219,9 @@ func TestSchemaVersioning(t *testing.T) {
 		}
 		return nil
 	}
+	wg.Add(1)
 	go func() {
+		defer wg.Done()
 		defer close(eventCh)
 		req := &binlogdatapb.VStreamRequest{Target: target, Position: startPos, TableLastPKs: nil, Filter: filter}
 		if err := tsv.VStream(ctx, req, send); err != nil {
@@ -257,6 +264,7 @@ func TestSchemaVersioning(t *testing.T) {
 	expectLogs(ctx, t, "Past stream", eventCh, output)
 
 	cancel()
+	wg.Wait()
 
 	log.Infof("\n\n\n=============================================== PAST EVENTS WITHOUT TRACK VERSIONS START HERE ======================\n\n\n")
 	tsv.EnableHistorian(false)
@@ -286,7 +294,9 @@ func TestSchemaVersioning(t *testing.T) {
 		}
 		return nil
 	}
+	wg.Add(1)
 	go func() {
+		defer wg.Done()
 		defer close(eventCh)
 		req := &binlogdatapb.VStreamRequest{Target: target, Position: startPos, TableLastPKs: nil, Filter: filter}
 		if err := tsv.VStream(ctx, req, send); err != nil {
@@ -331,6 +341,7 @@ func TestSchemaVersioning(t *testing.T) {
 
 	expectLogs(ctx, t, "Past stream", eventCh, output)
 	cancel()
+	wg.Wait()
 
 	client := framework.NewClient()
 	client.Execute("drop table vitess_version", nil)

diff --git a/go/vt/vttablet/tabletmanager/vreplication/engine.go b/go/vt/vttablet/tabletmanager/vreplication/engine.go
@@ -790,8 +790,10 @@ func (vre *Engine) WaitForPos(ctx context.Context, id int32, pos string) error {
 			}
 		case len(qr.Rows) == 0:
 			return fmt.Errorf("vreplication stream %d not found", id)
-		case len(qr.Rows) > 1 || len(qr.Rows[0]) != 3:
-			return fmt.Errorf("unexpected result: %v", qr)
+		case len(qr.Rows) > 1:
+			return fmt.Errorf("vreplication stream received more rows than expected, got %d instead of 1", len(qr.Rows))
+		case len(qr.Rows[0]) != 3:
+			return fmt.Errorf("vreplication stream received an unexpected number of columns, got %d instead of 3", len(qr.Rows[0]))
 		}
 
 		// When err is not nil then we got a retryable error and will loop again.

diff --git a/go/vt/vttablet/tabletmanager/vreplication/engine_test.go b/go/vt/vttablet/tabletmanager/vreplication/engine_test.go
@@ -426,7 +426,8 @@ func TestWaitForPosError(t *testing.T) {
 
 	dbClient.ExpectRequest("select pos, state, message from _vt.vreplication where id=1", &sqltypes.Result{Rows: [][]sqltypes.Value{{}}}, nil)
 	err = vre.WaitForPos(context.Background(), 1, "MariaDB/0-1-1084")
-	want = "unexpected result: &{[] 0 0 map[] 0 [[]]  0 }"
+	want = "vreplication stream received an unexpected number of columns, got 0 instead of 3"
+
 	assert.EqualError(t, err, want, "WaitForPos:")
 
 	dbClient.ExpectRequest("select pos, state, message from _vt.vreplication where id=1", &sqltypes.Result{Rows: [][]sqltypes.Value{{
@@ -435,7 +436,7 @@ func TestWaitForPosError(t *testing.T) {
 		sqltypes.NewVarBinary("MariaDB/0-1-1083"),
 	}}}, nil)
 	err = vre.WaitForPos(context.Background(), 1, "MariaDB/0-1-1084")
-	want = `unexpected result: &{[] 0 0 map[] 0 [[VARBINARY("MariaDB/0-1-1083")] [VARBINARY("MariaDB/0-1-1083")]]  0 }`
+	want = "vreplication stream received more rows than expected, got 2 instead of 1"
 	assert.EqualError(t, err, want, "WaitForPos:")
 }