cdc/sink: adjust kafka initialization logic (#3192) (#3566)

* This is an automated cherry-pick of #3192 Signed-off-by: ti-chi-bot <[email protected]> * fix conflicts. * rename tiflow to ticdc. * fix. * fix make check. Co-authored-by: Ling Jin <[email protected]> Co-authored-by: 3AceShowHand <[email protected]>
pingcap · Dec 20, 2021 · 7b2e231 · 7b2e231
1 parent 0242caf
commit 7b2e231
Show file tree

Hide file tree

Showing 6 changed files with 258 additions and 114 deletions.
diff --git a/.github/ISSUE_TEMPLATE/question.md b/.github/ISSUE_TEMPLATE/question.md
@@ -11,6 +11,6 @@ Before asking a question, make sure you have:
 
 - Searched existing Stack Overflow questions.
 - Googled your question.
-- Searched open and closed [GitHub issues](https://github.com/pingcap/ticdc/issues?q=is%3Aissue)
+- Searched open and closed [GitHub issues](https://github.com/pingcap/tiflow/issues?q=is%3Aissue)
 - Read the documentation:
     - [Detailed documentation](https://pingcap.com/docs/dev/reference/tools/ticdc/ticdc/)
diff --git a/cdc/sink/codec/interface.go b/cdc/sink/codec/interface.go
@@ -61,7 +61,7 @@ type MQMessage struct {
 	Schema    *string             // schema
 	Table     *string             // table
 	Type      model.MqMessageType // type
-	Protocol  Protocol     // protocol
+	Protocol  Protocol            // protocol
 	rowsCount int                 // rows in one MQ Message
 }
 

diff --git a/cdc/sink/mq.go b/cdc/sink/mq.go
@@ -406,7 +406,13 @@ func newKafkaSaramaSink(ctx context.Context, sinkURI *url.URL, filter *filter.Fi
 	topic := strings.TrimFunc(sinkURI.Path, func(r rune) bool {
 		return r == '/'
 	})
-	producer, err := kafka.NewKafkaSaramaProducer(ctx, sinkURI.Host, topic, config, errCh)
+	if topic == "" {
+		return nil, cerror.ErrKafkaInvalidConfig.GenWithStack("no topic is specified in sink-uri")
+	}
+
+	var protocol codec.Protocol
+	protocol.FromString(replicaConfig.Sink.Protocol)
+	producer, err := kafka.NewKafkaSaramaProducer(ctx, topic, protocol, config, errCh)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
@@ -436,8 +442,8 @@ func newPulsarSink(ctx context.Context, sinkURI *url.URL, filter *filter.Filter,
 	if s != "" {
 		opts["max-batch-size"] = s
 	}
-	// For now, it's a place holder. Avro format have to make connection to Schema Registery,
-	// and it may needs credential.
+	// For now, it's a placeholder. Avro format have to make connection to Schema Registry,
+	// and it may need credential.
 	credential := &security.Credential{}
 	sink, err := newMqSink(ctx, credential, producer, filter, replicaConfig, opts, errCh)
 	if err != nil {

diff --git a/cdc/sink/mq_test.go b/cdc/sink/mq_test.go
@@ -62,6 +62,12 @@ func (s mqSinkSuite) TestKafkaSink(c *check.C) {
 	c.Assert(err, check.IsNil)
 	opts := map[string]string{}
 	errCh := make(chan error, 1)
+
+	c.Assert(failpoint.Enable("github.com/pingcap/tiflow/cdc/sink/producer/kafka/SkipTopicAutoCreate", "return(true)"), check.IsNil)
+	defer func() {
+		_ = failpoint.Disable("github.com/pingcap/tiflow/cdc/sink/producer/kafka/SkipTopicAutoCreate")
+	}()
+
 	sink, err := newKafkaSaramaSink(ctx, sinkURI, fr, replicaConfig, opts, errCh)
 	c.Assert(err, check.IsNil)
 
@@ -159,6 +165,12 @@ func (s mqSinkSuite) TestKafkaSinkFilter(c *check.C) {
 	c.Assert(err, check.IsNil)
 	opts := map[string]string{}
 	errCh := make(chan error, 1)
+
+	c.Assert(failpoint.Enable("github.com/pingcap/tiflow/cdc/sink/producer/kafka/SkipTopicAutoCreate", "return(true)"), check.IsNil)
+	defer func() {
+		_ = failpoint.Disable("github.com/pingcap/tiflow/cdc/sink/producer/kafka/SkipTopicAutoCreate")
+	}()
+
 	sink, err := newKafkaSaramaSink(ctx, sinkURI, fr, replicaConfig, opts, errCh)
 	c.Assert(err, check.IsNil)