Kafka: writing custom Encoder/Serializer

[Kumar Pradeep]

I am trying to build a POC with Kafka 0.8.1. I am using my own java class as a 
Kafka message which has a bunch of String data types. For serializer.class 
property in my producer, 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 (in java), please do share. 
Appreciate a lot, thanks much.

I tried to use something like below, but I get the exception: Exception in 
thread "main" java.lang.NoSuchMethodException: 
test.EventsDataSerializer.<init>(kafka.utils.VerifiableProperties)
 at java.lang.Class.getConstructor0(Class.java:2971)


package test;

import java.io.IOException;

import com.fasterxml.jackson.core.JsonFactory;
import com.fasterxml.jackson.databind.ObjectMapper;

import kafka.message.Message;
import kafka.serializer.Decoder;
import kafka.serializer.Encoder;

public  class EventsDataSerializer implements Encoder<SimulateEvent>, 
Decoder<SimulateEvent> {
    
 public Message toMessage(SimulateEvent eventDetails) {
            try {
                    ObjectMapper mapper = new ObjectMapper(new JsonFactory());
                    byte[] serialized = mapper.writeValueAsBytes(eventDetails);
                    return new Message(serialized);
            } catch (IOException e) {
                    e.printStackTrace();
                    return null;   // TODO
            }
}
    public SimulateEvent toEvent(Message message) {
        SimulateEvent event = new SimulateEvent();

            ObjectMapper mapper = new ObjectMapper(new JsonFactory());
            try {
                    //TODO handle error
                    return mapper.readValue(message.payload().array(), 
SimulateEvent.class);
            } catch (IOException e) {
                    e.printStackTrace();
                    return null;
            }

    }

 public byte[] toBytes(SimulateEvent arg0) {
  // TODO Auto-generated method stub
  return null;
 }
 public SimulateEvent fromBytes(byte[] arg0) {
  // TODO Auto-generated method stub
  return null;
 }
}