sink(ticdc): pass client and admin client

cdc/sink/mq.go

@@ -369,7 +369,7 @@ func newKafkaSaramaSink(ctx context.Context, sinkURI *url.URL,
 		return nil, cerror.WrapError(cerror.ErrKafkaCreateTopic, err)
 	}
 
-	sProducer, err := kafka.NewKafkaSaramaProducer(ctx, topic, baseConfig, saramaConfig, errCh)
+	sProducer, err := kafka.NewKafkaSaramaProducer(ctx, client, adminClient, topic, baseConfig, saramaConfig, errCh)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}

cdc/sink/producer/kafka/kafka.go

@@ -51,7 +51,8 @@ type kafkaSaramaProducer struct {
 	// clientLock is used to protect concurrent access of asyncProducer and syncProducer.
 	// Since we don't close these two clients (which have an input chan) from the
 	// sender routine, data race or send on closed chan could happen.
-	clientLock    sync.RWMutex
+	clientLock sync.RWMutex
+	// This admin mainly used by `metricsMonitor` to fetch broker info.
 	admin         kafka.ClusterAdminClient
 	client        sarama.Client
 	asyncProducer sarama.AsyncProducer
@@ -317,23 +318,20 @@ func (k *kafkaSaramaProducer) run(ctx context.Context) error {
 var NewAdminClientImpl kafka.ClusterAdminClientCreator = kafka.NewSaramaAdminClient
 
 // NewKafkaSaramaProducer creates a kafka sarama producer
-func NewKafkaSaramaProducer(ctx context.Context, topic string, config *Config, saramaConfig *sarama.Config, errCh chan error) (*kafkaSaramaProducer, error) {
+func NewKafkaSaramaProducer(
+	ctx context.Context,
+	client sarama.Client,
+	admin kafka.ClusterAdminClient,
+	topic string,
+	config *Config,
+	saramaConfig *sarama.Config,
+	errCh chan error,
+) (*kafkaSaramaProducer, error) {
 	changefeedID := util.ChangefeedIDFromCtx(ctx)
 	role := util.RoleFromCtx(ctx)
 	log.Info("Starting kafka sarama producer ...", zap.Any("config", config),
 		zap.String("changefeed", changefeedID), zap.Any("role", role))
 
-	// this admin mainly used by `metricsMonitor` to fetch broker info.
-	admin, err := NewAdminClientImpl(config.BrokerEndpoints, saramaConfig)
-	if err != nil {
-		return nil, cerror.WrapError(cerror.ErrKafkaNewSaramaProducer, err)
-	}
-
-	client, err := sarama.NewClient(config.BrokerEndpoints, saramaConfig)
-	if err != nil {
-		return nil, cerror.WrapError(cerror.ErrKafkaNewSaramaProducer, err)
-	}
-
 	asyncProducer, err := sarama.NewAsyncProducerFromClient(client)
 	if err != nil {
 		return nil, cerror.WrapError(cerror.ErrKafkaNewSaramaProducer, err)

cdc/sink/producer/kafka/kafka_test.go

@@ -104,7 +104,11 @@ func (s *kafkaSuite) TestNewSaramaProducer(c *check.C) {
 	saramaConfig, err := NewSaramaConfig(ctx, config)
 	c.Assert(err, check.IsNil)
 	saramaConfig.Producer.Flush.MaxMessages = 1
-	producer, err := NewKafkaSaramaProducer(ctx, topic, config, saramaConfig, errCh)
+	client, err := sarama.NewClient(config.BrokerEndpoints, saramaConfig)
+	c.Assert(err, check.IsNil)
+	adminClient, err := NewAdminClientImpl(config.BrokerEndpoints, saramaConfig)
+	c.Assert(err, check.IsNil)
+	producer, err := NewKafkaSaramaProducer(ctx, client, adminClient, topic, config, saramaConfig, errCh)
 	c.Assert(err, check.IsNil)
 	c.Assert(producer.GetPartitionNum(), check.Equals, int32(2))
 
@@ -369,7 +373,11 @@ func (s *kafkaSuite) TestProducerSendMessageFailed(c *check.C) {
 	saramaConfig.Producer.Retry.Max = 2
 	saramaConfig.Producer.MaxMessageBytes = 8
 
-	producer, err := NewKafkaSaramaProducer(ctx, topic, config, saramaConfig, errCh)
+	client, err := sarama.NewClient(config.BrokerEndpoints, saramaConfig)
+	c.Assert(err, check.IsNil)
+	adminClient, err := NewAdminClientImpl(config.BrokerEndpoints, saramaConfig)
+	c.Assert(err, check.IsNil)
+	producer, err := NewKafkaSaramaProducer(ctx, client, adminClient, topic, config, saramaConfig, errCh)
 	defer func() {
 		err := producer.Close()
 		c.Assert(err, check.IsNil)
@@ -439,7 +447,11 @@ func (s *kafkaSuite) TestProducerDoubleClose(c *check.C) {
 	ctx = util.PutRoleInCtx(ctx, util.RoleTester)
 	saramaConfig, err := NewSaramaConfig(context.Background(), config)
 	c.Assert(err, check.IsNil)
-	producer, err := NewKafkaSaramaProducer(ctx, topic, config, saramaConfig, errCh)
+	client, err := sarama.NewClient(config.BrokerEndpoints, saramaConfig)
+	c.Assert(err, check.IsNil)
+	adminClient, err := NewAdminClientImpl(config.BrokerEndpoints, saramaConfig)
+	c.Assert(err, check.IsNil)
+	producer, err := NewKafkaSaramaProducer(ctx, client, adminClient, topic, config, saramaConfig, errCh)
 	defer func() {
 		err := producer.Close()
 		c.Assert(err, check.IsNil)