From ade1f9ef752944187af1417d2b4e1e1081eea7db Mon Sep 17 00:00:00 2001 From: Ling Jin <7138436+3AceShowHand@users.noreply.github.com> Date: Wed, 2 Apr 2025 16:44:19 +0800 Subject: [PATCH 1/3] This is an automated cherry-pick of #12112 Signed-off-by: ti-chi-bot --- cdc/owner/ddl_sink.go | 9 ++- .../mq/ddlproducer/kafka_ddl_producer.go | 6 +- cdc/sink/ddlsink/mq/mq_ddl_sink.go | 33 +++++----- cdc/sink/ddlsink/mq/pulsar_ddl_sink.go | 4 ++ pkg/sink/kafka/factory.go | 60 ++++++------------- pkg/sink/kafka/sarama_factory.go | 7 +++ pkg/sink/kafka/v2/factory.go | 12 ++-- 7 files changed, 62 insertions(+), 69 deletions(-) diff --git a/cdc/owner/ddl_sink.go b/cdc/owner/ddl_sink.go index ecbe8c3444..4c5d9250ee 100644 --- a/cdc/owner/ddl_sink.go +++ b/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", diff --git a/cdc/sink/ddlsink/mq/ddlproducer/kafka_ddl_producer.go b/cdc/sink/ddlsink/mq/ddlproducer/kafka_ddl_producer.go index 7a2d2b3847..d69bd2d6b7 100644 --- a/cdc/sink/ddlsink/mq/ddlproducer/kafka_ddl_producer.go +++ b/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) } } diff --git a/cdc/sink/ddlsink/mq/mq_ddl_sink.go b/cdc/sink/ddlsink/mq/mq_ddl_sink.go index 71af7f09f4..563bb0efeb 100644 --- a/cdc/sink/ddlsink/mq/mq_ddl_sink.go +++ b/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. @@ -144,9 +141,16 @@ func (k *DDLSink) WriteDDLEvent(ctx context.Context, ddl *model.DDLEvent) error func (k *DDLSink) WriteCheckpointTs(ctx context.Context, ts uint64, tables []*model.TableInfo, ) error { +<<<<<<< HEAD +======= + var ( + err error + partitionNum int32 + ) +>>>>>>> 4c631d5951 (kafka(ticdc): ddl sink close the underline sink if send ddl or checkpoint failed and refactor the kafka ddl sink (#12112)) msg, err := k.encoder.EncodeCheckpointEvent(ts) if err != nil { - return errors.Trace(err) + return err } if msg == nil { return nil @@ -156,14 +160,13 @@ func (k *DDLSink) WriteCheckpointTs(ctx context.Context, // This will be compatible with the old behavior. if len(tables) == 0 { topic := k.eventRouter.GetDefaultTopic() - partitionNum, err := k.topicManager.GetPartitionNum(ctx, topic) + 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 { @@ -171,13 +174,13 @@ func (k *DDLSink) WriteCheckpointTs(ctx context.Context, } topics := k.eventRouter.GetActiveTopics(tableNames) for _, topic := range topics { - partitionNum, err := k.topicManager.GetPartitionNum(ctx, topic) + 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 diff --git a/cdc/sink/ddlsink/mq/pulsar_ddl_sink.go b/cdc/sink/ddlsink/mq/pulsar_ddl_sink.go index 96f6722f11..e55ae1d698 100644 --- a/cdc/sink/ddlsink/mq/pulsar_ddl_sink.go +++ b/cdc/sink/ddlsink/mq/pulsar_ddl_sink.go @@ -101,5 +101,9 @@ func NewPulsarDDLSink( } s := newDDLSink(changefeedID, p, nil, topicManager, eventRouter, encoderBuilder.Build(), protocol) +<<<<<<< HEAD +======= + +>>>>>>> 4c631d5951 (kafka(ticdc): ddl sink close the underline sink if send ddl or checkpoint failed and refactor the kafka ddl sink (#12112)) return s, nil } diff --git a/pkg/sink/kafka/factory.go b/pkg/sink/kafka/factory.go index 7bdc73c268..1518dcdd91 100644 --- a/pkg/sink/kafka/factory.go +++ b/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,49 +111,25 @@ 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() { - go func() { - // We need to close it asynchronously. Otherwise, we might get stuck - // with an unhealthy(i.e. Network jitter, isolation) state of Kafka. - // Factory has a background thread to fetch and update the metadata. - // If we close the client synchronously, we might get stuck. - // Safety: - // * If the kafka cluster is running well, it will be closed as soon as possible. - // * If there is a problem with the kafka cluster, - // no data will be lost because this is a synchronous client. - // * There is a risk of goroutine leakage, but it is acceptable and our main - // goal is not to get stuck with the owner tick. - start := time.Now() - if err := p.client.Close(); err != nil { - log.Warn("Close Kafka DDL client with error", - zap.String("namespace", p.id.Namespace), - zap.String("changefeed", p.id.ID), - zap.Duration("duration", time.Since(start)), - zap.Error(err)) - } else { - log.Info("Kafka DDL client closed", - zap.String("namespace", p.id.Namespace), - zap.String("changefeed", p.id.ID), - zap.Duration("duration", time.Since(start))) - } - start = time.Now() - err := p.producer.Close() - if err != nil { - log.Error("Close Kafka DDL producer with error", - zap.String("namespace", p.id.Namespace), - zap.String("changefeed", p.id.ID), - zap.Duration("duration", time.Since(start)), - zap.Error(err)) - } else { - log.Info("Kafka DDL producer closed", - zap.String("namespace", p.id.Namespace), - zap.String("changefeed", p.id.ID), - zap.Duration("duration", time.Since(start))) - } - }() + start := time.Now() + err := p.producer.Close() + if err != nil { + log.Error("Close Kafka DDL producer with error", + zap.String("namespace", p.id.Namespace), + zap.String("changefeed", p.id.ID), + zap.Duration("duration", time.Since(start)), + zap.Error(err)) + } else { + log.Info("Kafka DDL producer closed", + zap.String("namespace", p.id.Namespace), + zap.String("changefeed", p.id.ID), + zap.Duration("duration", time.Since(start))) + } } type saramaAsyncProducer struct { diff --git a/pkg/sink/kafka/sarama_factory.go b/pkg/sink/kafka/sarama_factory.go index 1e50253641..d7c1642a59 100644 --- a/pkg/sink/kafka/sarama_factory.go +++ b/pkg/sink/kafka/sarama_factory.go @@ -109,6 +109,7 @@ func (f *saramaFactory) SyncProducer(ctx context.Context) (SyncProducer, error) return nil, err } config.MetricRegistry = f.registry +<<<<<<< HEAD client, err := newSaramaClientImpl(f.option.BrokerEndpoints, config) if err != nil { @@ -120,6 +121,12 @@ func (f *saramaFactory) SyncProducer(ctx context.Context) (SyncProducer, error) return nil, errors.Trace(err) } +======= + p, err := sarama.NewSyncProducer(f.option.BrokerEndpoints, config) + if err != nil { + return nil, errors.Trace(err) + } +>>>>>>> 4c631d5951 (kafka(ticdc): ddl sink close the underline sink if send ddl or checkpoint failed and refactor the kafka ddl sink (#12112)) return &saramaSyncProducer{ id: f.changefeedID, producer: p, diff --git a/pkg/sink/kafka/v2/factory.go b/pkg/sink/kafka/v2/factory.go index 56475a21f4..bed68ab040 100644 --- a/pkg/sink/kafka/v2/factory.go +++ b/pkg/sink/kafka/v2/factory.go @@ -25,6 +25,7 @@ import ( "github.com/pingcap/log" "github.com/pingcap/tiflow/cdc/model" "github.com/pingcap/tiflow/pkg/errors" + cerror "github.com/pingcap/tiflow/pkg/errors" "github.com/pingcap/tiflow/pkg/security" "github.com/pingcap/tiflow/pkg/sink/codec/common" pkafka "github.com/pingcap/tiflow/pkg/sink/kafka" @@ -277,12 +278,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 cerror.WrapError(cerror.ErrKafkaSendMessage, err) } // SendMessages produces a given set of messages, and returns only when all @@ -299,19 +301,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 cerror.WrapError(cerror.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)), From d1beb5c6726620c2616e5f9e60f1b548ddd69e0a Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Fri, 5 Jun 2026 23:04:34 +0800 Subject: [PATCH 2/3] fix some conflicts --- cdc/sink/ddlsink/mq/mq_ddl_sink.go | 11 ++--------- cdc/sink/ddlsink/mq/pulsar_ddl_sink.go | 4 ---- pkg/sink/kafka/sarama_factory.go | 9 --------- pkg/sink/kafka/v2/factory.go | 5 ++--- 4 files changed, 4 insertions(+), 25 deletions(-) diff --git a/cdc/sink/ddlsink/mq/mq_ddl_sink.go b/cdc/sink/ddlsink/mq/mq_ddl_sink.go index 563bb0efeb..a8483f7ec3 100644 --- a/cdc/sink/ddlsink/mq/mq_ddl_sink.go +++ b/cdc/sink/ddlsink/mq/mq_ddl_sink.go @@ -141,13 +141,6 @@ func (k *DDLSink) WriteDDLEvent(ctx context.Context, ddl *model.DDLEvent) error func (k *DDLSink) WriteCheckpointTs(ctx context.Context, ts uint64, tables []*model.TableInfo, ) error { -<<<<<<< HEAD -======= - var ( - err error - partitionNum int32 - ) ->>>>>>> 4c631d5951 (kafka(ticdc): ddl sink close the underline sink if send ddl or checkpoint failed and refactor the kafka ddl sink (#12112)) msg, err := k.encoder.EncodeCheckpointEvent(ts) if err != nil { return err @@ -160,7 +153,7 @@ func (k *DDLSink) WriteCheckpointTs(ctx context.Context, // This will be compatible with the old behavior. if len(tables) == 0 { topic := k.eventRouter.GetDefaultTopic() - partitionNum, err = k.topicManager.GetPartitionNum(ctx, topic) + partitionNum, err := k.topicManager.GetPartitionNum(ctx, topic) if err != nil { return err } @@ -174,7 +167,7 @@ func (k *DDLSink) WriteCheckpointTs(ctx context.Context, } topics := k.eventRouter.GetActiveTopics(tableNames) for _, topic := range topics { - partitionNum, err = k.topicManager.GetPartitionNum(ctx, topic) + partitionNum, err := k.topicManager.GetPartitionNum(ctx, topic) if err != nil { return err } diff --git a/cdc/sink/ddlsink/mq/pulsar_ddl_sink.go b/cdc/sink/ddlsink/mq/pulsar_ddl_sink.go index e55ae1d698..96f6722f11 100644 --- a/cdc/sink/ddlsink/mq/pulsar_ddl_sink.go +++ b/cdc/sink/ddlsink/mq/pulsar_ddl_sink.go @@ -101,9 +101,5 @@ func NewPulsarDDLSink( } s := newDDLSink(changefeedID, p, nil, topicManager, eventRouter, encoderBuilder.Build(), protocol) -<<<<<<< HEAD -======= - ->>>>>>> 4c631d5951 (kafka(ticdc): ddl sink close the underline sink if send ddl or checkpoint failed and refactor the kafka ddl sink (#12112)) return s, nil } diff --git a/pkg/sink/kafka/sarama_factory.go b/pkg/sink/kafka/sarama_factory.go index d7c1642a59..a1b9ff945f 100644 --- a/pkg/sink/kafka/sarama_factory.go +++ b/pkg/sink/kafka/sarama_factory.go @@ -109,8 +109,6 @@ func (f *saramaFactory) SyncProducer(ctx context.Context) (SyncProducer, error) return nil, err } config.MetricRegistry = f.registry -<<<<<<< HEAD - client, err := newSaramaClientImpl(f.option.BrokerEndpoints, config) if err != nil { return nil, errors.Trace(err) @@ -120,13 +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) } - -======= - p, err := sarama.NewSyncProducer(f.option.BrokerEndpoints, config) - if err != nil { - return nil, errors.Trace(err) - } ->>>>>>> 4c631d5951 (kafka(ticdc): ddl sink close the underline sink if send ddl or checkpoint failed and refactor the kafka ddl sink (#12112)) return &saramaSyncProducer{ id: f.changefeedID, producer: p, diff --git a/pkg/sink/kafka/v2/factory.go b/pkg/sink/kafka/v2/factory.go index bed68ab040..3182a3c131 100644 --- a/pkg/sink/kafka/v2/factory.go +++ b/pkg/sink/kafka/v2/factory.go @@ -25,7 +25,6 @@ import ( "github.com/pingcap/log" "github.com/pingcap/tiflow/cdc/model" "github.com/pingcap/tiflow/pkg/errors" - cerror "github.com/pingcap/tiflow/pkg/errors" "github.com/pingcap/tiflow/pkg/security" "github.com/pingcap/tiflow/pkg/sink/codec/common" pkafka "github.com/pingcap/tiflow/pkg/sink/kafka" @@ -284,7 +283,7 @@ func (s *syncWriter) SendMessage( Key: message.Key, Value: message.Value, }) - return cerror.WrapError(cerror.ErrKafkaSendMessage, err) + return errors.WrapError(errors.ErrKafkaSendMessage, err) } // SendMessages produces a given set of messages, and returns only when all @@ -302,7 +301,7 @@ func (s *syncWriter) SendMessages(ctx context.Context, topic string, partitionNu } } err := s.w.WriteMessages(ctx, msgs...) - return cerror.WrapError(cerror.ErrKafkaSendMessage, err) + return errors.WrapError(errors.ErrKafkaSendMessage, err) } // Close shuts down the producer; you must call this function before a producer From 2531309d71b9e833e36ed196499627dfae899a1b Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Fri, 5 Jun 2026 23:10:49 +0800 Subject: [PATCH 3/3] revert a lot of unncessary changes --- pkg/sink/kafka/factory.go | 53 ++++++++++++++++++++++++++++----------- 1 file changed, 39 insertions(+), 14 deletions(-) diff --git a/pkg/sink/kafka/factory.go b/pkg/sink/kafka/factory.go index 1518dcdd91..12192c85af 100644 --- a/pkg/sink/kafka/factory.go +++ b/pkg/sink/kafka/factory.go @@ -116,20 +116,45 @@ func (p *saramaSyncProducer) SendMessages(_ context.Context, topic string, parti } func (p *saramaSyncProducer) Close() { - start := time.Now() - err := p.producer.Close() - if err != nil { - log.Error("Close Kafka DDL producer with error", - zap.String("namespace", p.id.Namespace), - zap.String("changefeed", p.id.ID), - zap.Duration("duration", time.Since(start)), - zap.Error(err)) - } else { - log.Info("Kafka DDL producer closed", - zap.String("namespace", p.id.Namespace), - zap.String("changefeed", p.id.ID), - zap.Duration("duration", time.Since(start))) - } + go func() { + // We need to close it asynchronously. Otherwise, we might get stuck + // with an unhealthy(i.e. Network jitter, isolation) state of Kafka. + // Factory has a background thread to fetch and update the metadata. + // If we close the client synchronously, we might get stuck. + // Safety: + // * If the kafka cluster is running well, it will be closed as soon as possible. + // * If there is a problem with the kafka cluster, + // no data will be lost because this is a synchronous client. + // * There is a risk of goroutine leakage, but it is acceptable and our main + // goal is not to get stuck with the owner tick. + start := time.Now() + if err := p.client.Close(); err != nil { + log.Warn("Close Kafka DDL client with error", + zap.String("namespace", p.id.Namespace), + zap.String("changefeed", p.id.ID), + zap.Duration("duration", time.Since(start)), + zap.Error(err)) + } else { + log.Info("Kafka DDL client closed", + zap.String("namespace", p.id.Namespace), + zap.String("changefeed", p.id.ID), + zap.Duration("duration", time.Since(start))) + } + start = time.Now() + err := p.producer.Close() + if err != nil { + log.Error("Close Kafka DDL producer with error", + zap.String("namespace", p.id.Namespace), + zap.String("changefeed", p.id.ID), + zap.Duration("duration", time.Since(start)), + zap.Error(err)) + } else { + log.Info("Kafka DDL producer closed", + zap.String("namespace", p.id.Namespace), + zap.String("changefeed", p.id.ID), + zap.Duration("duration", time.Since(start))) + } + }() } type saramaAsyncProducer struct {