title: Publishing and Consuming Scala / Java case classes via Kafka
body: <p>What are good ways to publish and consume case classes via Kafka using Scala (or Java)?</p>
  
  <p>I found various partial examples including answers here, but no full examples.</p>
  
  <p>I managed to get this to work, using custom Kafka serialization, as follows, but I'd like feedback on if there is a better way to do it:</p>
  
  <p>Generic serializer class:</p>
  
  <pre><code>package mypackage
  
  import java.nio.charset.Charset
  import java.util
  
  import org.apache.kafka.common.errors.SerializationException
  import org.apache.kafka.common.serialization.Serializer
  import org.json4s.DefaultFormats
  import org.json4s.jackson.Serialization
  
  class ObjectSerializer[T &lt;: AnyRef] extends Serializer[T] {
  
    override def configure(config: util.Map[String, _], isKey: Boolean): Unit = {
    }
  
    override def serialize(topic: String, data: T): Array[Byte] = {
  
      if (data == null) {
        return Array.empty[Byte]
      } else {
        try {
          implicit val formats = DefaultFormats
          val json = Serialization.write[T](data)
          val bytes = json.getBytes(Charset.forName("utf-8"))
          return bytes
        } catch {
          case ex: Exception =&gt;
            throw new SerializationException(ex)
        }
      }
    }
  
    override def close(): Unit = {
    }
  }
  </code></pre>
  
  <p>Generic deserializer base class:</p>
  
  <pre><code>package mypackage
  
  import java.nio.charset.Charset
  import java.util
  import org.apache.kafka.common.errors.SerializationException
  import org.apache.kafka.common.serialization.Deserializer
  import org.json4s.DefaultFormats
  import org.json4s.jackson.Serialization
  
  class ObjectDeserializer[T &lt;: AnyRef](tType: Class[T]) extends Deserializer[T] {
  
    private val actualType = tType
  
    override def configure(props: util.Map[String, _], isKey: Boolean): Unit = {
    }
  
    override def deserialize(s: String, bytes: Array[Byte]): T = {
  
      if (bytes == null || bytes.length == 0) {
        return null.asInstanceOf[T]
      } else {
        try {
          val json = new String(bytes, Charset.forName("utf-8"))
          implicit val formats = DefaultFormats
          implicit val mf = Manifest.classType[T](actualType)
          val data = Serialization.read[T](json)(formats, mf)
          return data
        } catch {
          case ex: Exception =&gt;
            throw new SerializationException(ex)
        }
      }
    }
  
    override def close(): Unit = {
    }
  }
  </code></pre>
  
  <p>Specific deserializer class:</p>
  
  <pre><code>package mypackage
  
  import payoneer.labs.techExamples.dataModel.MyDataA
  
  class MyDataADeserializer extends ObjectDeserializer[MyDataA](classOf[MyDataA]) {
  }
  </code></pre>
  
  <p>Configuration:</p>
  
  <pre><code>ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer]
  ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[ObjectSerializer[MyDataA]]
  
  ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, classOf[StringDeserializer]
  ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, classOf[MyDataADeserializer]
  </code></pre>
  
  <p>Then use <code>KafkaProducer[String, MyDataA]</code> and <code>KafkaConsumer[String, MyDataA]</code>, assuming the key I want to use is a <code>String</code> and the value I want to use is a <code>MyDataA</code>.</p>
  
  <p>Then I can publishing using:</p>
  
  <pre><code>producer.send(new ProducerRecord[String, MyDataA](topic, key, value))
  </code></pre>
  
  <p>and consume using:</p>
  
  <pre><code>while (true) {
  
      val messages = consumer.poll(1000).iterator().asScala
  
      for (msg &lt;- messages) {
          val key : String = msg.key()
          val value : MyDataA = msg.value()
      }
  }
  </code></pre>
  
  <p>Related questions:</p>
  
  <ul>
  <li><a href="https://stackoverflow.com/q/41141924/38368">Send Custom Java Objects to Kafka Topic
  </a></li>
  <li><a href="https://stackoverflow.com/q/43612072/38368">Error serializing messages when sending to kafka topic
  </a></li>
  </ul>