diff --git a/client/http/client.go b/client/http/client.go index 437dd7fee371..ae0f7fee347c 100644 --- a/client/http/client.go +++ b/client/http/client.go @@ -437,7 +437,7 @@ func (c *client) SetPlacementRule(ctx context.Context, rule *Rule) error { http.MethodPost, bytes.NewBuffer(ruleJSON), nil) } -// SetPlacementRules sets the placement rules in batch. +// SetPlacementRuleInBatch sets the placement rules in batch. func (c *client) SetPlacementRuleInBatch(ctx context.Context, ruleOps []*RuleOp) error { ruleOpsJSON, err := json.Marshal(ruleOps) if err != nil { diff --git a/pkg/utils/tsoutil/tsoutil.go b/pkg/utils/tsoutil/tsoutil.go index 796012ae0311..43d8b09aa494 100644 --- a/pkg/utils/tsoutil/tsoutil.go +++ b/pkg/utils/tsoutil/tsoutil.go @@ -25,6 +25,11 @@ const ( logicalBits = (1 << physicalShiftBits) - 1 ) +// TimeToTS converts a `time.Time` to an `uint64` TS. +func TimeToTS(t time.Time) uint64 { + return ComposeTS(t.UnixNano()/int64(time.Millisecond), 0) +} + // ParseTS parses the ts to (physical,logical). func ParseTS(ts uint64) (time.Time, uint64) { physical, logical := ParseTSUint64(ts) diff --git a/server/cluster/cluster.go b/server/cluster/cluster.go index 1c3d8a03a980..ec8ca3a0d65f 100644 --- a/server/cluster/cluster.go +++ b/server/cluster/cluster.go @@ -2241,7 +2241,9 @@ func (c *RaftCluster) SetMinResolvedTS(storeID, minResolvedTS uint64) error { return nil } -func (c *RaftCluster) checkAndUpdateMinResolvedTS() (uint64, bool) { +// CheckAndUpdateMinResolvedTS checks and updates the min resolved ts of the cluster. +// This is exported for testing purpose. +func (c *RaftCluster) CheckAndUpdateMinResolvedTS() (uint64, bool) { c.Lock() defer c.Unlock() @@ -2284,7 +2286,7 @@ func (c *RaftCluster) runMinResolvedTSJob() { case <-ticker.C: interval = c.opt.GetMinResolvedTSPersistenceInterval() if interval != 0 { - if current, needPersist := c.checkAndUpdateMinResolvedTS(); needPersist { + if current, needPersist := c.CheckAndUpdateMinResolvedTS(); needPersist { c.storage.SaveMinResolvedTS(current) } } else { diff --git a/tests/integrations/client/http_client_test.go b/tests/integrations/client/http_client_test.go index 2bdd1d5b5ee0..85e3e5a4f881 100644 --- a/tests/integrations/client/http_client_test.go +++ b/tests/integrations/client/http_client_test.go @@ -20,6 +20,7 @@ import ( "net/http" "sort" "testing" + "time" "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" @@ -28,6 +29,7 @@ import ( "github.com/tikv/pd/pkg/schedule/labeler" "github.com/tikv/pd/pkg/schedule/placement" "github.com/tikv/pd/pkg/utils/testutil" + "github.com/tikv/pd/pkg/utils/tsoutil" "github.com/tikv/pd/tests" ) @@ -73,27 +75,30 @@ func (suite *httpClientTestSuite) TearDownSuite() { func (suite *httpClientTestSuite) TestGetMinResolvedTSByStoresIDs() { re := suite.Require() - var ( - minResolvedTS uint64 - storeMinResolvedTSMap map[uint64]uint64 - err error - ) - // Wait for the cluster-level min resolved TS to be initialized. + testMinResolvedTS := tsoutil.TimeToTS(time.Now()) + raftCluster := suite.cluster.GetLeaderServer().GetRaftCluster() + err := raftCluster.SetMinResolvedTS(1, testMinResolvedTS) + re.NoError(err) + // Make sure the min resolved TS is updated. testutil.Eventually(re, func() bool { - minResolvedTS, storeMinResolvedTSMap, err = suite.client.GetMinResolvedTSByStoresIDs(suite.ctx, nil) - re.NoError(err) - return minResolvedTS > 0 && len(storeMinResolvedTSMap) == 0 + minResolvedTS, _ := raftCluster.CheckAndUpdateMinResolvedTS() + return minResolvedTS == testMinResolvedTS }) + // Wait for the cluster-level min resolved TS to be initialized. + minResolvedTS, storeMinResolvedTSMap, err := suite.client.GetMinResolvedTSByStoresIDs(suite.ctx, nil) + re.NoError(err) + re.Equal(testMinResolvedTS, minResolvedTS) + re.Empty(storeMinResolvedTSMap) // Get the store-level min resolved TS. minResolvedTS, storeMinResolvedTSMap, err = suite.client.GetMinResolvedTSByStoresIDs(suite.ctx, []uint64{1}) re.NoError(err) - re.Greater(minResolvedTS, uint64(0)) + re.Equal(testMinResolvedTS, minResolvedTS) re.Len(storeMinResolvedTSMap, 1) re.Equal(minResolvedTS, storeMinResolvedTSMap[1]) // Get the store-level min resolved TS with an invalid store ID. minResolvedTS, storeMinResolvedTSMap, err = suite.client.GetMinResolvedTSByStoresIDs(suite.ctx, []uint64{1, 2}) re.NoError(err) - re.Greater(minResolvedTS, uint64(0)) + re.Equal(testMinResolvedTS, minResolvedTS) re.Len(storeMinResolvedTSMap, 2) re.Equal(minResolvedTS, storeMinResolvedTSMap[1]) re.Equal(uint64(math.MaxUint64), storeMinResolvedTSMap[2])