sink: Make ddl and dml encoders share schema manager (#6100)

downstreamadapter/sink/kafka/helper.go

@@ -25,7 +25,12 @@
 	"github.com/pingcap/ticdc/pkg/config"
 	"github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/sink/codec"
+<<<<<<< HEAD
 	"github.com/pingcap/ticdc/pkg/sink/codec/common"
+=======
+	codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common"
+	"github.com/pingcap/ticdc/pkg/sink/codec/schemamanager"
+>>>>>>> fb743a814 (sink: Make ddl and dml encoders share schema manager (#6100))
 	"github.com/pingcap/ticdc/pkg/sink/kafka"
 	"github.com/pingcap/tidb/br/pkg/utils"
 )
@@ -99,12 +104,28 @@
 		return kafkaComponent, protocol, errors.Trace(err)
 	}
 
+<<<<<<< HEAD
 	kafkaComponent.encoder, err = codec.NewEventEncoder(ctx, encoderConfig)
+=======
+	var schemaM schemamanager.SchemaManager
+	if isAvroLike {
+		schemaM, err = schemamanager.NewSchemaManager(ctx, encoderConfig)
+		if err != nil {
+			return comp, protocol, err
+		}
+	}
+
+	comp.encoderGroup, err = codec.NewEncoderGroup(sinkConfig, encoderConfig, comp.claimCheck, schemaM, changefeedID)
+>>>>>>> fb743a814 (sink: Make ddl and dml encoders share schema manager (#6100))
 	if err != nil {
 		return kafkaComponent, protocol, errors.Trace(err)
 	}
 
+<<<<<<< HEAD
 	kafkaComponent.adminClient, err = kafkaComponent.factory.AdminClient(ctx)
+=======
+	comp.encoder, err = codec.NewEventEncoder(encoderConfig, comp.claimCheck, schemaM)
+>>>>>>> fb743a814 (sink: Make ddl and dml encoders share schema manager (#6100))
 	if err != nil {
 		return kafkaComponent, protocol, errors.WrapError(errors.ErrKafkaNewProducer, err)
 	}

downstreamadapter/sink/kafka/sink.go

@@ -25,7 +25,13 @@ import (
 	"github.com/pingcap/ticdc/pkg/config"
 	"github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/metrics"
+<<<<<<< HEAD
 	"github.com/pingcap/ticdc/pkg/sink/codec/common"
+=======
+	"github.com/pingcap/ticdc/pkg/sink/codec"
+	codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common"
+	"github.com/pingcap/ticdc/pkg/sink/codec/schemamanager"
+>>>>>>> fb743a814 (sink: Make ddl and dml encoders share schema manager (#6100))
 	"github.com/pingcap/ticdc/pkg/sink/kafka"
 	"github.com/pingcap/ticdc/pkg/util"
 	"github.com/pingcap/ticdc/utils/chann"
@@ -67,10 +73,88 @@ func (s *sink) SinkType() commonType.SinkType {
 	return commonType.KafkaSinkType
 }
 
+<<<<<<< HEAD
 func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url.URL, sinkConfig *config.SinkConfig) error {
 	comp, _, err := newKafkaSinkComponent(ctx, changefeedID, uri, sinkConfig)
 	defer comp.close()
 	return err
+=======
+var createKafkaFactory = func(createSaramaFactory func() (kafka.Factory, error)) (kafka.Factory, error) {
+	return createSaramaFactory()
+}
+
+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 err
+	}
+
+	topic, err := helper.GetTopic(uri)
+	if err != nil {
+		return err
+	}
+
+	options := kafka.NewOptions()
+	if err = options.Apply(changefeedID, uri, sinkConfig); err != nil {
+		return err
+	}
+	options.Topic = topic
+
+	encoderConfig, err := helper.GetEncoderConfig(
+		changefeedID, uri, protocol, sinkConfig,
+		options.MaxMessageBytes, options.MaxBatchedBytes,
+	)
+	if err != nil {
+		return err
+	}
+
+	claimCheck, err := claimcheck.New(ctx, encoderConfig.LargeMessageHandle, changefeedID)
+	if err != nil {
+		return err
+	}
+	defer claimCheck.Close()
+
+	isAvroLike := protocol == config.ProtocolAvro || protocol == config.ProtocolDebeziumAvro
+	if _, err = eventrouter.NewEventRouter(sinkConfig, topic, false, isAvroLike); err != nil {
+		return err
+	}
+
+	if _, err = columnselector.New(sinkConfig); err != nil {
+		return err
+	}
+
+	factory, err := createKafkaFactory(func() (kafka.Factory, error) {
+		return kafka.NewSaramaFactory(ctx, options, changefeedID)
+	})
+	if err != nil {
+		return err
+	}
+
+	adminClient, err := factory.AdminClient(ctx)
+	if err != nil {
+		return err
+	}
+	defer adminClient.Close()
+
+	err = topicmanager.EnsureTopic(ctx, changefeedID, topic, options.DeriveTopicConfig(), adminClient)
+	if err != nil {
+		return err
+	}
+
+	var schemaM schemamanager.SchemaManager
+	if isAvroLike {
+		schemaM, err = schemamanager.NewSchemaManager(ctx, encoderConfig)
+		if err != nil {
+			return err
+		}
+	}
+
+	_, err = codec.NewEventEncoder(encoderConfig, claimCheck, schemaM)
+	if err != nil {
+		return err
+	}
+	return nil
+>>>>>>> fb743a814 (sink: Make ddl and dml encoders share schema manager (#6100))
 }
 
 func New(