diff --git a/tests/cluster.go b/tests/cluster.go index 893d454990..04d49765a7 100644 --- a/tests/cluster.go +++ b/tests/cluster.go @@ -875,6 +875,17 @@ func (c *TestCluster) GetLeaderServer() *TestServer { // WaitLeader is used to get leader. // If it exceeds the maximum number of loops, it will return an empty string. func (c *TestCluster) WaitLeader(ops ...WaitOption) string { + return c.waitLeaderExcept("", ops...) +} + +// WaitLeaderChange waits until all running servers agree on a leader that is +// different from oldLeader. If it exceeds the maximum number of loops, it will +// return an empty string. +func (c *TestCluster) WaitLeaderChange(oldLeader string, ops ...WaitOption) string { + return c.waitLeaderExcept(oldLeader, ops...) +} + +func (c *TestCluster) waitLeaderExcept(oldLeader string, ops ...WaitOption) string { option := &WaitOp{ retryTimes: WaitLeaderRetryTimes, waitInterval: WaitLeaderCheckInterval, @@ -897,7 +908,7 @@ func (c *TestCluster) WaitLeader(ops ...WaitOption) string { } } for name, num := range counter { - if num == running && c.GetServer(name).IsLeader() { + if name != oldLeader && num == running && c.GetServer(name).IsLeader() { time.Sleep(WaitLeaderReturnDelay) return name } diff --git a/tests/integrations/client/client_test.go b/tests/integrations/client/client_test.go index ca7458e133..54b0216e1c 100644 --- a/tests/integrations/client/client_test.go +++ b/tests/integrations/client/client_test.go @@ -201,10 +201,10 @@ func TestLeaderTransferAndMoveCluster(t *testing.T) { // Transfer leader. for range 3 { oldLeaderName := cluster.WaitLeader() - err := cluster.GetServer(oldLeaderName).ResignLeader() + err := cluster.GetServer(oldLeaderName).ResignLeaderWithRetry() re.NoError(err) - newLeaderName := cluster.WaitLeader() - re.NotEqual(oldLeaderName, newLeaderName) + newLeaderName := cluster.WaitLeaderChange(oldLeaderName) + re.NotEmpty(newLeaderName) } // ABC->ABCDEF @@ -252,11 +252,10 @@ func TestGetTSAfterTransferLeader(t *testing.T) { leaderSwitched.Store(true) return nil }) - err = cluster.GetServer(leader).ResignLeader() + err = cluster.GetServer(leader).ResignLeaderWithRetry() re.NoError(err) - newLeader := cluster.WaitLeader() + newLeader := cluster.WaitLeaderChange(leader) re.NotEmpty(newLeader) - re.NotEqual(leader, newLeader) leader = cluster.WaitLeader() re.NotEmpty(leader) err = cli.GetServiceDiscovery().CheckMemberChanged() @@ -643,8 +642,9 @@ func (suite *followerForwardAndHandleTestSuite) TestGetTsoByFollowerForwarding2( }) lastTS = checkTS(re, cli, lastTS) - re.NoError(suite.cluster.GetLeaderServer().ResignLeader()) - re.NotEmpty(suite.cluster.WaitLeader()) + oldLeaderName := suite.cluster.WaitLeader() + re.NoError(suite.cluster.GetServer(oldLeaderName).ResignLeaderWithRetry()) + re.NotEmpty(suite.cluster.WaitLeaderChange(oldLeaderName)) lastTS = checkTS(re, cli, lastTS) re.NoError(failpoint.Disable("github.com/tikv/pd/client/clients/tso/unreachableNetwork")) @@ -984,9 +984,8 @@ func TestConfigTTLAfterTransferLeader(t *testing.T) { !options.IsLocationReplacementEnabled() }) re.NoError(cluster.GetServer(leaderName).ResignLeaderWithRetry()) - newLeaderName := cluster.WaitLeader() + newLeaderName := cluster.WaitLeaderChange(leaderName) re.NotEmpty(newLeaderName) - re.NotEqual(leaderName, newLeaderName) leader = cluster.GetServer(newLeaderName) re.NotNil(leader) testutil.Eventually(re, func() bool { diff --git a/tests/server/apiv2/handlers/testutil.go b/tests/server/apiv2/handlers/testutil.go index 462b98dd99..a0bf7dcf6a 100644 --- a/tests/server/apiv2/handlers/testutil.go +++ b/tests/server/apiv2/handlers/testutil.go @@ -305,10 +305,20 @@ func MustFinishSplitKeyspaceGroup(re *require.Assertions, server *tests.TestServ if err != nil { return false } - if resp.StatusCode == http.StatusServiceUnavailable || - resp.StatusCode == http.StatusInternalServerError { + if resp.StatusCode == http.StatusServiceUnavailable { return false } + if resp.StatusCode == http.StatusInternalServerError { + // A TSO server can finish the split after this test observes that the + // target group is ready but before this request reaches PD. Treat that + // race as success only after verifying the intended final state. + manager := server.GetServer().GetKeyspaceGroupManager() + if manager == nil { + return false + } + group, err := manager.GetKeyspaceGroupByID(id) + return err == nil && group != nil && !group.IsSplitting() + } re.Equal(http.StatusOK, resp.StatusCode, string(data)) return true }) diff --git a/tests/server/apiv2/handlers/tso_keyspace_group_test.go b/tests/server/apiv2/handlers/tso_keyspace_group_test.go index a79c6d82b3..f89c795964 100644 --- a/tests/server/apiv2/handlers/tso_keyspace_group_test.go +++ b/tests/server/apiv2/handlers/tso_keyspace_group_test.go @@ -197,6 +197,9 @@ func (suite *keyspaceGroupTestSuite) TestSplitKeyspaceGroup() { re.Equal(kg1.Members, kg2.Members) // Finish the split and check the split state. MustFinishSplitKeyspaceGroup(re, suite.server, 2) + // A TSO server may finish the split before the helper's request arrives. + // Calling it again covers that already-finished race. + MustFinishSplitKeyspaceGroup(re, suite.server, 2) kg1 = MustLoadKeyspaceGroupByID(re, suite.server, 1) re.False(kg1.IsSplitting()) kg2 = MustLoadKeyspaceGroupByID(re, suite.server, 2) diff --git a/tests/server/storage/hot_region_storage_test.go b/tests/server/storage/hot_region_storage_test.go index fc6f01000e..154c4d41d5 100644 --- a/tests/server/storage/hot_region_storage_test.go +++ b/tests/server/storage/hot_region_storage_test.go @@ -128,45 +128,48 @@ func (s *hotRegionStorageTestSuite) checkHotRegionStorage(cluster *tests.TestClu err := leaderServer.GetRaftCluster().HandleStoreHeartbeat(&pdpb.StoreHeartbeatRequest{Stats: storeStats}, &pdpb.StoreHeartbeatResponse{}) re.NoError(err) } - var ( - iter storage.HotRegionStorageIterator - next *storage.HistoryHotRegion - err error - ) hotRegionStorage := leaderServer.GetServer().GetHistoryHotRegionStorage() + var writeRegions, readRegions []*storage.HistoryHotRegion testutil.Eventually(re, func() bool { // wait for the history hot region to be written to the storage - iter = hotRegionStorage.NewIterator([]string{utils.Write.String()}, startTime*1000, time.Now().UnixMilli()) - next, err = iter.Next() - return err == nil && next != nil + var err error + writeRegions, err = loadHistoryHotRegions(hotRegionStorage, utils.Write.String(), startTime*1000) + if err != nil || len(writeRegions) != 2 { + return false + } + readRegions, err = loadHistoryHotRegions(hotRegionStorage, utils.Read.String(), startTime*1000) + return err == nil && len(readRegions) == 2 }) - re.Equal(uint64(1), next.RegionID) - re.Equal(uint64(1), next.StoreID) - re.Equal(utils.Write.String(), next.HotRegionType) - next, err = iter.Next() - re.NoError(err) - re.NotNil(next) - re.Equal(uint64(2), next.RegionID) - re.Equal(uint64(2), next.StoreID) - re.Equal(utils.Write.String(), next.HotRegionType) - next, err = iter.Next() - re.NoError(err) - re.Nil(next) - iter = hotRegionStorage.NewIterator([]string{utils.Read.String()}, startTime*1000, time.Now().UnixMilli()) - next, err = iter.Next() - re.NoError(err) - re.NotNil(next) - re.Equal(uint64(3), next.RegionID) - re.Equal(uint64(1), next.StoreID) - re.Equal(utils.Read.String(), next.HotRegionType) - next, err = iter.Next() - re.NoError(err) - re.NotNil(next) - re.Equal(uint64(4), next.RegionID) - re.Equal(uint64(2), next.StoreID) - re.Equal(utils.Read.String(), next.HotRegionType) - next, err = iter.Next() - re.NoError(err) - re.Nil(next) + re.Equal(uint64(1), writeRegions[0].RegionID) + re.Equal(uint64(1), writeRegions[0].StoreID) + re.Equal(utils.Write.String(), writeRegions[0].HotRegionType) + re.Equal(uint64(2), writeRegions[1].RegionID) + re.Equal(uint64(2), writeRegions[1].StoreID) + re.Equal(utils.Write.String(), writeRegions[1].HotRegionType) + re.Equal(uint64(3), readRegions[0].RegionID) + re.Equal(uint64(1), readRegions[0].StoreID) + re.Equal(utils.Read.String(), readRegions[0].HotRegionType) + re.Equal(uint64(4), readRegions[1].RegionID) + re.Equal(uint64(2), readRegions[1].StoreID) + re.Equal(utils.Read.String(), readRegions[1].HotRegionType) +} + +func loadHistoryHotRegions( + hotRegionStorage *storage.HotRegionStorage, + hotRegionType string, + startTime int64, +) ([]*storage.HistoryHotRegion, error) { + iter := hotRegionStorage.NewIterator([]string{hotRegionType}, startTime, time.Now().UnixMilli()) + regions := make([]*storage.HistoryHotRegion, 0) + for { + next, err := iter.Next() + if err != nil { + return nil, err + } + if next == nil { + return regions, nil + } + regions = append(regions, next) + } } func (s *hotRegionStorageTestSuite) TestHotRegionStorageReservedDayConfigChange() {