owner: fix data race on ownerManager.campaignCancel

pkg/owner/manager.go

@@ -196,7 +196,9 @@ func (m *ownerManager) CampaignOwner(withTTL ...int) error {
 	}
 	m.sessionLease.Store(int64(session.Lease()))
 	m.wg.Add(1)
-	go m.campaignLoop(session)
+	var campaignContext context.Context
+	campaignContext, m.campaignCancel = context.WithCancel(m.ctx)
+	go m.campaignLoop(session, campaignContext)
 	return nil
 }
 
@@ -241,9 +243,7 @@ func (m *ownerManager) CampaignCancel() {
 	m.wg.Wait()
 }
 
-func (m *ownerManager) campaignLoop(etcdSession *concurrency.Session) {
-	var campaignContext context.Context
-	campaignContext, m.campaignCancel = context.WithCancel(m.ctx)
+func (m *ownerManager) campaignLoop(etcdSession *concurrency.Session, campaignContext context.Context) {
 	defer func() {
 		m.campaignCancel()
 		if r := recover(); r != nil {

pkg/owner/manager_test.go

@@ -439,6 +439,24 @@ func deleteLeader(cli *clientv3.Client, prefixKey string) error {
 	return errors.Trace(err)
 }
 
+func TestImmediatelyCancel(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		t.Skip("integration.NewClusterV3 will create file contains a colon which is not allowed on Windows")
+	}
+	integration.BeforeTestExternal(t)
+
+	tInfo := newTestInfo(t)
+	d := tInfo.ddl
+	defer tInfo.Close(t)
+	ownerManager := d.OwnerManager()
+	for i := 0; i < 100; i++ {
+		err := ownerManager.CampaignOwner()
+		//time.Sleep(100 * time.Millisecond)
+		require.NoError(t, err)
+		ownerManager.CampaignCancel()
+	}
+}
+
 func TestAcquireDistributedLock(t *testing.T) {
 	const addrFmt = "http://127.0.0.1:%d"
 	cfg := embed.NewConfig()