code: func (p *hashProducer) Distribute(topic string, messages ...*proto.Message) (offset int64, err error) {
  	if len(messages) == 0 {
  		return 0, errors.New("no messages")
  	}
  	part, err := messageHashPartition(messages[0].Key, p.partitions)
  	if err != nil {
  		return 0, fmt.Errorf("cannot hash message: %s", err)
  	}
  	// make sure that all messages within single call are to the same destination
  	for i := 2; i < len(messages); i++ {
  		mp, err := messageHashPartition(messages[i].Key, p.partitions)
  		if err != nil {
  			return 0, fmt.Errorf("cannot hash message: %s", err)
  		}
  		if part != mp {
  			return 0, errors.New("cannot publish messages to different destinations")
  		}
  	}
  
  	return p.producer.Produce(topic, part, messages...)
  }
docs: // Distribute write messages to given kafka topic, computing partition number from
  // the message key value. Message key must be not nil and all messages written
  // within single Produce call are atomically written to the same destination.
  //
  // All messages passed within single Produce call must hash to the same
  // destination, otherwise no message is written and error is returned.