owner_test (ticdc): fix unstable case TestHandleJobsDontBlock

cdc/owner/owner.go

@@ -33,6 +33,7 @@ import (
 	"github.com/pingcap/ticdc/pkg/version"
 	pd "github.com/tikv/pd/client"
 	"go.uber.org/zap"
+	"golang.org/x/time/rate"
 )
 
 type ownerJobType int
@@ -46,6 +47,10 @@ const (
 	ownerJobTypeQuery
 )
 
+// versionInconsistentLogRate represents the rate of log output when there are
+// captures with versions different from that of the owner
+const versionInconsistentLogRate = 1
+
 type ownerJob struct {
 	tp           ownerJobType
 	changefeedID model.ChangeFeedID
@@ -77,10 +82,10 @@ type Owner struct {
 
 	ownerJobQueueMu sync.Mutex
 	ownerJobQueue   []*ownerJob
-
+	// logLimiter controls cluster version check log output rate
+	logLimiter   *rate.Limiter
 	lastTickTime time.Time
-
-	closed int32
+	closed       int32
 
 	newChangefeed func(id model.ChangeFeedID, gcManager gc.Manager) *changefeed
 }
@@ -92,6 +97,7 @@ func NewOwner(pdClient pd.Client) *Owner {
 		gcManager:     gc.NewManager(pdClient),
 		lastTickTime:  time.Now(),
 		newChangefeed: newChangefeed,
+		logLimiter:    rate.NewLimiter(versionInconsistentLogRate, versionInconsistentLogRate),
 	}
 }
 
@@ -126,8 +132,6 @@ func (o *Owner) Tick(stdCtx context.Context, rawState orchestrator.ReactorState)
 	o.handleJobs()
 
 	if !o.clusterVersionConsistent(state.Captures) {
-		// sleep one second to avoid printing too much log
-		time.Sleep(1 * time.Second)
 		return state, nil
 	}
 	// Owner should update GC safepoint before initializing changefeed, so
