remove producer related changes from this pr.

cdc/sink/producer/kafka/kafka.go

@@ -457,20 +457,18 @@ func topicPreProcess(topic string, protocol codec.Protocol, config *Config, sara
 
 	// 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 & craft protocol.
-	if protocol == codec.ProtocolDefault || protocol == codec.ProtocolCraft {
-		brokerMessageMaxBytes, err := getBrokerMessageMaxBytes(admin)
-		if err != nil {
-			log.Warn("TiCDC cannot find `message.max.bytes` from broker's configuration")
-			return errors.Trace(err)
-		}
+	// we have to make sure it's not greater than `max-message-bytes`
+	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)
-		}
+	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)
 	}
 
 	// topic not created yet, and user does not specify the `partition-num` in the sink uri.