Repository: pingcap/tiflow (go)

Issue:
TiCDC sink initialization should check topic or broker's max message bytes configuration for all kinds of protocols.
### What did you do?

In v4.0.16, if use non `open-protocol`, such as `canal-json`, `canal`, `avro` to initialize the Kafka sink, changefeed may meet `message too large` error from the Sarama producer.

Furthermore, the TiCDC cluster was upgraded from v4.0.14 to v4.0.16, we should also consider upgrade compatibility, which means old existing changefeeds should boost and run normal.

### What did you expect to see?

changefeed works normally, no matter which kind of protocol.

### What did you see instead?

```
[CDC:ErrKafkaAsyncSendMessage]kafka: Failed to produce message to topic cdc-eps_rt_ods-ods_market_flw_lx_app_log_prod: kafka server: Message was too large, server rejected it to avoid allocation error."]
```

### Versions of the cluster

Upstream TiDB cluster version (execute `SELECT tidb_version();` in a MySQL client):

```console
(paste TiDB cluster version here)
```

TiCDC version (execute `cdc version`):

```console
v4.0.16
```

Pull request description:
sink(ticdc): set max-message-bytes default to 10m (#4036)
This is an automated cherry-pick of #4036

<!--
Thank you for contributing to TiDB-CDC! Please read MD's [CONTRIBUTING](https://github.com/pingcap/tidb-cdc/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.
-->

### What problem does this PR solve? <!--add issue link with summary if exists-->
close #4041
set Kafka producer's default `max-message-bytes` to 10M, and use the minimum value among sink-uri's `max-message-bytes`, broker's `message.max.bytes`, topic's `max.message.bytes` to initialize the producer.

### What is changed and how it works?
* set the default value of `max-message-bytes` to 10M
* use the min value of the 3 mentioned above to initialize the producer.

### Check List <!--REMOVE the items that are not applicable-->

Tests <!-- At least one of them must be included. -->

 - Unit test

Code changes

Side effects

Related changes

### Release note <!-- bugfixes or new feature need a release note -->

```release-note
Set `max-message-bytes` default to 10M, and use the min value with topic and broker to initialize the producer.
```


Patch:
diff --git a/cdc/sink/codec/json.go b/cdc/sink/codec/json.go
index f95cace1a..38f6eda35 100644
--- a/cdc/sink/codec/json.go
+++ b/cdc/sink/codec/json.go
@@ -26,6 +26,7 @@ import (
 	"github.com/pingcap/log"
 	timodel "github.com/pingcap/parser/model"
 	"github.com/pingcap/parser/mysql"
+	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"go.uber.org/zap"
 
@@ -35,8 +36,6 @@ import (
 const (
 	// BatchVersion1 represents the version of batch format
 	BatchVersion1 uint64 = 1
-	// DefaultMaxMessageBytes sets the default value for max-message-bytes
-	DefaultMaxMessageBytes int = 1 * 1024 * 1024 // 1M
 	// DefaultMaxBatchSize sets the default value for max-batch-size
 	DefaultMaxBatchSize int = 16
 )
@@ -317,13 +316,13 @@ type JSONEventBatchEncoder struct {
 	messageBuf   []*MQMessage
 	curBatchSize int
 	// configs
-	maxKafkaMessageSize int
-	maxBatchSize        int
+	maxMessageBytes int
+	maxBatchSize    int
 }
 
-// GetMaxKafkaMessageSize is only for unit testing.
-func (d *JSONEventBatchEncoder) GetMaxKafkaMessageSize() int {
-	return d.maxKafkaMessageSize
+// GetMaxMessageSize is only for unit testing.
+func (d *JSONEventBatchEncoder) GetMaxMessageSize() int {
+	return d.maxMessageBytes
 }
 
 // GetMaxBatchSize is only for unit testing.
@@ -402,15 +401,15 @@ func (d *JSONEventBatchEncoder) AppendRowChangedEvent(e *model.RowChangedEvent)
 		// for single message that longer than max-message-size, do not send it.
 		// 16 is the length of `keyLenByte` and `valueLenByte`, 8 is the length of `versionHead`
 		length := len(key) + len(value) + maximumRecordOverhead + 16 + 8
-		if length > d.maxKafkaMessageSize {
+		if length > d.maxMessageBytes {
 			log.Warn("Single message too large",
-				zap.Int("max-message-size", d.maxKafkaMessageSize), zap.Int("length", length), zap.Any("table", e.Table))
+				zap.Int("max-message-size", d.maxMessageBytes), zap.Int("length", length), zap.Any("table", e.Table))
 			return EncoderNoOperation, cerror.ErrJSONCodecRowTooLarge.GenWithStackByArgs()
 		}
 
 		if len(d.messageBuf) == 0 ||
 			d.curBatchSize >= d.maxBatchSize ||
-			d.messageBuf[len(d.messageBuf)-1].Length()+len(key)+len(value)+16 > d.maxKafkaMessageSize {
+			d.messageBuf[len(d.messageBuf)-1].Length()+len(key)+len(value)+16 > d.maxMessageBytes {
 
 			versionHead := make([]byte, 8)
 			binary.BigEndian.PutUint64(versionHead, BatchVersion1)
@@ -429,10 +428,10 @@ func (d *JSONEventBatchEncoder) AppendRowChangedEvent(e *model.RowChangedEvent)
 		message.Table = &e.Table.Table
 		message.IncRowsCount()
 
-		if message.Length() > d.maxKafkaMessageSize {
+		if message.Length() > d.maxMessageBytes {
 			// `len(d.messageBuf) == 1` is implied
 			log.Debug("Event does not fit into max-message-bytes. Adjust relevant configurations to avoid service interruptions.",
-				zap.Int("event-len", message.Length()), zap.Int("max-message-bytes", d.maxKafkaMessageSize))
+				zap.Int("event-len", message.Length()), zap.Int("max-message-bytes", d.maxMessageBytes))
 		}
 		d.curBatchSize++
 	}
@@ -550,17 +549,16 @@ func (d *JSONEventBatchEncoder) Reset() {
 // SetParams reads relevant parameters for Open Protocol
 func (d *JSONEventBatchEncoder) SetParams(params map[string]string) error {
 	var err error
+
+	d.maxMessageBytes = config.DefaultMaxMessageBytes
 	if maxMessageBytes, ok := params["max-message-bytes"]; ok {
-		d.maxKafkaMessageSize, err = strconv.Atoi(maxMessageBytes)
+		d.maxMessageBytes, err = strconv.Atoi(maxMessageBytes)
 		if err != nil {
 			return cerror.ErrKafkaInvalidConfig.Wrap(err)
 		}
-	} else {
-		d.maxKafkaMessageSize = DefaultMaxMessageBytes
 	}
-
-	if d.maxKafkaMessageSize <= 0 {
-		return cerror.ErrKafkaInvalidConfig.Wrap(errors.Errorf("invalid max-message-bytes %d", d.maxKafkaMessageSize))
+	if d.maxMessageBytes <= 0 {
+		return cerror.ErrKafkaInvalidConfig.Wrap(errors.Errorf("invalid max-message-bytes %d", d.maxMessageBytes))
 	}
 
 	if maxBatchSize, ok := params["max-batch-size"]; ok {
diff --git a/cdc/sink/mq.go b/cdc/sink/mq.go
index 8f9c0f8ff..8c58b5402 100644
--- a/cdc/sink/mq.go
+++ b/cdc/sink/mq.go
@@ -410,9 +410,7 @@ func newKafkaSaramaSink(ctx context.Context, sinkURI *url.URL, filter *filter.Fi
 		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)
+	producer, err := kafka.NewKafkaSaramaProducer(ctx, topic, config, errCh)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
diff --git a/cdc/sink/producer/kafka/kafka.go b/cdc/sink/producer/kafka/kafka.go
index 2d5ea34ae..85c576bc8 100644
--- a/cdc/sink/producer/kafka/kafka.go
+++ b/cdc/sink/producer/kafka/kafka.go
@@ -387,7 +387,7 @@ func (k *kafkaSaramaProducer) run(ctx context.Context) error {
 	}
 }
 
-func topicPreProcess(topic string, protocol codec.Protocol, config *Config, saramaConfig *sarama.Config) error {
+func topicPreProcess(topic string, config *Config, saramaConfig *sarama.Config) error {
 	// FIXME: find a way to remove this failpoint for workload the unit test
 	failpoint.Inject("SkipTopicAutoCreate", func() {
 		failpoint.Return(nil)
@@ -410,20 +410,18 @@ func topicPreProcess(topic string, protocol codec.Protocol, config *Config, sara
 	info, created := topics[topic]
 	// once we have found the topic, no matter `auto-create-topic`, make sure user input parameters are valid.
 	if created {
-		// make sure that topic's `max.message.bytes` is not less than given `max-message-bytes`
-		// else the producer will send message that too large to make topic reject, then changefeed would error.
-		// only the default `open protocol` and `craft protocol` use `max-message-bytes`, so check this for them.
-		if protocol == codec.ProtocolDefault {
-			topicMaxMessageBytes, err := getTopicMaxMessageBytes(admin, info)
-			if err != nil {
-				return cerror.WrapError(cerror.ErrKafkaNewSaramaProducer, err)
-			}
-			if topicMaxMessageBytes < config.MaxMessageBytes {
-				return cerror.ErrKafkaInvalidConfig.GenWithStack(
-					"topic already exist, and topic's max.message.bytes(%d) less than max-message-bytes(%d)."+
-						"Please make sure `max-message-bytes` not greater than topic `max.message.bytes`",
-					topicMaxMessageBytes, config.MaxMessageBytes)
-			}
+		// make sure that producer's `MaxMessageBytes` smaller than topic's `max.message.bytes`
+		topicMaxMessageBytes, err := getTopicMaxMessageBytes(admin, info)
+		if err != nil {
+			return cerror.WrapError(cerror.ErrKafkaNewSaramaProducer, err)
+		}
+
+		if topicMaxMessageBytes < config.MaxMessageBytes {
+			log.Warn("topic's `max.message.bytes` less than the user set `max-message-bytes`,"+
+				"use topic's `max.message.bytes` to initialize the Kafka producer",
+				zap.Int("max.message.bytes", topicMaxMessageBytes),
+				zap.Int("max-message-bytes", config.MaxMessageBytes))
+			saramaConfig.Producer.MaxMessageBytes = topicMaxMessageBytes
 		}
 
 		// no need to create the topic, but we would have to log user if they found enter wrong topic name later
@@ -443,22 +441,22 @@ func topicPreProcess(topic string, protocol codec.Protocol, config *Config, sara
 		return cerror.ErrKafkaInvalidConfig.GenWithStack("`auto-create-topic` is false, and topic not found")
 	}
 
-	// when try to create the topic, we don't know how to set the `max.message.bytes` for the topic.
-	// Kafka would create the topic with broker's `message.max.bytes`,
-	// we have to make sure it's not greater than `max-message-bytes` for the default open protocol.
-	if protocol == codec.ProtocolDefault {
-		brokerMessageMaxBytes, err := getBrokerMessageMaxBytes(admin)
-		if err != nil {
-			log.Warn("TiCDC cannot find `message.max.bytes` from broker's configuration")
-			return errors.Trace(err)
-		}
+	brokerMessageMaxBytes, err := getBrokerMessageMaxBytes(admin)
+	if err != nil {
+		log.Warn("TiCDC cannot find `message.max.bytes` from broker's configuration")
+		return errors.Trace(err)
+	}
 
-		if brokerMessageMaxBytes < config.MaxMessageBytes {
-			return cerror.ErrKafkaInvalidConfig.GenWithStack(
-				"broker's message.max.bytes(%d) less than max-message-bytes(%d)"+
-					"Please make sure `max-message-bytes` not greater than broker's `message.max.bytes`",
-				brokerMessageMaxBytes, config.MaxMessageBytes)
-		}
+	// when create the topic, `max.message.bytes` is decided by the broker,
+	// it would use broker's `message.max.bytes` to set topic's `max.message.bytes`.
+	// TiCDC need to make sure that the producer's `MaxMessageBytes` won't larger than
+	// broker's `message.max.bytes`.
+	if brokerMessageMaxBytes < config.MaxMessageBytes {
+		log.Warn("broker's `message.max.bytes` less than the user set `max-message-bytes`,"+
+			"use broker's `message.max.bytes` to initialize the Kafka producer",
+			zap.Int("message.max.bytes", brokerMessageMaxBytes),
+			zap.Int("max-message-bytes", config.MaxMessageBytes))
+		saramaConfig.Producer.MaxMessageBytes = brokerMessageMaxBytes
 	}
 
 	// topic not created yet, and user does not specify the `partition-num` in the sink uri.
@@ -487,14 +485,14 @@ func topicPreProcess(topic string, protocol codec.Protocol, config *Config, sara
 var newSaramaConfigImpl = newSaramaConfig
 
 // NewKafkaSaramaProducer creates a kafka sarama producer
-func NewKafkaSaramaProducer(ctx context.Context, topic string, protocol codec.Protocol, config *Config, errCh chan error) (*kafkaSaramaProducer, error) {
+func NewKafkaSaramaProducer(ctx context.Context, topic string, config *Config, errCh chan error) (*kafkaSaramaProducer, error) {
 	log.Info("Starting kafka sarama producer ...", zap.Reflect("config", config))
 	cfg, err := newSaramaConfigImpl(ctx, config)
 	if err != nil {
 		return nil, err
 	}
 
-	if err := topicPreProcess(topic, protocol, config, cfg); err != nil {
+	if err := topicPreProcess(topic, config, cfg); err != nil {
 		return nil, cerror.WrapError(cerror.ErrKafkaNewSaramaProducer, err)
 	}
 
diff --git a/errors.toml b/errors.toml
index bfdcc062a..2ec5b528a 100755
--- a/errors.toml
+++ b/errors.toml
@@ -431,6 +431,11 @@ error = '''
 locate region by id
 '''
 
+["CDC:ErrMQSinkUnknownProtocol"]
+error = '''
+unknown '%s' protocol for Message Queue sink
+'''
+
 ["CDC:ErrMarshalFailed"]
 error = '''
 marshal failed
diff --git a/pkg/config/mq_sink_protocol.go b/pkg/config/mq_sink_protocol.go
new file mode 100644
index 000000000..bac5ca088
--- /dev/null
+++ b/pkg/config/mq_sink_protocol.go
@@ -0,0 +1,85 @@
+// Copyright 2021 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package config
+
+import (
+	"strings"
+
+	cerror "github.com/pingcap/tiflow/pkg/errors"
+)
+
+const (
+	// ProtocolKey specifies the key of the protocol in the SinkURI.
+	ProtocolKey = "protocol"
+)
+
+// Protocol is the protocol of the mq message.
+type Protocol int
+
+// Enum types of the Protocol.
+const (
+	ProtocolDefault Protocol = iota
+	ProtocolCanal
+	ProtocolAvro
+	ProtocolMaxwell
+	ProtocolCanalJSON
+	ProtocolCraft
+	ProtocolOpen
+)
+
+// FromString converts the protocol from string to Protocol enum type.
+func (p *Protocol) FromString(protocol string) error {
+	switch strings.ToLower(protocol) {
+	case "default":
+		*p = ProtocolOpen
+	case "canal":
+		*p = ProtocolCanal
+	case "avro":
+		*p = ProtocolAvro
+	case "maxwell":
+		*p = ProtocolMaxwell
+	case "canal-json":
+		*p = ProtocolCanalJSON
+	case "craft":
+		*p = ProtocolCraft
+	case "open-protocol":
+		*p = ProtocolOpen
+	default:
+		return cerror.ErrMQSinkUnknownProtocol.GenWithStackByArgs(protocol)
+	}
+
+	return nil
+}
+
+// String converts the Protocol enum type string to string.
+func (p Protocol) String() string {
+	switch p {
+	case ProtocolDefault:
+		return "default"
+	case ProtocolCanal:
+		return "canal"
+	case ProtocolAvro:
+		return "avro"
+	case ProtocolMaxwell:
+		return "maxwell"
+	case ProtocolCanalJSON:
+		return "canal-json"
+	case ProtocolCraft:
+		return "craft"
+	case ProtocolOpen:
+		return "open-protocol"
+	default:
+		panic("unreachable")
+	}
+}
diff --git a/pkg/config/sink.go b/pkg/config/sink.go
index 90fff1245..722dc314c 100644
--- a/pkg/config/sink.go
+++ b/pkg/config/sink.go
@@ -13,6 +13,16 @@
 
 package config
 
+// DefaultMaxMessageBytes sets the default value for max-message-bytes
+const DefaultMaxMessageBytes = 10 * 1024 * 1024 // 10M
+
+// ForceEnableOldValueProtocols specifies which protocols need to be forced to enable old value.
+var ForceEnableOldValueProtocols = []string{
+	ProtocolCanal.String(),
+	ProtocolCanalJSON.String(),
+	ProtocolMaxwell.String(),
+}
+
 // SinkConfig represents sink config for a changefeed
 type SinkConfig struct {
 	DispatchRules []*DispatchRule `toml:"dispatchers" json:"dispatchers"`
diff --git a/pkg/errors/errors.go b/pkg/errors/errors.go
index bce326cdd..e834591ef 100644
--- a/pkg/errors/errors.go
+++ b/pkg/errors/errors.go
@@ -89,6 +89,7 @@ var (
 	ErrAsyncBroadcastNotSupport  = errors.Normalize("Async broadcasts not supported", errors.RFCCodeText("CDC:ErrAsyncBroadcastNotSupport"))
 	ErrKafkaInvalidConfig        = errors.Normalize("kafka config invalid", errors.RFCCodeText("CDC:ErrKafkaInvalidConfig"))
 	ErrSinkURIInvalid            = errors.Normalize("sink uri invalid", errors.RFCCodeText("CDC:ErrSinkURIInvalid"))
+	ErrMQSinkUnknownProtocol     = errors.Normalize("unknown '%s' protocol for Message Queue sink", errors.RFCCodeText("CDC:ErrMQSinkUnknownProtocol"))
 	ErrMySQLTxnError             = errors.Normalize("MySQL txn error", errors.RFCCodeText("CDC:ErrMySQLTxnError"))
 	ErrMySQLQueryError           = errors.Normalize("MySQL query error", errors.RFCCodeText("CDC:ErrMySQLQueryError"))
 	ErrMySQLConnectionError      = errors.Normalize("MySQL connection error", errors.RFCCodeText("CDC:ErrMySQLConnectionError"))
diff --git a/pkg/kafka/cluster_admin_client_mock_impl.go b/pkg/kafka/cluster_admin_client_mock_impl.go
new file mode 100644
index 000000000..34096313c
--- /dev/null
+++ b/pkg/kafka/cluster_admin_client_mock_impl.go
@@ -0,0 +1,97 @@
+// Copyright 2021 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package kafka
+
+import (
+	"strconv"
+
+	"github.com/Shopify/sarama"
+)
+
+const (
+	// DefaultMockTopicName specifies the default mock topic name.
+	DefaultMockTopicName = "mock_topic"
+	// defaultMockControllerID specifies the default mock controller ID.
+	defaultMockControllerID = 1
+)
+
+// defaultMaxMessageBytes specifies the default max message bytes.
+var defaultMaxMessageBytes = "10485760"
+
+// ClusterAdminClientMockImpl mock implements the admin client interface.
+type ClusterAdminClientMockImpl struct {
+	topics map[string]sarama.TopicDetail
+	// Cluster controller ID.
+	controllerID  int32
+	brokerConfigs []sarama.ConfigEntry
+}
+
+// NewClusterAdminClientMockImpl news a ClusterAdminClientMockImpl struct with default configurations.
+func NewClusterAdminClientMockImpl() *ClusterAdminClientMockImpl {
+	topics := make(map[string]sarama.TopicDetail)
+	configEntries := make(map[string]*string)
+	configEntries[TopicMaxMessageBytesConfigName] = &defaultMaxMessageBytes
+	topics[DefaultMockTopicName] = sarama.TopicDetail{
+		NumPartitions: 3,
+		ConfigEntries: configEntries,
+	}
+
+	brokerConfigs := []sarama.ConfigEntry{{
+		Name:  BrokerMessageMaxBytesConfigName,
+		Value: defaultMaxMessageBytes,
+	}}
+
+	return &ClusterAdminClientMockImpl{
+		topics:        topics,
+		controllerID:  defaultMockControllerID,
+		brokerConfigs: brokerConfigs,
+	}
+}
+
+// ListTopics returns all topics directly.
+func (c *ClusterAdminClientMockImpl) ListTopics() (map[string]sarama.TopicDetail, error) {
+	return c.topics, nil
+}
+
+// DescribeCluster returns the controller ID.
+func (c *ClusterAdminClientMockImpl) DescribeCluster() (brokers []*sarama.Broker, controllerID int32, err error) {
+	return nil, c.controllerID, nil
+}
+
+// DescribeConfig return brokerConfigs directly.
+func (c *ClusterAdminClientMockImpl) DescribeConfig(_ sarama.ConfigResource) ([]sarama.ConfigEntry, error) {
+	return c.brokerConfigs, nil
+}
+
+// CreateTopic adds topic into map.
+func (c *ClusterAdminClientMockImpl) CreateTopic(topic string, detail *sarama.TopicDetail, _ bool) error {
+	c.topics[topic] = *detail
+	return nil
+}
+
+// Close do nothing.
+func (c *ClusterAdminClientMockImpl) Close() error {
+	return nil
+}
+
+// GetDefaultMockTopicName returns the default topic name
+func (c *ClusterAdminClientMockImpl) GetDefaultMockTopicName() string {
+	return DefaultMockTopicName
+}
+
+// GetDefaultMaxMessageBytes returns defaultMaxMessageBytes as a number.
+func (c *ClusterAdminClientMockImpl) GetDefaultMaxMessageBytes() int {
+	topicMaxMessage, _ := strconv.Atoi(defaultMaxMessageBytes)
+	return topicMaxMessage
+}
diff --git a/pkg/kafka/config.go b/pkg/kafka/config.go
new file mode 100644
index 000000000..ea7133215
--- /dev/null
+++ b/pkg/kafka/config.go
@@ -0,0 +1,25 @@
+// Copyright 2021 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package kafka
+
+const (
+	// BrokerMessageMaxBytesConfigName specifies the largest record batch size allowed by
+	// Kafka brokers.
+	// See: https://kafka.apache.org/documentation/#brokerconfigs_message.max.bytes
+	BrokerMessageMaxBytesConfigName = "message.max.bytes"
+	// TopicMaxMessageBytesConfigName specifies the largest record batch size allowed by
+	// Kafka topics.
+	// See: https://kafka.apache.org/documentation/#topicconfigs_max.message.bytes
+	TopicMaxMessageBytesConfigName = "max.message.bytes"
+)


Test patch:
diff --git a/cdc/sink/codec/json_test.go b/cdc/sink/codec/json_test.go
index cdd3c128d..d59733a7e 100644
--- a/cdc/sink/codec/json_test.go
+++ b/cdc/sink/codec/json_test.go
@@ -21,6 +21,7 @@ import (
 	"github.com/pingcap/check"
 	"github.com/pingcap/parser/mysql"
 	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/pingcap/tiflow/pkg/config"
 	"github.com/pingcap/tiflow/pkg/util/testleak"
 )
 
@@ -227,7 +228,7 @@ func (s *batchSuite) TestParamsEdgeCases(c *check.C) {
 	err := encoder.SetParams(map[string]string{})
 	c.Assert(err, check.IsNil)
 	c.Assert(encoder.maxBatchSize, check.Equals, DefaultMaxBatchSize)
-	c.Assert(encoder.maxKafkaMessageSize, check.Equals, DefaultMaxMessageBytes)
+	c.Assert(encoder.maxMessageBytes, check.Equals, config.DefaultMaxMessageBytes)
 
 	err = encoder.SetParams(map[string]string{"max-message-bytes": "0"})
 	c.Assert(err, check.ErrorMatches, ".*invalid.*")
@@ -238,12 +239,12 @@ func (s *batchSuite) TestParamsEdgeCases(c *check.C) {
 	err = encoder.SetParams(map[string]string{"max-message-bytes": strconv.Itoa(math.MaxInt32)})
 	c.Assert(err, check.IsNil)
 	c.Assert(encoder.maxBatchSize, check.Equals, DefaultMaxBatchSize)
-	c.Assert(encoder.maxKafkaMessageSize, check.Equals, math.MaxInt32)
+	c.Assert(encoder.maxMessageBytes, check.Equals, math.MaxInt32)
 
 	err = encoder.SetParams(map[string]string{"max-message-bytes": strconv.Itoa(math.MaxUint32)})
 	c.Assert(err, check.IsNil)
 	c.Assert(encoder.maxBatchSize, check.Equals, DefaultMaxBatchSize)
-	c.Assert(encoder.maxKafkaMessageSize, check.Equals, math.MaxUint32)
+	c.Assert(encoder.maxMessageBytes, check.Equals, math.MaxUint32)
 
 	err = encoder.SetParams(map[string]string{"max-batch-size": "0"})
 	c.Assert(err, check.ErrorMatches, ".*invalid.*")
@@ -254,12 +255,12 @@ func (s *batchSuite) TestParamsEdgeCases(c *check.C) {
 	err = encoder.SetParams(map[string]string{"max-batch-size": strconv.Itoa(math.MaxInt32)})
 	c.Assert(err, check.IsNil)
 	c.Assert(encoder.maxBatchSize, check.Equals, math.MaxInt32)
-	c.Assert(encoder.maxKafkaMessageSize, check.Equals, DefaultMaxMessageBytes)
+	c.Assert(encoder.maxMessageBytes, check.Equals, config.DefaultMaxMessageBytes)
 
 	err = encoder.SetParams(map[string]string{"max-batch-size": strconv.Itoa(math.MaxUint32)})
 	c.Assert(err, check.IsNil)
 	c.Assert(encoder.maxBatchSize, check.Equals, math.MaxUint32)
-	c.Assert(encoder.maxKafkaMessageSize, check.Equals, DefaultMaxMessageBytes)
+	c.Assert(encoder.maxMessageBytes, check.Equals, config.DefaultMaxMessageBytes)
 }
 
 func (s *batchSuite) TestMaxMessageBytes(c *check.C) {
diff --git a/cdc/sink/mq_test.go b/cdc/sink/mq_test.go
index c2e0959d6..f8805c3ff 100644
--- a/cdc/sink/mq_test.go
+++ b/cdc/sink/mq_test.go
@@ -74,7 +74,7 @@ func (s mqSinkSuite) TestKafkaSink(c *check.C) {
 	encoder := sink.newEncoder()
 	c.Assert(encoder, check.FitsTypeOf, &codec.JSONEventBatchEncoder{})
 	c.Assert(encoder.(*codec.JSONEventBatchEncoder).GetMaxBatchSize(), check.Equals, 1)
-	c.Assert(encoder.(*codec.JSONEventBatchEncoder).GetMaxKafkaMessageSize(), check.Equals, 4194304)
+	c.Assert(encoder.(*codec.JSONEventBatchEncoder).GetMaxMessageSize(), check.Equals, 4194304)
 
 	// mock kafka broker processes 1 row changed event
 	leader.Returns(prodSuccess)
@@ -228,5 +228,5 @@ func (s mqSinkSuite) TestPulsarSinkEncoderConfig(c *check.C) {
 	encoder := sink.newEncoder()
 	c.Assert(encoder, check.FitsTypeOf, &codec.JSONEventBatchEncoder{})
 	c.Assert(encoder.(*codec.JSONEventBatchEncoder).GetMaxBatchSize(), check.Equals, 1)
-	c.Assert(encoder.(*codec.JSONEventBatchEncoder).GetMaxKafkaMessageSize(), check.Equals, 4194304)
+	c.Assert(encoder.(*codec.JSONEventBatchEncoder).GetMaxMessageSize(), check.Equals, 4194304)
 }
diff --git a/cdc/sink/producer/kafka/kafka_test.go b/cdc/sink/producer/kafka/kafka_test.go
index 87deb29bb..69c52eafc 100644
--- a/cdc/sink/producer/kafka/kafka_test.go
+++ b/cdc/sink/producer/kafka/kafka_test.go
@@ -152,7 +152,7 @@ func (s *kafkaSuite) TestSaramaProducer(c *check.C) {
 		_ = failpoint.Disable("github.com/pingcap/tiflow/cdc/sink/producer/kafka/SkipTopicAutoCreate")
 	}()
 
-	producer, err := NewKafkaSaramaProducer(ctx, topic, codec.ProtocolDefault, config, errCh)
+	producer, err := NewKafkaSaramaProducer(ctx, topic, config, errCh)
 	c.Assert(err, check.IsNil)
 	c.Assert(producer.GetPartitionNum(), check.Equals, int32(2))
 	for i := 0; i < 100; i++ {
@@ -275,11 +275,12 @@ func (s *kafkaSuite) TestTopicPreProcess(c *check.C) {
 
 	cfg, err := newSaramaConfigImpl(ctx, config)
 	c.Assert(err, check.IsNil)
+	c.Assert(cfg.Producer.MaxMessageBytes, check.Equals, config.MaxMessageBytes)
 
 	config.BrokerEndpoints = []string{""}
 	cfg.Metadata.Retry.Max = 1
 
-	err = topicPreProcess(topic, codec.ProtocolDefault, config, cfg)
+	err = topicPreProcess(topic, config, cfg)
 	c.Assert(errors.Cause(err), check.Equals, sarama.ErrOutOfBrokers)
 }
 
@@ -348,7 +349,7 @@ func (s *kafkaSuite) TestCreateProducerFailed(c *check.C) {
 	config.BrokerEndpoints = []string{"127.0.0.1:1111"}
 	topic := "topic"
 	c.Assert(failpoint.Enable("github.com/pingcap/tiflow/cdc/sink/producer/kafka/SkipTopicAutoCreate", "return(true)"), check.IsNil)
-	_, err := NewKafkaSaramaProducer(ctx, topic, codec.ProtocolDefault, config, errCh)
+	_, err := NewKafkaSaramaProducer(ctx, topic, config, errCh)
 	c.Assert(errors.Cause(err), check.ErrorMatches, "invalid version.*")
 
 	_ = failpoint.Disable("github.com/pingcap/tiflow/cdc/sink/producer/kafka/SkipTopicAutoCreate")
@@ -394,7 +395,7 @@ func (s *kafkaSuite) TestProducerSendMessageFailed(c *check.C) {
 	}()
 
 	errCh := make(chan error, 1)
-	producer, err := NewKafkaSaramaProducer(ctx, topic, codec.ProtocolDefault, config, errCh)
+	producer, err := NewKafkaSaramaProducer(ctx, topic, config, errCh)
 	defer func() {
 		_ = failpoint.Disable("github.com/pingcap/tiflow/cdc/sink/producer/kafka/SkipTopicAutoCreate")
 		err := producer.Close()
@@ -459,7 +460,7 @@ func (s *kafkaSuite) TestProducerDoubleClose(c *check.C) {
 	c.Assert(failpoint.Enable("github.com/pingcap/tiflow/cdc/sink/producer/kafka/SkipTopicAutoCreate", "return(true)"), check.IsNil)
 
 	errCh := make(chan error, 1)
-	producer, err := NewKafkaSaramaProducer(ctx, topic, codec.ProtocolDefault, config, errCh)
+	producer, err := NewKafkaSaramaProducer(ctx, topic, config, errCh)
 	defer func() {
 		err := producer.Close()
 		c.Assert(err, check.IsNil)
diff --git a/pkg/config/mq_sink_protocol_test.go b/pkg/config/mq_sink_protocol_test.go
new file mode 100644
index 000000000..f58d4d795
--- /dev/null
+++ b/pkg/config/mq_sink_protocol_test.go
@@ -0,0 +1,113 @@
+// Copyright 2021 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package config
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestFromString(t *testing.T) {
+	t.Parallel()
+	testCases := []struct {
+		protocol             string
+		expectedProtocolEnum Protocol
+		expectedErr          string
+	}{
+		{
+			protocol:    "random",
+			expectedErr: ".*unknown 'random' protocol for Message Queue sink.*",
+		},
+		{
+			protocol:             "default",
+			expectedProtocolEnum: ProtocolOpen,
+		},
+		{
+			protocol:             "canal",
+			expectedProtocolEnum: ProtocolCanal,
+		},
+		{
+			protocol:             "canal-json",
+			expectedProtocolEnum: ProtocolCanalJSON,
+		},
+		{
+			protocol:             "maxwell",
+			expectedProtocolEnum: ProtocolMaxwell,
+		},
+		{
+			protocol:             "avro",
+			expectedProtocolEnum: ProtocolAvro,
+		},
+		{
+			protocol:             "craft",
+			expectedProtocolEnum: ProtocolCraft,
+		},
+		{
+			protocol:             "open-protocol",
+			expectedProtocolEnum: ProtocolOpen,
+		},
+	}
+
+	for _, tc := range testCases {
+		var protocol Protocol
+		err := protocol.FromString(tc.protocol)
+		if tc.expectedErr != "" {
+			require.Regexp(t, tc.expectedErr, err)
+		} else {
+			require.Equal(t, tc.expectedProtocolEnum, protocol)
+		}
+	}
+}
+
+func TestString(t *testing.T) {
+	t.Parallel()
+	testCases := []struct {
+		protocolEnum     Protocol
+		expectedProtocol string
+	}{
+		{
+			protocolEnum:     ProtocolDefault,
+			expectedProtocol: "default",
+		},
+		{
+			protocolEnum:     ProtocolCanal,
+			expectedProtocol: "canal",
+		},
+		{
+			protocolEnum:     ProtocolCanalJSON,
+			expectedProtocol: "canal-json",
+		},
+		{
+			protocolEnum:     ProtocolMaxwell,
+			expectedProtocol: "maxwell",
+		},
+		{
+			protocolEnum:     ProtocolAvro,
+			expectedProtocol: "avro",
+		},
+		{
+			protocolEnum:     ProtocolCraft,
+			expectedProtocol: "craft",
+		},
+		{
+			protocolEnum:     ProtocolOpen,
+			expectedProtocol: "open-protocol",
+		},
+	}
+
+	for _, tc := range testCases {
+		require.Equal(t, tc.expectedProtocol, tc.protocolEnum.String())
+	}
+}