AvroSerde.scala

package org.trizzle.kafka

import java.io.{ByteArrayOutputStream, DataOutputStream}
import java.util

import com.sksamuel.avro4s._
import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient
import org.apache.kafka.common.serialization.{Deserializer, Serde, Serializer}


object AvroSerde{
  val MAGIC_BYTE = 0.toByte

  def keySerde[T : SchemaFor : ToRecord : FromRecord]
    (implicit schemaRegistryClient: SchemaRegistryClient) = new AvroSerde[T](true, schemaRegistryClient)
  def valueSerde[T : SchemaFor : ToRecord : FromRecord]
    (implicit schemaRegistryClient: SchemaRegistryClient) = new AvroSerde[T](false, schemaRegistryClient)
}

class AvroSerde[T: SchemaFor : ToRecord : FromRecord](isKey: Boolean, schemaRegistryClient: SchemaRegistryClient) extends Serde[T]{

  private val (_deserializer, _serializer) = (new AvroDeserializer[T](), new AvroSerializer[T](isKey, schemaRegistryClient))

  override def deserializer(): Deserializer[T] = _deserializer

  override def configure(configs: util.Map[String, _], isKey: Boolean): Unit = ???

  override def close(): Unit = ()

  override def serializer(): Serializer[T] = _serializer
}

class AvroSerializer[T : SchemaFor : ToRecord](isKey: Boolean, schemaRegistryClient: SchemaRegistryClient) extends Serializer[T]{

  override def configure(configs: util.Map[String, _], isKey: Boolean): Unit = ()

  override def serialize(topic: String, data: T): Array[Byte] = {
    val schema = implicitly[SchemaFor[T]].apply()
    val schemaId = schemaRegistryClient.register(topic + (if(isKey) "-key" else "-value"), schema)

    val output = new ByteArrayOutputStream()
    val w = new DataOutputStream(output)
    w.writeInt(schemaId)
    w.writeByte(AvroSerde.MAGIC_BYTE)
    w.flush()
    val writer = AvroOutputStream.binary(output)
    writer.write(data)
    writer.flush()
    val retval = output.toByteArray
    writer.close()
    output.close()
    retval
  }

  override def close(): Unit = ()
}

class AvroDeserializer[T : SchemaFor : FromRecord] extends Deserializer[T]{

  override def configure(configs: util.Map[String, _], isKey: Boolean): Unit = ()

  override def deserialize(topic: String, data: Array[Byte]): T = AvroInputStream.binary[T](data drop(5)).iterator().toList.head


  override def close(): Unit = ()
}

build.sbt

name := "kafka-streams-avro"

version := "1.0"

scalaVersion := "2.11.8"

resolvers += "Confluent" at "http://packages.confluent.io/maven/"

libraryDependencies += "org.apache.kafka" % "kafka-streams" % "0.10.1.0"

libraryDependencies += "io.confluent" % "kafka-avro-serializer" % "3.0.1"

libraryDependencies += "io.confluent" % "kafka-schema-registry-client" % "3.0.1"

libraryDependencies += "com.sksamuel.avro4s" % "avro4s-core_2.11" % "1.6.2"

libraryDependencies += "org.scalamock" %% "scalamock-scalatest-support" % "3.2.2" % "test"

SerdeSpec.scala

package org.trizzle.kafka

import org.scalamock.scalatest.MockFactory
import com.sksamuel.avro4s.SchemaFor
import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient
import org.scalatest.{FunSpec, Matchers}
import org.apache.avro.Schema
/**
  * Created by mark on 29/11/16.
  */
class SerdeSpec extends FunSpec with Matchers with MockFactory{
  describe("Avro4sSerder"){
    describe("#serialize,#deserialize"){
      it("should serialize and deserialze appropriately"){
        implicit val schemafor = SchemaFor[SomeData]
        implicit val schemaRegMock = stub[SchemaRegistryClient]
        (schemaRegMock.register _).when(where {
          (topic: String, schema: Schema) => "topic-value" == topic
        }) returns 1

        val sut = AvroSerde.valueSerde[SomeData]
        val bytes = sut.serializer().serialize("topic", SomeData(1, "mark"))
        val deserialized = sut.deserializer().deserialize("topic", bytes)
        deserialized.id should be(1)
        deserialized.name should be("mark")
      }
    }
  }

  case class SomeData(id: Int, name: String)
}

Basic Avro serialization for Kafka Streams serdes compatible with Kafka Schema Registry