owner: fix multiple owners exist when owner campaign key is deleted

cdc/owner.go

@@ -975,12 +975,18 @@ func (o *Owner) Run(ctx context.Context, tickTime time.Duration) error {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
+	go func() {
+		if err := o.watchCampaignKey(ctx); err != nil {
+			cancel()
+		}
+	}()
+
 	if err := o.throne(ctx); err != nil {
 		return err
 	}
 
-	ctx1, cancel := context.WithCancel(ctx)
-	defer cancel()
+	ctx1, cancel1 := context.WithCancel(ctx)
+	defer cancel1()
 	changedFeeds := o.watchFeedChange(ctx1)
 
 	ticker := time.NewTicker(tickTime)
@@ -994,7 +1000,10 @@ loop:
 			close(o.done)
 			break loop
 		case <-ctx.Done():
-			return ctx.Err()
+			// FIXME: cancel the context doesn't ensure all resources are destructed, is it reasonable?
+			// Anyway we just break loop here to ensure the following destruction.
+			err = ctx.Err()
+			break loop
 		case <-changedFeeds:
 		case <-ticker.C:
 		}
@@ -1019,6 +1028,36 @@ loop:
 	return err
 }
 
+// watchCampaignKey watches the aliveness of campaign owner key in etcd
+func (o *Owner) watchCampaignKey(ctx context.Context) error {
+	key := fmt.Sprintf("%s/%x", kv.CaptureOwnerKey, o.session.Lease())
+	resp, err := o.etcdClient.Client.Get(ctx, key)
+	if err != nil {
+		return cerror.WrapError(cerror.ErrPDEtcdAPIError, err)
+	}
+	if resp.Count == 0 {
+		return cerror.ErrOwnerCampaignKeyDeleted.GenWithStackByArgs()
+	}
+restart:
+	wch := o.etcdClient.Client.Watch(ctx, key, clientv3.WithRev(resp.Header.Revision+1))
+	for resp := range wch {
+		err := resp.Err()
+		if err != nil {
+			if err != mvcc.ErrCompacted {
+				log.Error("watch owner campaign key failed, restart the watcher", zap.Error(err))
+			}
+			goto restart
+		}
+		for _, ev := range resp.Events {
+			if ev.Type == clientv3.EventTypeDelete {
+				log.Warn("owner campaign key deleted", zap.String("key", key))
+				return cerror.ErrOwnerCampaignKeyDeleted.GenWithStackByArgs()
+			}
+		}
+	}
+	return nil
+}
+
 func (o *Owner) watchFeedChange(ctx context.Context) chan struct{} {
 	output := make(chan struct{}, 1)
 	go func() {

cdc/owner_test.go

@@ -16,7 +16,9 @@ package cdc
 import (
 	"context"
 	"errors"
+	"fmt"
 	"net/url"
+	"sync"
 	"time"
 
 	"github.com/pingcap/check"
@@ -768,3 +770,53 @@ func (s *ownerSuite) TestChangefeedApplyDDLJob(c *check.C) {
 		c.Assert(cf.tables, check.DeepEquals, expectTables[i])
 	}
 }
+
+func (s *ownerSuite) TestWatchCampaignKey(c *check.C) {
+	ctx := context.Background()
+	capture, err := NewCapture(ctx, []string{s.clientURL.String()},
+		&security.Credential{}, "127.0.0.1:12034", &processorOpts{})
+	c.Assert(err, check.IsNil)
+	err = capture.Campaign(ctx)
+	c.Assert(err, check.IsNil)
+
+	cctx, cancel := context.WithCancel(ctx)
+	owner, err := NewOwner(cctx, nil, &security.Credential{}, capture.session,
+		DefaultCDCGCSafePointTTL, time.Millisecond*200)
+	c.Assert(err, check.IsNil)
+
+	// check campaign key deleted can be detected
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		err := owner.watchCampaignKey(cctx)
+		c.Assert(cerror.ErrOwnerCampaignKeyDeleted.Equal(err), check.IsTrue)
+		cancel()
+	}()
+	// ensure the watch loop has started
+	time.Sleep(time.Millisecond * 100)
+	etcdCli := owner.etcdClient.Client.Unwrap()
+	key := fmt.Sprintf("%s/%x", kv.CaptureOwnerKey, owner.session.Lease())
+	_, err = etcdCli.Delete(ctx, key)
+	c.Assert(err, check.IsNil)
+	wg.Wait()
+
+	// check key is deleted before watch loop starts
+	cctx, cancel = context.WithCancel(ctx)
+	err = owner.watchCampaignKey(cctx)
+	c.Assert(cerror.ErrOwnerCampaignKeyDeleted.Equal(err), check.IsTrue)
+
+	// check the watch routine can be canceled
+	err = capture.Campaign(ctx)
+	c.Assert(err, check.IsNil)
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		err := owner.watchCampaignKey(cctx)
+		c.Assert(err, check.IsNil)
+	}()
+	// ensure the watch loop has started
+	time.Sleep(time.Millisecond * 100)
+	cancel()
+	wg.Wait()
+}

errors.toml

@@ -451,6 +451,11 @@ error = '''
 old value is not enabled
 '''
 
+["CDC:ErrOwnerCampaignKeyDeleted"]
+error = '''
+owner campaign key deleted
+'''
+
 ["CDC:ErrOwnerChangefeedNotFound"]
 error = '''
 changefeed %s not found in owner cache

pkg/errors/errors.go

@@ -173,4 +173,5 @@ var (
 	ErrChangefeedAbnormalState    = errors.Normalize("changefeed in abnormal state: %s, replication status: %+v", errors.RFCCodeText("CDC:ErrChangefeedAbnormalState"))
 	ErrInvalidAdminJobType        = errors.Normalize("invalid admin job type: %d", errors.RFCCodeText("CDC:ErrInvalidAdminJobType"))
 	ErrOwnerEtcdWatch             = errors.Normalize("etcd watch returns error", errors.RFCCodeText("CDC:ErrOwnerEtcdWatch"))
+	ErrOwnerCampaignKeyDeleted    = errors.Normalize("owner campaign key deleted", errors.RFCCodeText("CDC:ErrOwnerCampaignKeyDeleted"))
 )