code: func (kz *Kazoo) CreateTopic(name string, partitionCount int, replicationFactor int, topicConfig map[string]string) error {
  	topic := kz.Topic(name)
  
  	// Official kafka sdk checks if topic exists, then always writes the config unconditionally
  	// but only writes the partition map if ones does not exist.
  	exists, err := topic.Exists()
  	if err != nil {
  		return err
  	} else if exists {
  		return ErrTopicExists
  	}
  
  	brokerList, err := kz.brokerIDList()
  	if err != nil {
  		return err
  	}
  
  	partitionList, err := topic.generatePartitionAssignments(brokerList, partitionCount, replicationFactor)
  	if err != nil {
  		return err
  	}
  
  	configData, err := topic.marshalConfig(topicConfig)
  	if err != nil {
  		return err
  	}
  
  	partitionData, err := topic.marshalPartitions(partitionList)
  	if err != nil {
  		return err
  	}
  
  	if err = kz.createOrUpdate(topic.configPath(), configData, false); err != nil {
  		return err
  	}
  
  	if err = kz.create(topic.metadataPath(), partitionData, false); err != nil {
  		return err
  	}
  
  	return nil
  }
docs: // CreateTopic creates a new kafka topic with the specified parameters and properties