code: func (p *Producer) flush(records []*k.PutRecordsRequestEntry, reason string) {
  	p.Logger.WithFields(log.Fields{
  		"records": len(records),
  		"reason":  reason,
  	}).Info("flush")
  
  	out, err := p.Client.PutRecords(&k.PutRecordsInput{
  		StreamName: &p.StreamName,
  		Records:    records,
  	})
  
  	if err != nil {
  		p.Logger.WithError(err).Error("flush")
  		p.backoff(len(records))
  		p.flush(records, "error")
  		return
  	}
  
  	failed := *out.FailedRecordCount
  	if failed == 0 {
  		p.Backoff.Reset()
  		return
  	}
  
  	p.backoff(int(failed))
  	p.flush(failures(records, out.Records), "retry")
  }
docs: // flush records and retry failures if necessary.