Merge pull request #63 from AbsaOSS/fix/schemas-evolution

felipemmelo · web-flow · commit f5bd1571c712 · 2019-10-01T10:32:55.000+02:00
Fixing schema evolution for Confluent-compliant records.
diff --git a/pom.xml b/pom.xml
@@ -11,7 +11,7 @@
 	<modelVersion>4.0.0</modelVersion>
 	<groupId>za.co.absa</groupId>
 	<artifactId>abris_2.11</artifactId>
-	<version>2.2.3</version>
+	<version>2.2.4</version>
 	<name>abris</name>
 	<description>Provides seamless integration between Avro and Spark Structured APIs.</description>
 	<url>https://github.com/AbsaOSS/ABRiS</url>
diff --git a/src/main/scala/za/co/absa/abris/avro/AvroSerDe.scala b/src/main/scala/za/co/absa/abris/avro/AvroSerDe.scala
@@ -198,7 +198,7 @@ object AvroSerDe {
 
       if (schemaId.isEmpty) {
         throw new InvalidParameterException(s"Schema could not be registered for topic '$topic'. Make sure that the Schema Registry " +
-            s"is available, the parameters are correct and the schemas ar compatible")
+            s"is available, the parameters are correct and the schemas are compatible")
       }
       else {
         logger.info(s"Schema successfully registered for topic '$topic' with id '{${schemaId.get}}'.")
diff --git a/src/main/scala/za/co/absa/abris/avro/read/confluent/ScalaConfluentKafkaAvroDeserializer.scala b/src/main/scala/za/co/absa/abris/avro/read/confluent/ScalaConfluentKafkaAvroDeserializer.scala
@@ -31,14 +31,7 @@ import za.co.absa.abris.avro.read.ScalaDatumReader
   *
   * Please, invest some time in understanding how it works and above all, read the documentation for the method 'deserialize()'.
   */
-class ScalaConfluentKafkaAvroDeserializer(val topic: Option[String], val readerSchema: Option[Schema]) {
-
-  if (topic.isEmpty && readerSchema.isEmpty) {
-    throw new IllegalArgumentException("Neither topic nor reader Schema were informed. If you want a specific schema to" +
-      " be used for reading pass it as the readerSchema value. Otherwise, if you'd like the schema to be retrieved from" +
-      " SchemaRegistry, pass in the topic being consume and inform the SchemaRegistry URLs by calling " +
-      " 'configure' in this object using SchemaManager.PARAM_SCHEMA_REGISTRY_URL as the key in the map.")
-  }
+class ScalaConfluentKafkaAvroDeserializer(val readerSchema: Schema) {
 
   private val decoderFactory = DecoderFactory.get()
   private val idSchemaReader = scala.collection.mutable.Map[Int,ScalaDatumReader[ScalaAvroRecord]]()
@@ -62,9 +55,8 @@ class ScalaConfluentKafkaAvroDeserializer(val topic: Option[String], val readerS
     * 1. This uses the [[ScalaDatumReader]] to parse the bytes.
     * 2. This takes into account Confluent's specific metadata included in the payload (e.g. schema id), thus, it will
     *    not work on regular binary Avro records.
-    * 3. If there is a topic defined in the constructor and access to Schema Registry is configured, the schema retrieved
-    *    from the later will be considered the writer schema, otherwise, the reader schema passed to the constructor will
-    *    be used as both, reader and writer (thus notice that either, topic or reader schema must be informed).
+    * 3. The schema retrieved from Schema Registry will be considered the writer schema, otherwise, the reader schema
+    *    passed to the constructor will be used as both, reader and writer.
     * 4. The Avro DatumReader is cached based on the schema id, thus, if a new id is received as part of the payload, a new
     *    DatumReader will be created for that id, with a new schema being retrieved, iff the topic is informed and Schema
     *    Registry is configured.
@@ -89,7 +81,7 @@ class ScalaConfluentKafkaAvroDeserializer(val topic: Option[String], val readerS
       val buffer = getByteBuffer(payload)
 
       schemaId = buffer.getInt()
-      val writerSchema = getWriterSchema(topic, schemaId)
+      val writerSchema = getWriterSchema(schemaId)
 
       val length = buffer.limit() - 1 - ConfluentConstants.SCHEMA_ID_SIZE_BYTES
       val start = buffer.position() + buffer.arrayOffset()
@@ -107,12 +99,12 @@ class ScalaConfluentKafkaAvroDeserializer(val topic: Option[String], val readerS
     * If there is a topic defined and the Schema Registry has been configured, the writer schema will be retrieved from
     * Schema Registry, otherwise, the reader schema passed on to the constructor will also the considered the writer's.
     */
-  private def getWriterSchema(topic: Option[String], id: Int): Schema = {
-    if (topic.isDefined && SchemaManager.isSchemaRegistryConfigured) {
+  private def getWriterSchema(id: Int): Schema = {
+    if (SchemaManager.isSchemaRegistryConfigured) {
       SchemaManager.getById(id).get
     }
     else {
-      readerSchema.get
+      readerSchema
     }
   }
 
@@ -136,7 +128,7 @@ class ScalaConfluentKafkaAvroDeserializer(val topic: Option[String], val readerS
     * the documentation of [[ScalaConfluentKafkaAvroDeserializer.deserialize()]] to understand the implications of schema
     * changes.
     */
-  private def getDatumReader(writerSchema: Schema, readerSchema: Option[Schema], id: Int): ScalaDatumReader[ScalaAvroRecord] = {
+  private def getDatumReader(writerSchema: Schema, readerSchema: Schema, id: Int): ScalaDatumReader[ScalaAvroRecord] = {
     idSchemaReader.getOrElseUpdate(id, createDatumReader(writerSchema, readerSchema))
   }
 
@@ -145,8 +137,7 @@ class ScalaConfluentKafkaAvroDeserializer(val topic: Option[String], val readerS
     *
     * If the reader schema passed on to the constructor is undefined, the writer schema is also considered the reader one.
     */
-  private def createDatumReader(writerSchema: Schema, readerSchema: Option[Schema]): ScalaDatumReader[ScalaAvroRecord] = {
-    new ScalaDatumReader[ScalaAvroRecord](writerSchema,
-      if (readerSchema.isDefined) readerSchema.get else writerSchema)
+  private def createDatumReader(writerSchema: Schema, readerSchema: Schema): ScalaDatumReader[ScalaAvroRecord] = {
+    new ScalaDatumReader[ScalaAvroRecord](writerSchema, readerSchema)
   }
-}
+}
diff --git a/src/main/scala/za/co/absa/abris/avro/serde/AvroReaderFactory.scala b/src/main/scala/za/co/absa/abris/avro/serde/AvroReaderFactory.scala
@@ -48,22 +48,30 @@ private[avro] object AvroReaderFactory {
     * the parameters to do it are defined.
     */
   def createConfiguredConfluentAvroReader(schemaPath: Option[String], schemaRegistryConf: Option[Map[String,String]]): ScalaConfluentKafkaAvroDeserializer = {
-    val schema = if (schemaPath.isDefined) Some(AvroSchemaUtils.load(schemaPath.get)) else None
+    val schema = resolveSchema(schemaPath, schemaRegistryConf)
     val configs = if (schemaRegistryConf.isDefined) schemaRegistryConf.get else Map[String,String]()
-    val topic = if (configs.contains(SchemaManager.PARAM_SCHEMA_REGISTRY_TOPIC)) Some(configs(SchemaManager.PARAM_SCHEMA_REGISTRY_TOPIC)) else None
 
-    val reader = new ScalaConfluentKafkaAvroDeserializer(topic, schema)
-    reader.configureSchemaRegistry(configs)
-    reader
+    createConfiguredConfluentAvroReader(schema, configs)
   }
 
   /**
     * Creates an instance of [[ScalaConfluentKafkaAvroDeserializer]] and configures its Schema Registry access in case
     * the parameters to do it are defined.
     */
   def createConfiguredConfluentAvroReader(schema: Schema, schemaRegistryConf: Map[String,String]): ScalaConfluentKafkaAvroDeserializer = {
-    val reader = new ScalaConfluentKafkaAvroDeserializer(None, Some(schema))
+    val reader = new ScalaConfluentKafkaAvroDeserializer(schema)
     reader.configureSchemaRegistry(schemaRegistryConf)
     reader
   }
+
+  private def resolveSchema(schemaPath: Option[String], schemaRegistryConf: Option[Map[String,String]]): Schema = {
+    if (schemaPath.isEmpty && schemaRegistryConf.isEmpty) {
+      throw new IllegalArgumentException("Schema could not be resolved: neither path nor Schema Registry configuration provided.")
+    }
+
+    schemaPath match {
+      case Some(path) => AvroSchemaUtils.load(path)
+      case None => AvroSchemaUtils.load(schemaRegistryConf.get)
+    }
+  }
 }
diff --git a/src/main/scala/za/co/absa/abris/examples/ConfluentKafkaAvroReader.scala b/src/main/scala/za/co/absa/abris/examples/ConfluentKafkaAvroReader.scala
@@ -47,6 +47,7 @@ object ConfluentKafkaAvroReader {
     val stream = spark
       .readStream
       .format("kafka")
+      .option("startingOffsets", "earliest")
       .addOptions(properties) // 1. this method will add the properties starting with "option."; 2. security options can be set in the properties file
 
     val deserialized = configureExample(stream, properties)
@@ -65,7 +66,7 @@ object ConfluentKafkaAvroReader {
   private def configureExample(stream: DataStreamReader,props: Properties): Dataset[Row] = {
     import za.co.absa.abris.avro.AvroSerDe._
     if (props.getProperty(PARAM_EXAMPLE_SHOULD_USE_SCHEMA_REGISTRY).toBoolean) {
-      stream.fromConfluentAvro("value", None, Some(props.getSchemaRegistryConfigurations(PARAM_OPTION_SUBSCRIBE)))(RETAIN_ORIGINAL_SCHEMA)
+      stream.fromConfluentAvro("value", None, Some(props.getSchemaRegistryConfigurations(PARAM_OPTION_SUBSCRIBE)))(RETAIN_SELECTED_COLUMN_ONLY)
     }
     else {
       stream.fromConfluentAvro("value", Some(props.getProperty(PARAM_PAYLOAD_AVRO_SCHEMA)), None)(RETAIN_SELECTED_COLUMN_ONLY)
diff --git a/src/main/scala/za/co/absa/abris/examples/data/generation/ComplexRecordsGenerator.scala b/src/main/scala/za/co/absa/abris/examples/data/generation/ComplexRecordsGenerator.scala
@@ -41,7 +41,7 @@ object ComplexRecordsGenerator {
   private val avroParser = new AvroToSparkParser()
   private val random = new Random()
 
-  def usedAvroSchema = plainSchema
+  def usedAvroSchema: String = plainSchema
 
   def generateRecords(howMany: Int): List[GenericRecord] = {
     val result = new Array[GenericRecord](howMany)
@@ -123,7 +123,7 @@ object ComplexRecordsGenerator {
   }
   
   private def randomSeqOfLongs(listSize: Int) = {
-    randomListOfLongs(listSize).asScala.toSeq
+    randomListOfLongs(listSize).asScala
   }
   
   private def randomListOfStrings(listSize: Int, stringLength: Int) = {
@@ -135,7 +135,7 @@ object ComplexRecordsGenerator {
   }
 
   private def randomSeqOfStrings(listSize: Int, stringLength: Int) = {
-    randomListOfStrings(listSize, stringLength).asScala.toSeq    
+    randomListOfStrings(listSize, stringLength).asScala
   }  
   
   private def randomString(length: Int): String = {
@@ -144,7 +144,7 @@ object ComplexRecordsGenerator {
   }
 
   private def recordToBean(record: GenericRecord): Bean = {    
-    new Bean(
+    Bean(
       record.get("bytes").toString().getBytes(),
       record.get("string").asInstanceOf[String],
       record.get("int").asInstanceOf[Int],
@@ -156,4 +156,4 @@ object ComplexRecordsGenerator {
       record.get("fixed").toString().getBytes,
       record.get("map").asInstanceOf[Map[String, java.util.ArrayList[Long]]])
   }
-}
+}
diff --git a/src/main/scala/za/co/absa/abris/examples/data/generation/TestSchemas.scala b/src/main/scala/za/co/absa/abris/examples/data/generation/TestSchemas.scala
@@ -57,21 +57,22 @@ object TestSchemas {
   }"""
 
   val NATIVE_COMPLETE_SCHEMA = """{
-     "namespace": "all-types.test",
-     "type": "record",
-     "name": "native_complete",
-     "fields":[                  
-         {"name": "bytes", "type": "bytes" },
-         { "name": "string",      "type": ["string", "null"] },     
-         { "name": "int",         "type": ["int",    "null"] },
-         { "name": "long",        "type": ["long",   "null"] },
- 		     { "name": "double",      "type": ["double", "null"] },
- 		     { "name": "float",       "type": ["float",  "null"] },
- 		     { "name": "boolean",     "type": ["boolean","null"] }, 		      		     
- 		     { "name": "array", "type": {"type": "array", "items": "string"} }, 		     
- 		     {"name": "map", "type": { "type": "map", "values": {"type": "array", "items": "long"}}},
- 		     {"name": "fixed",  "type": {"type": "fixed", "size": 13, "name": "fixed"}}
-     ]
+      "namespace": "all-types.test",
+      "type": "record",
+      "name": "native_complete",
+      "fields":
+      [
+          {"name": "bytes", "type": "bytes" },
+          { "name": "string",      "type": ["string", "null"] , "doc":"a simple doc", "default": "default"},
+          { "name": "int",         "type": ["int",    "null"] },
+          { "name": "long",        "type": ["long",   "null"] },
+          { "name": "double",      "type": ["double", "null"] },
+  		    { "name": "float",       "type": ["float",  "null"] },
+  		    { "name": "boolean",     "type": ["boolean","null"] },
+  		    { "name": "array", "type": {"type": "array", "items": "string"} },
+  		    {"name": "map", "type": { "type": "map", "values": {"type": "array", "items": "long"}}},
+  		    {"name": "fixed",  "type": {"type": "fixed", "size": 40, "name": "fixed"}}
+      ]
   }"""
 
   val NATIVE_SCHEMA_SPEC = """{
diff --git a/src/test/resources/AvroReadingExample.properties b/src/test/resources/AvroReadingExample.properties
@@ -30,7 +30,7 @@ key.schema.id=latest
 example.should.use.schema.registry=true
 
 key.schema.naming.strategy=record.name
-value.schema.naming.strategy=topic.record.name
+value.schema.naming.strategy=topic.name
 
 schema.name=RecordName
 schema.namespace=RecordNamespace
diff --git a/src/test/resources/DataframeWritingExample.properties b/src/test/resources/DataframeWritingExample.properties
@@ -5,13 +5,13 @@ job.master=local[2]
 key.avro.schema=src/test/resources/example_key_schema.avsc
 payload.avro.schema=src/test/resources/example_payload_schema.avsc
 
-avro.record.name=RecordName
+avro.record.name=native_complete
 
-avro.record.namespace=RecordNamespace
+avro.record.namespace=all-types.test
 
 log.level=INFO
 
-test.data.entries=5
+test.data.entries=1
 
 parquet.data.source=some_parquet_file
 
@@ -21,7 +21,7 @@ parquet.data.source.column=errCol
 num.partitions=1
 
 # if true, schema will be inferred from Dataframe, otherwise the value set to avro.schema will be used
-infer.schema=true
+infer.schema=false
 
 # if true, the application will keep running in a loop, otherwise it will execute just once
 execution.repeat=false
@@ -37,7 +37,7 @@ option.topic=test_topic
 
 key.schema.naming.strategy=record.name
 
-value.schema.naming.strategy=topic.record.name
+value.schema.naming.strategy=topic.name
 
 # security options (comment in case the Kafka cluster is not secured)
 #option.kafka.security.protocol=SSL
diff --git a/src/test/resources/example_payload_schema.avsc b/src/test/resources/example_payload_schema.avsc
@@ -4,15 +4,15 @@
      "name": "native_complete",
      "fields":[                  
          {"name": "bytes", "type": "bytes" },
-         { "name": "string",      "type": ["string", "null"] , "doc":"a simple doc"},             
+         { "name": "string",      "type": ["string", "null"] , "doc":"a simple doc", "default": "default"},
          { "name": "int",         "type": ["int",    "null"] },
          { "name": "long",        "type": ["long",   "null"] },
- 		     { "name": "double",      "type": ["double", "null"] },
- 		     { "name": "float",       "type": ["float",  "null"] },
- 		     { "name": "boolean",     "type": ["boolean","null"] }, 		      		     
- 		     { "name": "array", "type": {"type": "array", "items": "string"} }, 		     
- 		     {"name": "map", "type": { "type": "map", "values": {"type": "array", "items": "long"}}},
- 		     {"name": "fixed",  "type": {"type": "fixed", "size": 40, "name": "fixed"}}
+ 		 { "name": "double",      "type": ["double", "null"] },
+ 		 { "name": "float",       "type": ["float",  "null"] },
+ 		 { "name": "boolean",     "type": ["boolean","null"] },
+ 		 { "name": "array", "type": {"type": "array", "items": "string"} },
+ 		 {"name": "map", "type": { "type": "map", "values": {"type": "array", "items": "long"}}},
+ 		 {"name": "fixed",  "type": {"type": "fixed", "size": 40, "name": "fixed"}}
      ]
   }
   
diff --git a/src/test/scala/za/co/absa/abris/avro/read/confluent/ScalaConfluentKafkaAvroDeserializerSpec.scala b/src/test/scala/za/co/absa/abris/avro/read/confluent/ScalaConfluentKafkaAvroDeserializerSpec.scala
@@ -53,31 +53,28 @@ class ScalaConfluentKafkaAvroDeserializerSpec extends FlatSpec with BeforeAndAft
     avroRecord = AvroDataUtils.mapToGenericRecord(testData, plainSchema)
   }
 
-  behavior of new ScalaConfluentKafkaAvroDeserializer(None, Some(schema)).getClass.getName
-
-  it should "throw at constructor if neither topic nor Schema is informed" in {
-    assertThrows[IllegalArgumentException] {new ScalaConfluentKafkaAvroDeserializer(None, None)}
-  }
+  behavior of new ScalaConfluentKafkaAvroDeserializer(schema).getClass.getName
 
   it should "deserialize Confluent's Avro records assuming constructor Schema was used by both, writer and reader" in {
-    val deserializer = new ScalaConfluentKafkaAvroDeserializer(None, Some(schema))
-    val deserializedRecord = deserializer.deserialize(addConfluentHeader(avroRecord, schema))
+    val deserializer = new ScalaConfluentKafkaAvroDeserializer(schema)
+    val deserializedRecord = deserializer.deserialize(addConfluentHeader(avroRecord, schema, schemaId = 1))
     for (testDataEntry <- testData) {
       assert(testDataEntry._2 == deserializedRecord.get(testDataEntry._1))
     }
   }
 
-  it should "deserialize Confluent's Avro records retrieving schema from Schema Registry using topic" in {
-    val deserializer = new ScalaConfluentKafkaAvroDeserializer(Some("any_topic_since_this_is_mocked"), None)
-    SchemaManager.setConfiguredSchemaRegistry(new MockedSchemaRegistryClient)
-    val deserializedRecord = deserializer.deserialize(addConfluentHeader(avroRecord, schema))
+  it should "deserialize Confluent's Avro records retrieving schema from Schema Registry using schema id" in {
+    val schemaId = 2
+    val deserializer = new ScalaConfluentKafkaAvroDeserializer(schema)
+    SchemaManager.setConfiguredSchemaRegistry(new MockedSchemaRegistryClient(schemaId, schema))
+    val deserializedRecord = deserializer.deserialize(addConfluentHeader(avroRecord, schema, schemaId))
     for (testDataEntry <- testData) {
       assert(testDataEntry._2 == deserializedRecord.get(testDataEntry._1))
     }
   }
 
   it should "throw when Avro records are not Confluent compliant" in {
-    val deserializer = new ScalaConfluentKafkaAvroDeserializer(None, Some(schema))
+    val deserializer = new ScalaConfluentKafkaAvroDeserializer(schema)
     val avroRecordBytes = AvroDataUtils.recordToBytes(avroRecord)
     assertThrows[SerializationException] {deserializer.deserialize(avroRecordBytes)}
   }
@@ -90,29 +87,29 @@ class ScalaConfluentKafkaAvroDeserializerSpec extends FlatSpec with BeforeAndAft
     * the deserializer under test is expected fail anyway.
     */
 
-  private def addConfluentHeader(record: GenericRecord, schema: Schema): Array[Byte] = {
-    val out = getConfluentHeaderStream
-    appendBinaryRecordToConfluentStream(record, out)
+  private def addConfluentHeader(record: GenericRecord, schema: Schema, schemaId: Int): Array[Byte] = {
+    val out = getConfluentHeaderStream(schemaId)
+    appendBinaryRecordToConfluentStream(record, out, schema)
     out.toByteArray
   }
 
-  private def getConfluentHeaderStream: ByteArrayOutputStream = {
+  private def getConfluentHeaderStream(schemaId: Int): ByteArrayOutputStream = {
     val out = new ByteArrayOutputStream()
     out.write(ConfluentConstants.MAGIC_BYTE)
-    out.write(ByteBuffer.allocate(ConfluentConstants.SCHEMA_ID_SIZE_BYTES).putInt(8).array)
+    out.write(ByteBuffer.allocate(ConfluentConstants.SCHEMA_ID_SIZE_BYTES).putInt(schemaId).array)
     out
   }
 
-  private def appendBinaryRecordToConfluentStream(record: GenericRecord, out: ByteArrayOutputStream) = {
+  private def appendBinaryRecordToConfluentStream(record: GenericRecord, out: ByteArrayOutputStream, schema: Schema) = {
     val encoder = EncoderFactory.get().directBinaryEncoder(out, null)
     val writer = new SpecificDatumWriter[GenericRecord](schema)
     writer.write(record, encoder)
     encoder.flush()
   }
 
-  private class MockedSchemaRegistryClient extends SchemaRegistryClient {
+  private class MockedSchemaRegistryClient(val schemaId: Int, val schema: Schema) extends SchemaRegistryClient {
 
-    override def getVersion(s: String, schema: Schema): Int = 8
+    override def getVersion(s: String, schema: Schema): Int = schemaId
 
     override def getAllSubjects: util.Collection[String] = new util.ArrayList[String]()
 
@@ -134,7 +131,7 @@ class ScalaConfluentKafkaAvroDeserializerSpec extends FlatSpec with BeforeAndAft
 
     override def testCompatibility(s: String, schema: Schema) = true
 
-    override def register(s: String, schema: Schema): Int = 8
+    override def register(s: String, schema: Schema): Int = schemaId
 
     override def deleteSchemaVersion(conf: java.util.Map[String,String], subject: String, version: String): Integer = ???
     override def deleteSchemaVersion(subject: String, version: String): Integer = ???
@@ -143,4 +140,4 @@ class ScalaConfluentKafkaAvroDeserializerSpec extends FlatSpec with BeforeAndAft
     override def getAllVersions(subject: String): java.util.List[Integer] = ???
     override def getId(subject: String, schema: org.apache.avro.Schema): Int = ???
   }
-}
+}

Original file line number	Diff line number	Diff line change
`@@ -198,7 +198,7 @@ object AvroSerDe {`
`198`	`198`
`199`	`199`	`if (schemaId.isEmpty) {`
`200`	`200`	`throw new InvalidParameterException(s"Schema could not be registered for topic '$topic'. Make sure that the Schema Registry " +`
`201`		`- s"is available, the parameters are correct and the schemas ar compatible")`
	`201`	`+ s"is available, the parameters are correct and the schemas are compatible")`
`202`	`202`	`}`
`203`	`203`	`else {`
`204`	`204`	`logger.info(s"Schema successfully registered for topic '$topic' with id '{${schemaId.get}}'.")`
Original file line number	Diff line number	Diff line change
`@@ -41,7 +41,7 @@ object ComplexRecordsGenerator {`
`41`	`41`	`private val avroParser = new AvroToSparkParser()`
`42`	`42`	`private val random = new Random()`
`43`	`43`
`44`		`- def usedAvroSchema = plainSchema`
	`44`	`+ def usedAvroSchema: String = plainSchema`
`45`	`45`
`46`	`46`	`def generateRecords(howMany: Int): List[GenericRecord] = {`
`47`	`47`	`val result = new Array[GenericRecord](howMany)`
`@@ -123,7 +123,7 @@ object ComplexRecordsGenerator {`
`123`	`123`	`}`
`124`	`124`
`125`	`125`	`private def randomSeqOfLongs(listSize: Int) = {`
`126`		`- randomListOfLongs(listSize).asScala.toSeq`
	`126`	`+ randomListOfLongs(listSize).asScala`
`127`	`127`	`}`
`128`	`128`
`129`	`129`	`private def randomListOfStrings(listSize: Int, stringLength: Int) = {`
`@@ -135,7 +135,7 @@ object ComplexRecordsGenerator {`
`135`	`135`	`}`
`136`	`136`
`137`	`137`	`private def randomSeqOfStrings(listSize: Int, stringLength: Int) = {`
`138`		`- randomListOfStrings(listSize, stringLength).asScala.toSeq`
	`138`	`+ randomListOfStrings(listSize, stringLength).asScala`
`139`	`139`	`}`
`140`	`140`
`141`	`141`	`private def randomString(length: Int): String = {`
`@@ -144,7 +144,7 @@ object ComplexRecordsGenerator {`
`144`	`144`	`}`
`145`	`145`
`146`	`146`	`private def recordToBean(record: GenericRecord): Bean = {`
`147`		`- new Bean(`
	`147`	`+ Bean(`
`148`	`148`	`record.get("bytes").toString().getBytes(),`
`149`	`149`	`record.get("string").asInstanceOf[String],`
`150`	`150`	`record.get("int").asInstanceOf[Int],`
`@@ -156,4 +156,4 @@ object ComplexRecordsGenerator {`
`156`	`156`	`record.get("fixed").toString().getBytes,`
`157`	`157`	`record.get("map").asInstanceOf[Map[String, java.util.ArrayList[Long]]])`
`158`	`158`	`}`
`159`		`-}`
	`159`	`+}`