code: func (p *producer) Produce(topic string, partition int32, messages ...*proto.Message) (offset int64, err error) {
  	for retry := 0; retry < p.conf.RetryLimit; retry++ {
  		if retry != 0 {
  			time.Sleep(p.conf.RetryWait)
  		}
  
  		offset, err = p.produce(topic, partition, messages...)
  
  		switch err {
  		case nil:
  			for i, msg := range messages {
  				msg.Offset = int64(i) + offset
  			}
  			if retry != 0 {
  				p.conf.Logger.Debug("Produced message after retry",
  					"retry", retry,
  					"topic", topic)
  			}
  			return offset, err
  		case io.EOF, syscall.EPIPE:
  			// p.produce call is closing connection when this error shows up,
  			// but it's also returning it so that retry loop can count this
  			// case
  			// we cannot handle this error here, because there is no direct
  			// access to connection
  		default:
  			if err := p.broker.muRefreshMetadata(); err != nil {
  				p.conf.Logger.Debug("cannot refresh metadata",
  					"error", err)
  			}
  		}
  		p.conf.Logger.Debug("Cannot produce messages",
  			"retry", retry,
  			"topic", topic,
  			"error", err)
  	}
  	p.conf.Logger.Debug("Abort to produce after retrying messages",
  		"retry", p.conf.RetryLimit,
  		"topic", topic)
  	return 0, err
  
  }
docs: // Produce writes messages to the given destination. Writes within the call are
  // atomic, meaning either all or none of them are written to kafka.  Produce
  // has a configurable amount of retries which may be attempted when common
  // errors are encountered.  This behaviour can be configured with the
  // RetryLimit and RetryWait attributes.
  //
  // Upon a successful call, the message's Offset field is updated.