protocol(ticdc): remove useless code and setup DefaultMaxMessageBytes (…

…#4070)

cdc/sink/producer/kafka/kafka.go

@@ -61,8 +61,8 @@ type Config struct {
 func NewConfig() *Config {
 	return &Config{
 		Version: "2.4.0",
-		// MaxMessageBytes will be used to initialize producer, we set the default value (1M) identical to kafka broker.
-		MaxMessageBytes:   1 * 1024 * 1024,
+		// MaxMessageBytes will be used to initialize producer.
+		MaxMessageBytes:   config.DefaultMaxMessageBytes,
 		ReplicationFactor: 1,
 		Compression:       "none",
 		Credential:        &security.Credential{},

cdc/sink/producer/kafka/kafka_test.go

@@ -287,6 +287,13 @@ func (s *kafkaSuite) TestTopicPreProcess(c *check.C) {
 func (s *kafkaSuite) TestNewSaramaConfig(c *check.C) {
 	defer testleak.AfterTest(c)()
 	ctx := context.Background()
+	// Test default values.
+	defaultConfig := NewConfig()
+	c.Assert(defaultConfig.MaxMessageBytes, check.Equals, 10*1024*1024)
+	c.Assert(defaultConfig.ReplicationFactor, check.Equals, int16(1))
+	c.Assert(defaultConfig.Compression, check.Equals, "none")
+	c.Assert(defaultConfig.AutoCreate, check.Equals, true)
+	// Exception testing
 	config := NewConfig()
 	config.Version = "invalid"
 	_, err := newSaramaConfigImpl(ctx, config)

errors.toml

@@ -431,11 +431,6 @@ error = '''
 locate region by id
 '''
 
-["CDC:ErrMQSinkUnknownProtocol"]
-error = '''
-unknown '%s' protocol for Message Queue sink
-'''
-
 ["CDC:ErrMarshalFailed"]
 error = '''
 marshal failed

pkg/config/sink.go

@@ -16,13 +16,6 @@ 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"`

pkg/errors/errors.go

@@ -89,7 +89,6 @@ 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"))