Avro deserializer for Flink's Data Stream API Kafka Source

AvroDeserializationSchema.scala

import org.apache.avro.io.DatumReader
import org.apache.avro.io.DecoderFactory
import org.apache.avro.specific.SpecificDatumReader
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.java.typeutils.TypeExtractor

class AvroDeserializationSchema[T <: SpecificRecordBase: ClassTag](private val baselineMessage: () ⇒ T) extends DeserializationSchema[T] {

  override def deserialize(message: Array[Byte]): T = {
    val bm = baselineMessage()
    val dw: DatumReader[T] = new SpecificDatumReader(bm.getSchema)
    dw.read(bm, DecoderFactory.get.binaryDecoder(message, null))
  }

  override def isEndOfStream(nextElement: T): Boolean = false

  override def getProducedType(): TypeInformation[T] = {
    TypeExtractor.getForClass(implicitly[ClassTag[T]].runtimeClass.asInstanceOf[Class[T]])
  }

}