sink(ticdc): asynchronously close the mq producers

cdc/owner/ddl_sink.go

@@ -264,5 +264,8 @@ func (s *ddlSinkImpl) close(ctx context.Context) (err error) {
 		err = s.syncPointStore.Close()
 	}
 	s.wg.Wait()
-	return err
+	if err != nil && errors.Cause(err) != context.Canceled {
+		return err
+	}
+	return nil
 }

cdc/processor/pipeline/table_actor.go

@@ -140,7 +140,7 @@ func NewTableActor(cdcCtx cdcContext.Context,
 
 	startTime := time.Now()
 	log.Info("table actor starting",
-		zap.String("changefeed", changefeedVars.ID),
+		zap.String("changefeed", table.changefeedID),
 		zap.String("tableName", tableName),
 		zap.Int64("tableID", tableID))
 	if err := table.start(cctx); err != nil {
@@ -152,7 +152,7 @@ func NewTableActor(cdcCtx cdcContext.Context,
 		return nil, errors.Trace(err)
 	}
 	log.Info("table actor started",
-		zap.String("changefeed", changefeedVars.ID),
+		zap.String("changefeed", table.changefeedID),
 		zap.String("tableName", tableName),
 		zap.Int64("tableID", tableID),
 		zap.Duration("duration", time.Since(startTime)))

cdc/sink/manager.go

@@ -17,7 +17,6 @@ import (
 	"context"
 	"sync"
 	"sync/atomic"
-	"time"
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
@@ -83,18 +82,12 @@ func (m *Manager) Close(ctx context.Context) error {
 	defer m.tableSinksMu.Unlock()
 	tableSinkTotalRowsCountCounter.DeleteLabelValues(m.changefeedID)
 	if m.bufSink != nil {
-		log.Info("sinkManager try close bufSink",
-			zap.String("changefeed", m.changefeedID))
-		start := time.Now()
-		if err := m.bufSink.Close(ctx); err != nil {
-			log.Info("close bufSink failed",
+		if err := m.bufSink.Close(ctx); err != nil && errors.Cause(err) != context.Canceled {
+			log.Warn("close bufSink failed",
 				zap.String("changefeed", m.changefeedID),
-				zap.Duration("duration", time.Since(start)))
+				zap.Error(err))
 			return err
 		}
-		log.Info("close bufSink success",
-			zap.String("changefeed", m.changefeedID),
-			zap.Duration("duration", time.Since(start)))
 	}
 	return nil
 }

cdc/sink/mq.go

@@ -309,9 +309,10 @@ func (k *mqSink) EmitDDLEvent(ctx context.Context, ddl *model.DDLEvent) error {
 	return errors.Trace(err)
 }
 
+// Close the producer asynchronously, does not care closed successfully or not.
 func (k *mqSink) Close(ctx context.Context) error {
-	err := k.mqProducer.Close()
-	return errors.Trace(err)
+	go k.mqProducer.Close()
+	return nil
 }
 
 func (k *mqSink) Barrier(cxt context.Context, tableID model.TableID) error {