KafkaCustomDecoder.scala

class KafkaCustomDecoder[T](implicit keyRule: RuleLike[AvroValue, T]) extends kafka.serializer.Decoder[VA[T]] with Logging {
  import kafka.message.Message
  import java.util.Properties

  def fromAvro(value: GenericRecord) =
    Avro.fromAvro[T](Avro.wrap(value))

  def fromBytes(bytes: Array[Byte]): VA[T] = {
    val props: Properties = new Properties()
    props.put("schema.registry.url", "http://localhost:8081")
    props.put("key.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer")
    props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer")

    val vProps: kafka.utils.VerifiableProperties = new kafka.utils.VerifiableProperties(props);

    val avroDecoder = new KafkaAvroDecoder(vProps)
    val genericRecord: GenericRecord = avroDecoder.fromBytes(bytes).asInstanceOf[GenericRecord]

    val decoded: VA[T] = Avro.fromAvro[T](Avro.wrap(genericRecord))

    decoded
  }
}