Use GenericRecordBuilder to handle missing default values, convert to…

… java.util.List for array type

src/main/scala/com/databricks/spark/avro/AvroOutputWriter.scala

@@ -27,7 +27,7 @@ import org.apache.hadoop.fs.Path
 
 import scala.collection.immutable.Map
 import org.apache.avro.generic.GenericData.Record
-import org.apache.avro.generic.{GenericData, GenericRecord}
+import org.apache.avro.generic.{GenericData, GenericRecord, GenericRecordBuilder}
 import org.apache.avro.{Schema, SchemaBuilder}
 import org.apache.avro.mapred.AvroKey
 import org.apache.avro.mapreduce.AvroKeyOutputFormat
@@ -40,6 +40,8 @@ import org.apache.spark.sql.execution.datasources.OutputWriter
 import org.apache.spark.sql.types._
 
 import scala.collection.JavaConversions._
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ArrayBuffer
 
 // NOTE: This class is instantiated and used on executor side only, no need to be serializable.
 private[avro] class AvroOutputWriter(
@@ -174,12 +176,14 @@ private[avro] class AvroOutputWriter(
             val sourceArray = item.asInstanceOf[Seq[Any]]
             val sourceArraySize = sourceArray.size
             val targetArray = new Array[Any](sourceArraySize)
+            // val targetArray = new util.ArrayList[Any]()
             var idx = 0
             while (idx < sourceArraySize) {
               targetArray(idx) = elementConverter(sourceArray(idx))
+              // targetArray.add(elementConverter(sourceArray(idx)))
               idx += 1
             }
-            targetArray
+            targetArray.toSeq.asJava
           }
         }
       case MapType(StringType, valueType, _) =>
@@ -260,7 +264,11 @@ private[avro] class AvroOutputWriter(
                 }
               }
             }
-            record
+            if(forceAvroSchema.isDefined) {
+              new GenericRecordBuilder(record).build()
+            } else {
+              record
+            }
           }
         }
     }