owner_test (ticdc): fix unstable case TestHandleJobsDontBlock

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) {
@@ -373,22 +372,24 @@ func (s *ownerSuite) TestHandleJobsDontBlock(c *check.C) {
 	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 +406,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 <-ctx1.Done():
+			c.Fatal(ctx1.Err())
+		case <-ticker.C:
+			_, err = owner.Tick(ctx, state)
+			c.Assert(err, check.IsNil)
+		case <-done:
+			break WorkLoop
+		}
+	}
+	c.Assert(errIn, check.IsNil)
+	c.Assert(infos[cf1], check.NotNil)
+	c.Assert(infos[cf2], check.IsNil)
 }