kafka: unify sink errors and replace failpoint tests

downstreamadapter/sink/eventrouter/topic/expression.go

@@ -68,15 +68,14 @@ func (e Expression) validate() error {
 		return nil
 	}
 
-	return errors.ErrKafkaInvalidTopicExpression.GenWithStackByArgs(e)
+	return errors.ErrKafkaInvalidConfig.GenWithStack("invalid topic expression: %s", e)
 }
 
 // ValidateForAvro checks whether topic pattern is {schema}_{table}, the only allowed
 func (e Expression) validateForAvro() error {
 	if ok := avroTopicNameRE.MatchString(string(e)); !ok {
-		return errors.ErrKafkaInvalidTopicExpression.GenWithStackByArgs(e,
-			"topic rule for Avro must contain {schema} and {table}",
-		)
+		return errors.ErrKafkaInvalidConfig.GenWithStack(
+			"invalid topic expression %s: topic rule for Avro must contain {schema} and {table}", e)
 	}
 
 	return nil

downstreamadapter/sink/eventrouter/topic/expression_test.go

@@ -265,11 +265,11 @@ func TestInvalidExpression(t *testing.T) {
 	topicExpr := Expression(invalidExpr)
 
 	err := topicExpr.validate()
-	require.ErrorIs(t, err, errors.ErrKafkaInvalidTopicExpression)
+	require.ErrorIs(t, err, errors.ErrKafkaInvalidConfig)
 	require.ErrorContains(t, err, invalidExpr)
 
 	err = topicExpr.validateForAvro()
-	require.ErrorIs(t, err, errors.ErrKafkaInvalidTopicExpression)
+	require.ErrorIs(t, err, errors.ErrKafkaInvalidConfig)
 	require.ErrorContains(t, err, "Avro")
 	require.ErrorContains(t, err, invalidExpr)
 }

downstreamadapter/sink/kafka/helper.go

@@ -21,19 +21,18 @@ import (
 	"github.com/pingcap/ticdc/downstreamadapter/sink/eventrouter"
 	"github.com/pingcap/ticdc/downstreamadapter/sink/helper"
 	"github.com/pingcap/ticdc/downstreamadapter/sink/topicmanager"
-	commonType "github.com/pingcap/ticdc/pkg/common"
+	"github.com/pingcap/ticdc/pkg/common"
 	"github.com/pingcap/ticdc/pkg/config"
-	"github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/sink/codec"
-	"github.com/pingcap/ticdc/pkg/sink/codec/common"
+	codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common"
 	"github.com/pingcap/ticdc/pkg/sink/kafka"
 	"github.com/pingcap/ticdc/pkg/sink/kafka/claimcheck"
 	"github.com/pingcap/tidb/br/pkg/utils"
 )
 
 type components struct {
 	encoderGroup   codec.EncoderGroup
-	encoder        common.EventEncoder
+	encoder        codecCommon.EventEncoder
 	columnSelector *columnselector.ColumnSelectors
 	eventRouter    *eventrouter.EventRouter
 	topicManager   topicmanager.TopicManager
@@ -56,7 +55,7 @@ func (c components) close() {
 
 func newKafkaSinkComponent(
 	ctx context.Context,
-	changefeedID commonType.ChangeFeedID,
+	changefeedID common.ChangeFeedID,
 	sinkURI *url.URL,
 	sinkConfig *config.SinkConfig,
 ) (components, config.Protocol, error) {
@@ -72,63 +71,63 @@ func newKafkaSinkComponent(
 	}()
 	protocol, err := helper.GetProtocol(utils.GetOrZero(sinkConfig.Protocol))
 	if err != nil {
-		return comp, config.ProtocolUnknown, errors.Trace(err)
+		return comp, config.ProtocolUnknown, err
 	}
 
 	topic, err := helper.GetTopic(sinkURI)
 	if err != nil {
-		return comp, protocol, errors.Trace(err)
+		return comp, protocol, err
 	}
 
 	options := kafka.NewOptions()
 	if err = options.Apply(changefeedID, sinkURI, sinkConfig); err != nil {
-		return comp, protocol, errors.WrapError(errors.ErrKafkaInvalidConfig, err)
+		return comp, protocol, err
 	}
 	options.Topic = topic
 
 	comp.factory, err = kafka.NewSaramaFactory(ctx, options, changefeedID)
 	if err != nil {
-		return comp, protocol, errors.WrapError(errors.ErrKafkaNewProducer, err)
+		return comp, protocol, err
 	}
 
 	isAvroLike := protocol == config.ProtocolAvro || protocol == config.ProtocolDebeziumAvro
 	comp.eventRouter, err = eventrouter.NewEventRouter(
 		sinkConfig, topic, false, isAvroLike)
 	if err != nil {
-		return comp, protocol, errors.Trace(err)
+		return comp, protocol, err
 	}
 
 	comp.columnSelector, err = columnselector.New(sinkConfig)
 	if err != nil {
-		return comp, protocol, errors.Trace(err)
+		return comp, protocol, err
 	}
 
 	encoderConfig, err := helper.GetEncoderConfig(
 		changefeedID, sinkURI, protocol, sinkConfig,
 		options.MaxMessageBytes, options.MaxBatchedBytes,
 	)
 	if err != nil {
-		return comp, protocol, errors.Trace(err)
+		return comp, protocol, err
 	}
 
 	comp.claimCheck, err = claimcheck.New(ctx, encoderConfig.LargeMessageHandle, changefeedID)
 	if err != nil {
-		return comp, protocol, errors.Trace(err)
+		return comp, protocol, err
 	}
 
 	comp.encoderGroup, err = codec.NewEncoderGroup(ctx, sinkConfig, encoderConfig, comp.claimCheck, changefeedID)
 	if err != nil {
-		return comp, protocol, errors.Trace(err)
+		return comp, protocol, err
 	}
 
 	comp.encoder, err = codec.NewEventEncoder(ctx, encoderConfig, comp.claimCheck)
 	if err != nil {
-		return comp, protocol, errors.Trace(err)
+		return comp, protocol, err
 	}
 
 	comp.adminClient, err = comp.factory.AdminClient(ctx)
 	if err != nil {
-		return comp, protocol, errors.WrapError(errors.ErrKafkaNewProducer, err)
+		return comp, protocol, err
 	}
 
 	comp.topicManager, err = topicmanager.GetTopicManagerAndTryCreateTopic(
@@ -139,7 +138,7 @@ func newKafkaSinkComponent(
 		comp.adminClient,
 	)
 	if err != nil {
-		return comp, protocol, errors.Trace(err)
+		return comp, protocol, err
 	}
 	return comp, protocol, nil
 }

downstreamadapter/sink/kafka/sink.go

@@ -22,13 +22,13 @@ import (
 	"github.com/pingcap/ticdc/downstreamadapter/sink/columnselector"
 	"github.com/pingcap/ticdc/downstreamadapter/sink/eventrouter"
 	"github.com/pingcap/ticdc/downstreamadapter/sink/helper"
-	commonType "github.com/pingcap/ticdc/pkg/common"
+	"github.com/pingcap/ticdc/pkg/common"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/pingcap/ticdc/pkg/config"
 	"github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/metrics"
 	"github.com/pingcap/ticdc/pkg/sink/codec"
-	"github.com/pingcap/ticdc/pkg/sink/codec/common"
+	codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common"
 	"github.com/pingcap/ticdc/pkg/sink/kafka"
 	"github.com/pingcap/ticdc/pkg/sink/kafka/claimcheck"
 	"github.com/pingcap/ticdc/pkg/util"
@@ -44,7 +44,7 @@ const (
 )
 
 type sink struct {
-	changefeedID commonType.ChangeFeedID
+	changefeedID common.ChangeFeedID
 
 	dmlProducer      kafka.AsyncProducer
 	ddlProducer      kafka.SyncProducer
@@ -67,24 +67,24 @@ type sink struct {
 	ctx      context.Context
 }
 
-func (s *sink) SinkType() commonType.SinkType {
-	return commonType.KafkaSinkType
+func (s *sink) SinkType() common.SinkType {
+	return common.KafkaSinkType
 }
 
-func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url.URL, sinkConfig *config.SinkConfig) error {
+func Verify(ctx context.Context, changefeedID common.ChangeFeedID, uri *url.URL, sinkConfig *config.SinkConfig) error {
 	protocol, err := helper.GetProtocol(util.GetOrZero(sinkConfig.Protocol))
 	if err != nil {
-		return errors.Trace(err)
+		return err
 	}
 
 	topic, err := helper.GetTopic(uri)
 	if err != nil {
-		return errors.Trace(err)
+		return err
 	}
 
 	options := kafka.NewOptions()
 	if err = options.Apply(changefeedID, uri, sinkConfig); err != nil {
-		return errors.WrapError(errors.ErrKafkaInvalidConfig, err)
+		return err
 	}
 	options.Topic = topic
 
@@ -93,7 +93,7 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url.
 		options.MaxMessageBytes, options.MaxBatchedBytes,
 	)
 	if err != nil {
-		return errors.Trace(err)
+		return err
 	}
 
 	claimCheck, err := claimcheck.New(ctx, encoderConfig.LargeMessageHandle, changefeedID)
@@ -104,27 +104,27 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url.
 
 	isAvroLike := protocol == config.ProtocolAvro || protocol == config.ProtocolDebeziumAvro
 	if _, err = eventrouter.NewEventRouter(sinkConfig, topic, false, isAvroLike); err != nil {
-		return errors.Trace(err)
+		return err
 	}
 
 	if _, err = columnselector.New(sinkConfig); err != nil {
-		return errors.Trace(err)
+		return err
 	}
 
 	factory, err := kafka.NewSaramaFactory(ctx, options, changefeedID)
 	if err != nil {
-		return errors.WrapError(errors.ErrKafkaNewProducer, err)
+		return err
 	}
 
 	adminClient, err := factory.AdminClient(ctx)
 	if err != nil {
-		return errors.WrapError(errors.ErrKafkaNewProducer, err)
+		return err
 	}
 	defer adminClient.Close()
 
 	topics, err := adminClient.GetTopicsMeta([]string{topic}, false)
 	if err != nil {
-		return errors.Trace(err)
+		return err
 	}
 	if _, exists := topics[topic]; !exists {
 		topicConfig := options.DeriveTopicConfig()
@@ -142,30 +142,30 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url.
 			ReplicationFactor: topicConfig.ReplicationFactor,
 		}, true)
 		if err != nil {
-			return errors.WrapError(errors.ErrKafkaCreateTopic, err)
+			return err
 		}
 	}
 
 	_, err = codec.NewEventEncoder(ctx, encoderConfig, claimCheck)
 	if err != nil {
-		return errors.Trace(err)
+		return err
 	}
 	return nil
 }
 
 func New(
-	ctx context.Context, changefeedID commonType.ChangeFeedID, sinkURI *url.URL, sinkConfig *config.SinkConfig, keyspaceID uint32,
+	ctx context.Context, changefeedID common.ChangeFeedID, sinkURI *url.URL, sinkConfig *config.SinkConfig, keyspaceID uint32,
 ) (*sink, error) {
 	comp, protocol, err := newKafkaSinkComponent(ctx, changefeedID, sinkURI, sinkConfig)
 	if err != nil {
-		return nil, errors.Trace(err)
+		return nil, err
 	}
 	return newWithComponents(ctx, changefeedID, keyspaceID, protocol, comp)
 }
 
 func newWithComponents(
 	ctx context.Context,
-	changefeedID commonType.ChangeFeedID,
+	changefeedID common.ChangeFeedID,
 	keyspaceID uint32,
 	protocol config.Protocol,
 	comp components,
@@ -236,7 +236,7 @@ func (s *sink) Run(ctx context.Context) error {
 	})
 	err := g.Wait()
 	s.isNormal.Store(false)
-	return errors.Trace(err)
+	return err
 }
 
 func (s *sink) IsNormal() bool {
@@ -307,7 +307,7 @@ func (s *sink) calculateKeyPartitions(ctx context.Context) error {
 	for {
 		select {
 		case <-ctx.Done():
-			return errors.Trace(ctx.Err())
+			return context.Cause(ctx)
 		default:
 			event, ok := s.eventChan.Get()
 			if !ok {
@@ -328,7 +328,7 @@ func (s *sink) calculateKeyPartitions(ctx context.Context) error {
 			selector := s.comp.columnSelector.GetForTableInfo(event.TableInfo)
 			events, err := helper.NewMQRowEvents(event, topic, partitionNum, partitionGenerator, selector)
 			if err != nil {
-				return errors.Trace(err)
+				return err
 			}
 			s.rowChan.Push(events...)
 		}
@@ -339,7 +339,7 @@ func (s *sink) nonBatchEncodeRun(ctx context.Context) error {
 	for {
 		select {
 		case <-ctx.Done():
-			return errors.Trace(ctx.Err())
+			return context.Cause(ctx)
 		default:
 			event, ok := s.rowChan.Get()
 			if !ok {
@@ -432,7 +432,7 @@ func (s *sink) sendMessages(ctx context.Context) error {
 	for {
 		select {
 		case <-ctx.Done():
-			return errors.Trace(ctx.Err())
+			return context.Cause(ctx)
 		case future, ok := <-outCh:
 			if !ok {
 				log.Info("kafka sink encoder's output channel closed",
@@ -481,7 +481,7 @@ func (s *sink) sendDDLEvent(event *commonEvent.DDLEvent) error {
 				zap.Stringer("changefeed", s.changefeedID))
 			continue
 		}
-		common.SetDDLMessageLogInfo(message, e)
+		codecCommon.SetDDLMessageLogInfo(message, e)
 		topic := s.comp.eventRouter.GetTopicForDDL(e)
 		// Notice: We must call GetPartitionNum here,
 		// which will be responsible for automatically creating topics when they don't exist.
@@ -531,14 +531,14 @@ func (s *sink) sendCheckpoint(ctx context.Context) error {
 	}()
 
 	var (
-		msg          *common.Message
+		msg          *codecCommon.Message
 		partitionNum int32
 		err          error
 	)
 	for {
 		select {
 		case <-ctx.Done():
-			return errors.Trace(ctx.Err())
+			return context.Cause(ctx)
 		case ts, ok := <-s.checkpointChan:
 			if !ok {
 				log.Warn("kafka sink checkpoint channel closed",
@@ -555,7 +555,7 @@ func (s *sink) sendCheckpoint(ctx context.Context) error {
 			if msg == nil {
 				continue
 			}
-			common.SetCheckpointMessageLogInfo(msg, ts)
+			codecCommon.SetCheckpointMessageLogInfo(msg, ts)
 
 			tableNames := s.getAllTableNames(ts)
 			// NOTICE: When there are no tables to replicate,