custom serialization in kafka using scala

Question

I am trying to build a POC with Kafka 0.10. I am using my own scala domain class as a Kafka message which has a bunch of String data types. I cannot use the default serializer class or the String serializer class that comes with Kafka library. I guess I need to write my own serializer and feed it to the producer properties. If you are aware of writing an example custom serializer in Kafka (using scala), please do share. Appreciate a lot, thanks much.

Answer 1

You can write custom serializer easily in scala.You just need to extend class kafka.serializer.Encoder and override method toBytes and provide serialization logic in this method. Example code-

import com.google.gson.Gson
import kafka.utils.VerifiableProperties
class MessageSerializer(props : VerifiableProperties) extends 
kafka.serializer.Encoder[Message]{

  private val gson: Gson = new Gson()

  override def toBytes(t: Message): Array[Byte] = {
    val jsonStr = gson.toJson(t)
    jsonStr.getBytes
  }
}

In this code we are using google gson to serialize Message in json format, however you can use any other serialization framework.

Now you just need to provide this serializer in properties while instantiating producer i.e- props.put("serializer.class","codec.MessageSerializer");

Edit:- There is another way also to do it by extending Serializer directly. code-

import com.google.gson.Gson
import org.apache.kafka.common.serialization.Serializer
class MessageSerializer extends Serializer[Message]{
  private val gson: Gson = new Gson()
  override def configure(configs: util.Map[String, _], isKey: Boolean): 
  Unit = {
    // nothing to do
  }

  override def serialize(topic: String, data: Message): Array[Byte] = {
    if(data == null) 
      null
    else
      gson.toJson(data).getBytes
  }

  override def close(): Unit = {
    //nothing to do
  }
}