changefeed: fix goroutine leak when changefeed is paused or stopped (#…

…1066) (#1075)

Signed-off-by: ti-srebot <ti-srebot@pingcap.com>

cdc/changefeed.go

@@ -114,6 +114,9 @@ type changeFeed struct {
 	lastRebalanceTime time.Time
 
 	etcdCli kv.CDCEtcdClient
+
+	// context cancel function for all internal goroutines
+	cancel context.CancelFunc
 }
 
 // String implements fmt.Stringer interface.
@@ -943,18 +946,19 @@ func (c *changeFeed) startSyncPointTicker(ctx context.Context, interval time.Dur
 
 func (c *changeFeed) Close() {
 	err := c.ddlHandler.Close()
-	if err != nil {
+	if err != nil && errors.Cause(err) != context.Canceled {
 		log.Warn("failed to close ddl handler", zap.Error(err))
 	}
 	err = c.sink.Close()
-	if err != nil {
+	if err != nil && errors.Cause(err) != context.Canceled {
 		log.Warn("failed to close owner sink", zap.Error(err))
 	}
 	if c.syncpointStore != nil {
 		err = c.syncpointStore.Close()
-		if err != nil {
+		if err != nil && errors.Cause(err) != context.Canceled {
 			log.Warn("failed to close owner sink", zap.Error(err))
 		}
 	}
+	c.cancel()
 	log.Info("changefeed closed", zap.String("id", c.id))
 }

cdc/owner.go

@@ -342,13 +342,13 @@ func (o *Owner) newChangeFeed(
 		select {
 		case <-ctx.Done():
 		case err = <-errCh:
+			cancel()
 		}
 		if err != nil && errors.Cause(err) != context.Canceled {
-			log.Error("error on running owner", zap.Error(err))
+			log.Error("error on running changefeed", zap.Error(err), zap.String("changefeed", id))
 		} else {
-			log.Info("owner exited")
+			log.Info("changefeed exited", zap.String("changfeed", id))
 		}
-		cancel()
 	}()
 
 	err = primarySink.Initialize(ctx, sinkTableInfo)
@@ -394,6 +394,7 @@ func (o *Owner) newChangeFeed(
 		sink:              primarySink,
 		cyclicEnabled:     info.Config.Cyclic.IsEnabled(),
 		lastRebalanceTime: time.Now(),
+		cancel:            cancel,
 	}
 	return cf, nil
 }

cdc/owner_test.go

@@ -484,6 +484,7 @@ func (s *ownerSuite) TestHandleAdmin(c *check.C) {
 			cancel: cancel,
 			wg:     errg,
 		},
+		cancel: cancel,
 	}
 	errCh := make(chan error, 1)
 	sink, err := sink.NewSink(ctx, cfID, "blackhole://", f, replicaConf, map[string]string{}, errCh)
@@ -537,6 +538,15 @@ func (s *ownerSuite) TestHandleAdmin(c *check.C) {
 	st, _, err := owner.etcdClient.GetChangeFeedStatus(ctx, cfID)
 	c.Assert(err, check.IsNil)
 	c.Assert(st.AdminJobType, check.Equals, model.AdminStop)
+	// check changefeed context is canceled
+	select {
+	case <-cctx.Done():
+	default:
+		c.Fatal("changefeed context is expected canceled")
+	}
+
+	cctx, cancel = context.WithCancel(ctx)
+	sampleCF.cancel = cancel
 
 	c.Assert(owner.EnqueueJob(model.AdminJob{CfID: cfID, Type: model.AdminResume}), check.IsNil)
 	c.Assert(owner.handleAdminJob(ctx), check.IsNil)
@@ -568,6 +578,12 @@ func (s *ownerSuite) TestHandleAdmin(c *check.C) {
 	st, _, err = owner.etcdClient.GetChangeFeedStatus(ctx, cfID)
 	c.Assert(err, check.IsNil)
 	c.Assert(st.AdminJobType, check.Equals, model.AdminRemove)
+	// check changefeed context is canceled
+	select {
+	case <-cctx.Done():
+	default:
+		c.Fatal("changefeed context is expected canceled")
+	}
 }
 
 func (s *ownerSuite) TestChangefeedApplyDDLJob(c *check.C) {

cdc/processor.go

@@ -329,12 +329,14 @@ func (p *processor) positionWorker(ctx context.Context) error {
 			if inErr != nil {
 				if errors.Cause(inErr) != context.Canceled {
 					logError := log.Error
+					errField := zap.Error(inErr)
 					if cerror.ErrAdminStopProcessor.Equal(inErr) {
 						logError = log.Warn
+						errField = zap.String("error", inErr.Error())
 					}
 					logError(
 						"update info failed",
-						zap.String("changefeed", p.changefeedID), zap.Error(inErr),
+						zap.String("changefeed", p.changefeedID), errField,
 					)
 				}
 				if p.isStopped() || cerror.ErrAdminStopProcessor.Equal(inErr) {