owner: fix a bug which lead to replication stopped and no error report (#1814)

cdc/changefeed.go

@@ -572,6 +572,7 @@ func (c *changeFeed) handleMoveTableJobs(ctx context.Context, captures map[model
 			if !exist {
 				// the target capture is not exist, add table to orphanTables.
 				c.orphanTables[tableID] = replicaInfo.StartTs
+				delete(c.moveTableJobs, tableID)
 				log.Warn("the target capture is not exist, sent the table to orphanTables", zap.Reflect("job", job))
 				continue
 			}

cdc/changefeed_test.go

@@ -0,0 +1,130 @@
+// Copyright 2021 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package cdc
+
+import (
+	"context"
+	"net/url"
+	"time"
+
+	"github.com/pingcap/check"
+	"github.com/pingcap/ticdc/cdc/kv"
+	"github.com/pingcap/ticdc/cdc/model"
+	"github.com/pingcap/ticdc/pkg/etcd"
+	"github.com/pingcap/ticdc/pkg/util"
+	"github.com/pingcap/ticdc/pkg/util/testleak"
+	"go.etcd.io/etcd/clientv3"
+	"go.etcd.io/etcd/clientv3/concurrency"
+	"go.etcd.io/etcd/embed"
+	"golang.org/x/sync/errgroup"
+)
+
+type changefeedSuite struct {
+	e         *embed.Etcd
+	clientURL *url.URL
+	client    kv.CDCEtcdClient
+	sess      *concurrency.Session
+	ctx       context.Context
+	cancel    context.CancelFunc
+	errg      *errgroup.Group
+}
+
+var _ = check.Suite(&changefeedSuite{})
+
+func (s *changefeedSuite) SetUpTest(c *check.C) {
+	dir := c.MkDir()
+	var err error
+	s.clientURL, s.e, err = etcd.SetupEmbedEtcd(dir)
+	c.Assert(err, check.IsNil)
+	client, err := clientv3.New(clientv3.Config{
+		Endpoints:   []string{s.clientURL.String()},
+		DialTimeout: 3 * time.Second,
+	})
+	c.Assert(err, check.IsNil)
+	sess, err := concurrency.NewSession(client)
+	c.Assert(err, check.IsNil)
+	s.sess = sess
+	s.client = kv.NewCDCEtcdClient(context.Background(), client)
+	s.ctx, s.cancel = context.WithCancel(context.Background())
+	s.errg = util.HandleErrWithErrGroup(s.ctx, s.e.Err(), func(e error) { c.Log(e) })
+}
+
+func (s *changefeedSuite) TearDownTest(c *check.C) {
+	s.e.Close()
+	s.cancel()
+	err := s.errg.Wait()
+	if err != nil {
+		c.Errorf("Error group error: %s", err)
+	}
+	s.client.Close() //nolint:errcheck
+}
+
+func (s *changefeedSuite) TestHandleMoveTableJobs(c *check.C) {
+	defer testleak.AfterTest(c)()
+	defer s.TearDownTest(c)
+	changefeed := new(changeFeed)
+	captureID1 := "capture1"
+	captureID2 := "capture2"
+	changefeed.id = "changefeed-test"
+	changefeed.leaseID = s.sess.Lease()
+	changefeed.etcdCli = s.client
+	changefeed.status = new(model.ChangeFeedStatus)
+	changefeed.orphanTables = map[model.TableID]model.Ts{}
+	captures := make(map[model.CaptureID]*model.CaptureInfo)
+	captures[captureID1] = &model.CaptureInfo{
+		ID: captureID1,
+	}
+	captures[captureID2] = &model.CaptureInfo{
+		ID: captureID2,
+	}
+	changefeed.taskStatus = make(map[model.CaptureID]*model.TaskStatus)
+	changefeed.taskStatus[captureID1] = &model.TaskStatus{Tables: map[model.TableID]*model.TableReplicaInfo{1: {}}}
+	changefeed.taskStatus[captureID2] = &model.TaskStatus{Tables: map[model.TableID]*model.TableReplicaInfo{}}
+	changefeed.moveTableJobs = make(map[model.TableID]*model.MoveTableJob)
+	changefeed.moveTableJobs[1] = &model.MoveTableJob{
+		TableID:          1,
+		From:             captureID1,
+		To:               captureID2,
+		TableReplicaInfo: new(model.TableReplicaInfo),
+	}
+	err := changefeed.handleMoveTableJobs(s.ctx, captures)
+	c.Assert(err, check.IsNil)
+	taskStatuses, err := s.client.GetAllTaskStatus(s.ctx, changefeed.id)
+	c.Assert(err, check.IsNil)
+	taskStatuses[captureID1].ModRevision = 0
+	c.Assert(taskStatuses, check.DeepEquals, model.ProcessorsInfos{captureID1: {
+		Tables: map[model.TableID]*model.TableReplicaInfo{},
+		Operation: map[model.TableID]*model.TableOperation{1: {
+			Delete: true,
+			Flag:   model.OperFlagMoveTable,
+		}},
+	}})
+
+	// finish operation
+	err = s.client.PutTaskStatus(s.ctx, changefeed.id, captureID1, &model.TaskStatus{
+		Tables:    map[model.TableID]*model.TableReplicaInfo{},
+		Operation: map[model.TableID]*model.TableOperation{},
+	})
+	c.Assert(err, check.IsNil)
+	delete(changefeed.taskStatus[captureID1].Operation, 1)
+
+	// capture2 offline
+	delete(captures, captureID2)
+	delete(changefeed.taskStatus, captureID2)
+
+	err = changefeed.handleMoveTableJobs(s.ctx, captures)
+	c.Assert(err, check.IsNil)
+	c.Assert(changefeed.orphanTables, check.HasKey, model.TableID(1))
+	c.Assert(changefeed.moveTableJobs, check.HasLen, 0)
+}