Cherry-pick 19f9c95 with conflicts

go/mysql/server_test.go

@@ -1424,7 +1424,7 @@ func TestListenerShutdown(t *testing.T) {
 
 	l.Shutdown()
 
-	assert.EqualValues(t, 1, connRefuse.Get(), "connRefuse")
+	waitForConnRefuse(t, 1)
 
 	err = conn.Ping()
 	require.EqualError(t, err, "Server shutdown in progress (errno 1053) (sqlstate 08S01)")
@@ -1436,6 +1436,24 @@ func TestListenerShutdown(t *testing.T) {
 	require.Equal(t, "Server shutdown in progress", sqlErr.Message)
 }
 
+func waitForConnRefuse(t *testing.T, valWanted int64) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	tick := time.NewTicker(100 * time.Millisecond)
+	defer tick.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			require.FailNow(t, "connRefuse did not reach %v", valWanted)
+		case <-tick.C:
+			if connRefuse.Get() == valWanted {
+				return
+			}
+		}
+	}
+}
+
 func TestParseConnAttrs(t *testing.T) {
 	expected := map[string]string{
 		"_client_version": "8.0.11",

go/vt/mysqlctl/fakemysqldaemon.go

@@ -286,13 +286,6 @@ func (fmd *FakeMysqlDaemon) GetServerUUID(ctx context.Context) (string, error) {
 	return "000000", nil
 }
 
-// CurrentPrimaryPositionLocked is thread-safe.
-func (fmd *FakeMysqlDaemon) CurrentPrimaryPositionLocked(pos replication.Position) {
-	fmd.mu.Lock()
-	defer fmd.mu.Unlock()
-	fmd.CurrentPrimaryPosition = pos
-}
-
 // ReplicationStatus is part of the MysqlDaemon interface.
 func (fmd *FakeMysqlDaemon) ReplicationStatus() (replication.ReplicationStatus, error) {
 	if fmd.ReplicationStatusError != nil {
@@ -316,6 +309,8 @@ func (fmd *FakeMysqlDaemon) ReplicationStatus() (replication.ReplicationStatus,
 
 // PrimaryStatus is part of the MysqlDaemon interface.
 func (fmd *FakeMysqlDaemon) PrimaryStatus(ctx context.Context) (replication.PrimaryStatus, error) {
+	fmd.mu.Lock()
+	defer fmd.mu.Unlock()
 	if fmd.PrimaryStatusError != nil {
 		return replication.PrimaryStatus{}, fmd.PrimaryStatusError
 	}
@@ -380,8 +375,27 @@ func (fmd *FakeMysqlDaemon) GetPreviousGTIDs(ctx context.Context, binlog string)
 }
 
 // PrimaryPosition is part of the MysqlDaemon interface.
+<<<<<<< HEAD
 func (fmd *FakeMysqlDaemon) PrimaryPosition() (replication.Position, error) {
 	return fmd.CurrentPrimaryPosition, nil
+=======
+func (fmd *FakeMysqlDaemon) PrimaryPosition(ctx context.Context) (replication.Position, error) {
+	return fmd.GetPrimaryPositionLocked(), nil
+}
+
+// GetPrimaryPositionLocked gets the primary position while holding the lock.
+func (fmd *FakeMysqlDaemon) GetPrimaryPositionLocked() replication.Position {
+	fmd.mu.Lock()
+	defer fmd.mu.Unlock()
+	return fmd.CurrentPrimaryPosition
+}
+
+// SetPrimaryPositionLocked is thread-safe.
+func (fmd *FakeMysqlDaemon) SetPrimaryPositionLocked(pos replication.Position) {
+	fmd.mu.Lock()
+	defer fmd.mu.Unlock()
+	fmd.CurrentPrimaryPosition = pos
+>>>>>>> 19f9c95632 (Flaky test fixes (#16940))
 }
 
 // IsReadOnly is part of the MysqlDaemon interface.

go/vt/vtctl/grpcvtctldserver/server_test.go

@@ -12620,7 +12620,7 @@ func TestTabletExternallyReparented(t *testing.T) {
 				defer func() {
 					topofactory.SetError(nil)
 
-					ctx, cancel := context.WithTimeout(ctx, time.Millisecond*10)
+					ctx, cancel := context.WithTimeout(ctx, time.Second*10)
 					defer cancel()
 
 					resp, err := vtctld.GetTablets(ctx, &vtctldatapb.GetTabletsRequest{})

go/vt/vttablet/tabletmanager/framework_test.go

@@ -111,8 +111,8 @@ func newTestEnv(t *testing.T, ctx context.Context, sourceKeyspace string, source
 	tmclienttest.SetProtocol(fmt.Sprintf("go.vt.vttablet.tabletmanager.framework_test_%s", t.Name()), tenv.protoName)
 
 	tenv.mysqld = mysqlctl.NewFakeMysqlDaemon(fakesqldb.New(t))
-	var err error
-	tenv.mysqld.CurrentPrimaryPosition, err = replication.ParsePosition(gtidFlavor, gtidPosition)
+	curPosition, err := replication.ParsePosition(gtidFlavor, gtidPosition)
+	tenv.mysqld.SetPrimaryPositionLocked(curPosition)
 	require.NoError(t, err)
 
 	return tenv