@@ -280,7 +284,9 @@ func (o *Owner) clusterVersionConsistent(captures map[model.CaptureID]*model.Cap
 	myVersion := version.ReleaseVersion
 	for _, capture := range captures {
 		if myVersion != capture.Version {
-			log.Warn("the capture version is different with the owner", zap.Reflect("capture", capture), zap.String("my-version", myVersion))
+			if o.logLimiter.Allow() {
+				log.Warn("the capture version is different with the owner", zap.Reflect("capture", capture), zap.String("owner-version", myVersion))
+			}
 			return false
 		}
 	}

cdc/owner/owner_test.go

@@ -18,7 +18,6 @@ import (
 	"context"
 	"fmt"
 	"math"
-	"sync"
 	"time"
 
 	"github.com/pingcap/check"
@@ -55,7 +54,7 @@ func createOwner4Test(ctx cdcContext.Context, c *check.C) (*Owner, *orchestrator
 			return safePoint, nil
 		},
 	}
-	cf := NewOwner4Test(func(ctx cdcContext.Context, startTs uint64) (DDLPuller, error) {
+	owner := NewOwner4Test(func(ctx cdcContext.Context, startTs uint64) (DDLPuller, error) {
 		return &mockDDLPuller{resolvedTs: startTs - 1}, nil
 	}, func(ctx cdcContext.Context) (AsyncSink, error) {
 		return &mockAsyncSink{}, nil
@@ -73,7 +72,7 @@ func createOwner4Test(ctx cdcContext.Context, c *check.C) (*Owner, *orchestrator
 	captureBytes, err := ctx.GlobalVars().CaptureInfo.Marshal()
 	c.Assert(err, check.IsNil)
 	tester.MustUpdate(cdcKey.String(), captureBytes)
-	return cf, state, tester
+	return owner, state, tester
 }
 
 func (s *ownerSuite) TestCreateRemoveChangefeed(c *check.C) {
@@ -371,24 +370,27 @@ func (s *ownerSuite) TestHandleJobsDontBlock(c *check.C) {
 	defer testleak.AfterTest(c)()
 	ctx := cdcContext.NewBackendContext4Test(false)
 	owner, state, tester := createOwner4Test(ctx, c)
+
 	statusProvider := owner.StatusProvider()
 	// work well
-	changefeedID := "test-changefeed"
-	changefeedInfo := &model.ChangeFeedInfo{
+	cf1 := "test-changefeed"
+	cfInfo1 := &model.ChangeFeedInfo{
 		StartTs: oracle.GoTimeToTS(time.Now()),
 		Config:  config.GetDefaultReplicaConfig(),
+		State:   model.StateNormal,
 	}
-	changefeedStr, err := changefeedInfo.Marshal()
+	changefeedStr, err := cfInfo1.Marshal()
 	c.Assert(err, check.IsNil)
 	cdcKey := etcd.CDCKey{
 		Tp:           etcd.CDCKeyTypeChangefeedInfo,
-		ChangefeedID: changefeedID,
+		ChangefeedID: cf1,
 	}
 	tester.MustUpdate(cdcKey.String(), []byte(changefeedStr))
 	_, err = owner.Tick(ctx, state)
 	tester.MustApplyPatches()
 	c.Assert(err, check.IsNil)
-	c.Assert(owner.changefeeds, check.HasKey, changefeedID)
+
+	c.Assert(owner.changefeeds, check.HasKey, cf1)
 
 	// add an non-consistent version capture
 	captureInfo := &model.CaptureInfo{
@@ -405,37 +407,53 @@ func (s *ownerSuite) TestHandleJobsDontBlock(c *check.C) {
 	tester.MustUpdate(cdcKey.String(), v)
 
 	// try to add another changefeed
-	changefeedID1 := "test-changefeed1"
-	changefeedInfo1 := &model.ChangeFeedInfo{
+	cf2 := "test-changefeed1"
+	cfInfo2 := &model.ChangeFeedInfo{
 		StartTs: oracle.GoTimeToTS(time.Now()),
 		Config:  config.GetDefaultReplicaConfig(),
+		State:   model.StateNormal,
 	}
-	changefeedStr1, err := changefeedInfo1.Marshal()
+	changefeedStr1, err := cfInfo2.Marshal()
 	c.Assert(err, check.IsNil)
 	cdcKey = etcd.CDCKey{
 		Tp:           etcd.CDCKeyTypeChangefeedInfo,
-		ChangefeedID: changefeedID,
+		ChangefeedID: cf2,
 	}
 	tester.MustUpdate(cdcKey.String(), []byte(changefeedStr1))
 	_, err = owner.Tick(ctx, state)
 	tester.MustApplyPatches()
 	c.Assert(err, check.IsNil)
 	// make sure this changefeed add failed, which means that owner are return
 	// in clusterVersionConsistent check
-	c.Assert(owner.changefeeds[changefeedID1], check.IsNil)
+	c.Assert(owner.changefeeds[cf2], check.IsNil)
 
 	// make sure statusProvider works well
 	ctx1, cancel := context.WithTimeout(context.Background(), time.Second*5)
 	defer cancel()
-	wg := sync.WaitGroup{}
-	wg.Add(1)
+
+	var errIn error
+	var infos map[model.ChangeFeedID]*model.ChangeFeedInfo
+	done := make(chan struct{})
 	go func() {
-		infos, err := statusProvider.GetAllChangeFeedInfo(ctx1)
-		c.Assert(err, check.IsNil)
-		c.Assert(infos[changefeedID], check.NotNil)
-		c.Assert(infos[changefeedID1], check.IsNil)
-		wg.Done()
+		infos, errIn = statusProvider.GetAllChangeFeedInfo(ctx1)
+		done <- struct{}{}
 	}()
-	_, err = owner.Tick(ctx, state)
-	c.Assert(err, check.IsNil)
+
+	ticker := time.NewTicker(20 * time.Millisecond)
+	defer ticker.Stop()
+WorkLoop:
+	for {
+		select {
+		case <-done:
+			break WorkLoop
+		case <-ctx1.Done():
+			c.Fatal(ctx1.Err())
+		case <-ticker.C:
+			_, err = owner.Tick(ctx, state)
+			c.Assert(err, check.IsNil)
+		}
+	}
+	c.Assert(errIn, check.IsNil)
+	c.Assert(infos[cf1], check.NotNil)
+	c.Assert(infos[cf2], check.IsNil)
 }