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

@@ -310,8 +310,13 @@ func (k *mqSink) EmitDDLEvent(ctx context.Context, ddl *model.DDLEvent) error {
 }
 
 func (k *mqSink) Close(ctx context.Context) error {
-	err := k.mqProducer.Close()
-	return errors.Trace(err)
+	closedCh := k.mqProducer.AsyncClose()
+	select {
+	case <-ctx.Done():
+		return errors.Trace(ctx.Err())
+	case err := <-closedCh:
+		return errors.Trace(err)
+	}
 }
 
 func (k *mqSink) Barrier(cxt context.Context, tableID model.TableID) error {

cdc/sink/mq_flush_worker_test.go

@@ -69,6 +69,10 @@ func (m *mockProducer) Close() error {
 	panic("Not used")
 }
 
+func (m *mockProducer) AsyncClose() chan error {
+	panic("Not used")
+}
+
 func (m *mockProducer) InjectError(err error) {
 	m.mockErr <- err
 }

cdc/sink/producer/kafka/kafka.go

@@ -195,6 +195,16 @@ func (k *kafkaSaramaProducer) stop() {
 	close(k.closeCh)
 }
 
+// AsyncClose close the produce in an asynchronous way.
+func (k *kafkaSaramaProducer) AsyncClose() chan error {
+	ret := make(chan error, 1)
+	go func() {
+		ret <- k.Close()
+		close(ret)
+	}()
+	return ret
+}
+
 // Close closes the sync and async clients.
 func (k *kafkaSaramaProducer) Close() error {
 	log.Info("stop the kafka producer", zap.String("changefeed", k.id), zap.Any("role", k.role))

cdc/sink/producer/mq_producer.go

@@ -34,4 +34,6 @@ type Producer interface {
 	Flush(ctx context.Context) error
 	// Close closes the producer and client(s).
 	Close() error
+	// AsyncClose close the produce in an asynchronous way.
+	AsyncClose() chan error
 }

cdc/sink/producer/pulsar/producer.go

@@ -146,3 +146,13 @@ func (p *Producer) Close() error {
 	p.client.Close()
 	return nil
 }
+
+// AsyncClose close the produce in an asynchronous way.
+func (p *Producer) AsyncClose() chan error {
+	ret := make(chan error, 1)
+	go func() {
+		ret <- p.Close()
+		close(ret)
+	}()
+	return ret
+}