Example of case class serializer and deserializer for Avro

CaseClassAvroSerde.scala

class CaseClassDeserializer[CC](isKey: Boolean)(implicit format: RecordFormat[CC]) extends Deserializer[CC] with Logging {
    private val deserializer = new KafkaAvroDeserializer(null, Map(
      AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG -> Config.tracktivity.kafka.connect.`schema-registry-url`,
      KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG -> "false"
    ).asJava)

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

    override def close(): Unit = deserializer.close()

    override def deserialize(topic: String, data: Array[Byte]): CC = {
      if (data == null) null.asInstanceOf[CC]
      else {
        val record = deserializer.deserialize(topic, data).asInstanceOf[GenericRecord]
        if (record == null || format == null) logger.info(s"DESER ${record} from topic ${topic} using format ${format}. Based on data ${data}")
        format.from(record)
      }
    }
  }

  class CaseClassSerializer[CC](isKey: Boolean)(implicit format: RecordFormat[CC]) extends Serializer[CC] {
    private val serializer = new KafkaAvroSerializer(null, Map(
      AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG -> Config.tracktivity.kafka.connect.`schema-registry-url`
    ).asJava)

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

    override def close(): Unit = serializer.close()

    override def serialize(topic: String, data: CC): Array[Byte] = {
      val record = format.to(data)
      serializer.serialize(topic, record)
    }
  }

  class CaseClassSerde[CC](isKey: Boolean)(implicit format: RecordFormat[CC]) extends Serde[CC] {

    override def deserializer(): Deserializer[CC] = new CaseClassDeserializer(isKey)

    override def serializer(): Serializer[CC] = new CaseClassSerializer(isKey)

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

    override def close(): Unit = {}
  }