kafka: unify sink errors and replace failpoint tests (#5786)

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,18 +21,17 @@ 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/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
@@ -131,10 +130,11 @@ func newKafkaSinkComponentWithFactory(ctx context.Context,
 
 func newKafkaSinkComponent(
 	ctx context.Context,
-	changefeedID commonType.ChangeFeedID,
+	changefeedID common.ChangeFeedID,
 	sinkURI *url.URL,
 	sinkConfig *config.SinkConfig,
 ) (components, config.Protocol, error) {
+<<<<<<< HEAD
 	return newKafkaSinkComponentWithFactory(ctx, changefeedID, sinkURI, sinkConfig, kafka.NewSaramaFactory)
 }
 
@@ -145,4 +145,88 @@ func newKafkaSinkComponentForTest(
 	sinkConfig *config.SinkConfig,
 ) (components, config.Protocol, error) {
 	return newKafkaSinkComponentWithFactory(ctx, changefeedID, sinkURI, sinkConfig, kafka.NewMockFactory)
+=======
+	var (
+		comp components
+		err  error
+	)
+	// must release resources when error occurs.
+	defer func() {
+		if err != nil {
+			comp.close()
+		}
+	}()
+	protocol, err := helper.GetProtocol(utils.GetOrZero(sinkConfig.Protocol))
+	if err != nil {
+		return comp, config.ProtocolUnknown, err
+	}
+
+	topic, err := helper.GetTopic(sinkURI)
+	if err != nil {
+		return comp, protocol, err
+	}
+
+	options := kafka.NewOptions()
+	if err = options.Apply(changefeedID, sinkURI, sinkConfig); err != nil {
+		return comp, protocol, err
+	}
+	options.Topic = topic
+
+	comp.factory, err = kafka.NewSaramaFactory(ctx, options, changefeedID)
+	if err != nil {
+		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, err
+	}
+
+	comp.columnSelector, err = columnselector.New(sinkConfig)
+	if err != nil {
+		return comp, protocol, err
+	}
+
+	encoderConfig, err := helper.GetEncoderConfig(
+		changefeedID, sinkURI, protocol, sinkConfig,
+		options.MaxMessageBytes, options.MaxBatchedBytes,
+	)
+	if err != nil {
+		return comp, protocol, err
+	}
+
+	comp.claimCheck, err = claimcheck.New(ctx, encoderConfig.LargeMessageHandle, changefeedID)
+	if err != nil {
+		return comp, protocol, err
+	}
+
+	comp.encoderGroup, err = codec.NewEncoderGroup(ctx, sinkConfig, encoderConfig, comp.claimCheck, changefeedID)
+	if err != nil {
+		return comp, protocol, err
+	}
+
+	comp.encoder, err = codec.NewEventEncoder(ctx, encoderConfig, comp.claimCheck)
+	if err != nil {
+		return comp, protocol, err
+	}
+
+	comp.adminClient, err = comp.factory.AdminClient(ctx)
+	if err != nil {
+		return comp, protocol, err
+	}
+
+	comp.topicManager, err = topicmanager.GetTopicManagerAndTryCreateTopic(
+		ctx,
+		changefeedID,
+		topic,
+		options.DeriveTopicConfig(),
+		comp.adminClient,
+	)
+	if err != nil {
+		return comp, protocol, err
+	}
+	return comp, protocol, nil
+>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786))
 }