CI: Address data race in TestSchemaVersioning (#15998)

Signed-off-by: Matt Lord <[email protected]>
vitessio · May 23, 2024 · 7c6d5e5 · 7c6d5e5
1 parent afbce6a
commit 7c6d5e5
Showing 1 changed file with 11 additions and 0 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)