kafka(ticdc): ddl sink close the underline sink if send ddl or checkpoint failed and refactor the kafka ddl sink (#12112)

cdc/owner/ddl_sink.go

@@ -173,7 +173,12 @@ func (s *ddlSinkImpl) retrySinkAction(ctx context.Context, name string, action f
 			zap.Bool("retryable", isRetryable),
 			zap.Error(err))
 
-		s.sink = nil
+		if s.sink != nil {
+			s.sink.Close()
+			s.sink = nil
+			log.Info("close the ddl sink, rebuild it when trying again",
+				zap.String("namespace", s.changefeedID.Namespace), zap.String("changefeed", s.changefeedID.ID))
+		}
 		if isRetryable {
 			s.reportWarning(err)
 		} else {
@@ -199,7 +204,7 @@ func (s *ddlSinkImpl) observedRetrySinkAction(ctx context.Context, name string,
 	defer ticker.Stop()
 	for {
 		select {
-		case err := <-errCh:
+		case err = <-errCh:
 			return err
 		case <-ticker.C:
 			log.Info("owner ddl sink performs an action too long",

cdc/sink/ddlsink/mq/ddlproducer/kafka_ddl_producer.go

@@ -69,8 +69,7 @@ func (k *kafkaDDLProducer) SyncBroadcastMessage(ctx context.Context, topic strin
 	case <-ctx.Done():
 		return ctx.Err()
 	default:
-		err := k.syncProducer.SendMessages(ctx, topic, totalPartitionsNum, message)
-		return cerror.WrapError(cerror.ErrKafkaSendMessage, err)
+		return k.syncProducer.SendMessages(ctx, topic, totalPartitionsNum, message)
 	}
 }
 
@@ -88,8 +87,7 @@ func (k *kafkaDDLProducer) SyncSendMessage(ctx context.Context, topic string,
 	case <-ctx.Done():
 		return errors.Trace(ctx.Err())
 	default:
-		err := k.syncProducer.SendMessage(ctx, topic, partitionNum, message)
-		return cerror.WrapError(cerror.ErrKafkaSendMessage, err)
+		return k.syncProducer.SendMessage(ctx, topic, partitionNum, message)
 	}
 }
 

cdc/sink/ddlsink/mq/mq_ddl_sink.go

@@ -16,7 +16,6 @@ package mq
 import (
 	"context"
 
-	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/sink/ddlsink"
@@ -100,7 +99,7 @@ func newDDLSink(
 func (k *DDLSink) WriteDDLEvent(ctx context.Context, ddl *model.DDLEvent) error {
 	msg, err := k.encoder.EncodeDDLEvent(ddl)
 	if err != nil {
-		return errors.Trace(err)
+		return err
 	}
 	if msg == nil {
 		log.Info("Skip ddl event", zap.Uint64("commitTs", ddl.CommitTs),
@@ -124,19 +123,17 @@ func (k *DDLSink) WriteDDLEvent(ctx context.Context, ddl *model.DDLEvent) error
 	// then the auto-created topic will not be created as configured by ticdc.
 	partitionNum, err := k.topicManager.GetPartitionNum(ctx, topic)
 	if err != nil {
-		return errors.Trace(err)
+		return err
 	}
 
 	if partitionRule == PartitionAll {
-		err = k.statistics.RecordDDLExecution(func() error {
+		return k.statistics.RecordDDLExecution(func() error {
 			return k.producer.SyncBroadcastMessage(ctx, topic, partitionNum, msg)
 		})
-		return errors.Trace(err)
 	}
-	err = k.statistics.RecordDDLExecution(func() error {
+	return k.statistics.RecordDDLExecution(func() error {
 		return k.producer.SyncSendMessage(ctx, topic, 0, msg)
 	})
-	return errors.Trace(err)
 }
 
 // WriteCheckpointTs sends the checkpoint ts to the MQ system.
@@ -146,7 +143,7 @@ func (k *DDLSink) WriteCheckpointTs(ctx context.Context,
 ) error {
 	msg, err := k.encoder.EncodeCheckpointEvent(ts)
 	if err != nil {
-		return errors.Trace(err)
+		return err
 	}
 	if msg == nil {
 		return nil
@@ -158,12 +155,11 @@ func (k *DDLSink) WriteCheckpointTs(ctx context.Context,
 		topic := k.eventRouter.GetDefaultTopic()
 		partitionNum, err := k.topicManager.GetPartitionNum(ctx, topic)
 		if err != nil {
-			return errors.Trace(err)
+			return err
 		}
 		log.Debug("Emit checkpointTs to default topic",
 			zap.String("topic", topic), zap.Uint64("checkpointTs", ts))
-		err = k.producer.SyncBroadcastMessage(ctx, topic, partitionNum, msg)
-		return errors.Trace(err)
+		return k.producer.SyncBroadcastMessage(ctx, topic, partitionNum, msg)
 	}
 	var tableNames []model.TableName
 	for _, table := range tables {
@@ -173,11 +169,11 @@ func (k *DDLSink) WriteCheckpointTs(ctx context.Context,
 	for _, topic := range topics {
 		partitionNum, err := k.topicManager.GetPartitionNum(ctx, topic)
 		if err != nil {
-			return errors.Trace(err)
+			return err
 		}
 		err = k.producer.SyncBroadcastMessage(ctx, topic, partitionNum, msg)
 		if err != nil {
-			return errors.Trace(err)
+			return err
 		}
 	}
 	return nil

pkg/sink/kafka/factory.go

@@ -98,10 +98,10 @@ func (p *saramaSyncProducer) SendMessage(
 		Value:     sarama.ByteEncoder(message.Value),
 		Partition: partitionNum,
 	})
-	return err
+	return cerror.WrapError(cerror.ErrKafkaSendMessage, err)
 }
 
-func (p *saramaSyncProducer) SendMessages(ctx context.Context, topic string, partitionNum int32, message *common.Message) error {
+func (p *saramaSyncProducer) SendMessages(_ context.Context, topic string, partitionNum int32, message *common.Message) error {
 	msgs := make([]*sarama.ProducerMessage, partitionNum)
 	for i := 0; i < int(partitionNum); i++ {
 		msgs[i] = &sarama.ProducerMessage{
@@ -111,7 +111,8 @@ func (p *saramaSyncProducer) SendMessages(ctx context.Context, topic string, par
 			Partition: int32(i),
 		}
 	}
-	return p.producer.SendMessages(msgs)
+	err := p.producer.SendMessages(msgs)
+	return cerror.WrapError(cerror.ErrKafkaSendMessage, err)
 }
 
 func (p *saramaSyncProducer) Close() {

pkg/sink/kafka/sarama_factory.go

@@ -109,7 +109,6 @@ func (f *saramaFactory) SyncProducer(ctx context.Context) (SyncProducer, error)
 		return nil, err
 	}
 	config.MetricRegistry = f.registry
-
 	client, err := newSaramaClientImpl(f.option.BrokerEndpoints, config)
 	if err != nil {
 		return nil, errors.Trace(err)
@@ -119,7 +118,6 @@ func (f *saramaFactory) SyncProducer(ctx context.Context) (SyncProducer, error)
 		closeSaramaClientOnFailure(f.changefeedID, client, "close sarama client after sync producer init failed")
 		return nil, errors.Trace(err)
 	}
-
 	return &saramaSyncProducer{
 		id:       f.changefeedID,
 		producer: p,

pkg/sink/kafka/v2/factory.go

@@ -277,12 +277,13 @@ func (s *syncWriter) SendMessage(
 	topic string, partitionNum int32,
 	message *common.Message,
 ) error {
-	return s.w.WriteMessages(ctx, kafka.Message{
+	err := s.w.WriteMessages(ctx, kafka.Message{
 		Topic:     topic,
 		Partition: int(partitionNum),
 		Key:       message.Key,
 		Value:     message.Value,
 	})
+	return errors.WrapError(errors.ErrKafkaSendMessage, err)
 }
 
 // SendMessages produces a given set of messages, and returns only when all
@@ -299,19 +300,17 @@ func (s *syncWriter) SendMessages(ctx context.Context, topic string, partitionNu
 			Partition: i,
 		}
 	}
-	return s.w.WriteMessages(ctx, msgs...)
+	err := s.w.WriteMessages(ctx, msgs...)
+	return errors.WrapError(errors.ErrKafkaSendMessage, err)
 }
 
 // Close shuts down the producer; you must call this function before a producer
 // object passes out of scope, as it may otherwise leak memory.
 // You must call this before calling Close on the underlying client.
 func (s *syncWriter) Close() {
-	log.Info("kafka sync producer start closing",
-		zap.String("namespace", s.changefeedID.Namespace),
-		zap.String("changefeed", s.changefeedID.ID))
 	start := time.Now()
 	if err := s.w.Close(); err != nil {
-		log.Warn("Close kafka sync producer failed",
+		log.Warn("Close kafka sync producer meet error",
 			zap.String("namespace", s.changefeedID.Namespace),
 			zap.String("changefeed", s.changefeedID.ID),
 			zap.Duration("duration", time.Since(start)),