avro binary with schema

bijection-avro/src/main/scala/com/twitter/bijection/avro/AvroCodecs.scala

@@ -97,14 +97,16 @@ object SpecificAvroCodecs {
     withCompression(CodecFactory.snappyCodec())
 
   /**
-   * Returns Injection capable of serializing and deserializing a compiled avro record using org.apache.avro.io.BinaryEncoder
+   * Returns Injection capable of serializing and deserializing a compiled avro record using org.apache.avro.io.BinaryEncoder.
+   * Fetches the schema from the specified class in order to be compatible with generated Scala classes.
    * @tparam T compiled Avro record
    * @return Injection
    */
   def toBinary[T <: SpecificRecordBase: ClassTag]: Injection[T, Array[Byte]] = {
-    val klass = classTag[T].runtimeClass.asInstanceOf[Class[T]]
-    val writer = new SpecificDatumWriter[T](klass)
-    val reader = new SpecificDatumReader[T](klass)
+    val record = classTag[T].runtimeClass.newInstance().asInstanceOf[T]
+    val schema = record.getSchema
+    val writer = new SpecificDatumWriter[T](schema)
+    val reader = new SpecificDatumReader[T](schema)
     new BinaryAvroCodec[T](writer, reader)
   }
 

bijection-avro/src/test/java/avro/FiscalRecordScala.scala

@@ -0,0 +1,49 @@
+/** MACHINE-GENERATED FROM AVRO SCHEMA. DO NOT EDIT DIRECTLY */
+package avro
+
+case class FiscalRecordScala(var calendarDate: String, var fiscalWeek: Option[Int], var fiscalYear: Option[Int]) extends org.apache.avro.specific.SpecificRecordBase {
+  def this() = this("", Some(1), Some(1))
+  def get(field: Int): AnyRef = {
+    field match {
+      case pos if pos == 0 => {
+        calendarDate
+      }.asInstanceOf[AnyRef]
+      case pos if pos == 1 => {
+        fiscalWeek match {
+          case Some(x) => x
+          case None => null
+        }
+      }.asInstanceOf[AnyRef]
+      case pos if pos == 2 => {
+        fiscalYear match {
+          case Some(x) => x
+          case None => null
+        }
+      }.asInstanceOf[AnyRef]
+      case _ => new org.apache.avro.AvroRuntimeException("Bad index")
+    }
+  }
+  def put(field: Int, value: Any): Unit = {
+    field match {
+      case pos if pos == 0 => this.calendarDate = {
+        value match {
+          case (value: org.apache.avro.util.Utf8) => value.toString
+          case _ => value
+        }
+      }.asInstanceOf[String]
+      case pos if pos == 1 => this.fiscalWeek = {
+        Option(value)
+      }.asInstanceOf[Option[Int]]
+      case pos if pos == 2 => this.fiscalYear = {
+        Option(value)
+      }.asInstanceOf[Option[Int]]
+      case _ => new org.apache.avro.AvroRuntimeException("Bad index")
+    }
+    ()
+  }
+  def getSchema: org.apache.avro.Schema = FiscalRecordScala.SCHEMA$
+}
+
+object FiscalRecordScala {
+  val SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"FiscalRecordScala\",\"namespace\":\"avro\",\"fields\":[{\"name\":\"calendarDate\",\"type\":\"string\"},{\"name\":\"fiscalWeek\",\"type\":[\"int\",\"null\"]},{\"name\":\"fiscalYear\",\"type\":[\"int\",\"null\"]}]}")
+}

bijection-avro/src/test/scala/com/twitter/bijection/avro/SpecificAvroCodecsSpecification.scala

@@ -3,7 +3,7 @@ package com.twitter.bijection.avro
 import org.scalatest._
 import com.twitter.bijection.{ Injection, BaseProperties }
 import org.apache.avro.Schema
-import avro.FiscalRecord
+import avro.{ FiscalRecordScala, FiscalRecord }
 
 /**
  * @author Muhammad Ashraf
@@ -94,6 +94,14 @@ class SpecificAvroCodecsSpecification extends WordSpec with Matchers with BasePr
       assert(attempt.get == testRecord)
     }
 
+    "Round trip specific record using Binary Injection With Schema for Scala case classes" in {
+      implicit val specificBinaryInjection = SpecificAvroCodecs.toBinary[FiscalRecordScala]
+      val testRecord = FiscalRecordScala("2012-01-01", Some(1), Some(12))
+      val bytes = Injection[FiscalRecordScala, Array[Byte]](testRecord)
+      val attempt = Injection.invert[FiscalRecordScala, Array[Byte]](bytes)
+      assert(attempt.get == testRecord)
+    }
+
     "Round trip specific record using Json Injection" in {
       implicit val specificJsonInjection = SpecificAvroCodecs.toJson[FiscalRecord](testSchema)
       val testRecord = buildSpecificAvroRecord(("2012-01-01", 1, 12))

bijection-core/src/main/scala/com/twitter/bijection/StringInjections.scala

@@ -54,7 +54,7 @@ trait StringInjections extends NumericInjections {
           val thisBuf = if (maxSpace > buf.limit) CharBuffer.allocate(maxSpace) else buf
 
           // this is the error free result
-          @inline def assertUnderFlow(cr: CoderResult) { if(!cr.isUnderflow) cr.throwException }
+          @inline def assertUnderFlow(cr: CoderResult) { if (!cr.isUnderflow) cr.throwException }
           assertUnderFlow(dec.reset.decode(bb, thisBuf, true))
           assertUnderFlow(dec.flush(thisBuf))
           // set the limit to be the position