title: Kafka consumer couldn't receive serialize Object?
body: <p>So i want to implement simple application which send notification kafka producer to kafka consumer.So far i have successfully send String message to producer to consumer.But when i try to send notification object kafka consumer didn't receive any objects.This is the code i have used.</p>
  
  <pre><code>public class Notification implements Serializable{
  
      private String name;
      private String message;
      private long currentTimeStamp;
  
      public String getName() {
          return name;
      }
  
      public void setName(String name) {
          this.name = name;
      }
  
      public String getMessage() {
          return message;
      }
  
      public void setMessage(String message) {
          this.message = message;
      }
  
      public long getCurrentTimeStamp() {
          return currentTimeStamp;
      }
  
      public void setCurrentTimeStamp(long currentTimeStamp) {
          this.currentTimeStamp = currentTimeStamp;
      }
  
      @Override
      public boolean equals(Object o) {
          if (this == o) return true;
          if (o == null || getClass() != o.getClass()) return false;
  
          Notification that = (Notification) o;
  
          if (currentTimeStamp != that.currentTimeStamp) return false;
          if (message != null ? !message.equals(that.message) : that.message != null) return false;
          if (name != null ? !name.equals(that.name) : that.name != null) return false;
  
          return true;
      }
  
      @Override
      public int hashCode() {
          int result = name != null ? name.hashCode() : 0;
          result = 31 * result + (message != null ? message.hashCode() : 0);
          result = 31 * result + (int) (currentTimeStamp ^ (currentTimeStamp &gt;&gt;&gt; 32));
          return result;
      }
  
      @Override
      public String toString() {
          return "Notification{" +
                  "name='" + name + '\'' +
                  ", message='" + message + '\'' +
                  ", currentTimeStamp=" + currentTimeStamp +
                  '}';
      }
  }
  </code></pre>
  
  <p>And this is producer</p>
  
  <pre><code>public class KafkaProducer {
      static String topic = "kafka-tutorial";
  
  
      public static void main(String[] args) {
          System.out.println("Start Kafka producer");
          Properties properties = new Properties();
          properties.put("metadata.broker.list", "localhost:9092");
          properties.put("serializer.class", "dev.innova.kafka.tutorial.producer.CustomSerializer");
          ProducerConfig producerConfig = new ProducerConfig(properties);
          kafka.javaapi.producer.Producer&lt;String, Notification&gt; producer = new kafka.javaapi.producer.Producer&lt;String, Notification&gt;(producerConfig);
          KeyedMessage&lt;String, Notification&gt; message = new KeyedMessage&lt;String, Notification&gt;(topic, createNotification());
          System.out.println("send Message to broker");
          producer.send(message);
          producer.close();
  
      }
  
      private static Notification createNotification(){
          Notification notification = new Notification();
          notification.setMessage("Sample Message");
          notification.setName("Sajith");
          notification.setCurrentTimeStamp(System.currentTimeMillis());
          return notification;
      }
  }
  </code></pre>
  
  <p>And this is consumer</p>
  
  <pre><code>public class KafkaConcumer extends Thread {
      final static String clientId = "SimpleConsumerDemoClient";
      final static String TOPIC = "kafka-tutorial";
      ConsumerConnector consumerConnector;
  
  
      public KafkaConcumer() {
          Properties properties = new Properties();
          properties.put("zookeeper.connect","localhost:2181");
          properties.put("group.id","test-group");
          properties.put("serializer.class", "dev.innova.kafka.tutorial.producer.CustomSerializer");
          properties.put("zookeeper.session.timeout.ms", "400");
          properties.put("zookeeper.sync.time.ms", "200");
          properties.put("auto.commit.interval.ms", "1000");
          ConsumerConfig consumerConfig = new ConsumerConfig(properties);
          consumerConnector = Consumer.createJavaConsumerConnector(consumerConfig);
      }
  
      @Override
      public void run() {
          Map&lt;String, Integer&gt; topicCountMap = new HashMap&lt;String, Integer&gt;();
          topicCountMap.put(TOPIC, new Integer(1));
          Map&lt;String, List&lt;KafkaStream&lt;byte[], byte[]&gt;&gt;&gt; consumerMap = consumerConnector.createMessageStreams(topicCountMap);
          KafkaStream&lt;byte[], byte[]&gt; stream =  consumerMap.get(TOPIC).get(0);
          ConsumerIterator&lt;byte[], byte[]&gt; it = stream.iterator();
          System.out.println("It :" + it.size());
          while(it.hasNext()){
              System.out.println(new String(it.next().message()));
          }
      }
  
      private static void printMessages(ByteBufferMessageSet messageSet) throws UnsupportedEncodingException {
          for(MessageAndOffset messageAndOffset: messageSet) {
              ByteBuffer payload = messageAndOffset.message().payload();
              byte[] bytes = new byte[payload.limit()];
              payload.get(bytes);
              System.out.println(new String(bytes, "UTF-8"));
          }
      }
  }
  </code></pre>
  
  <p>And finally i have used customserializer to serialize and deserialize object.</p>
  
  <pre><code>public class CustomSerializer implements Encoder&lt;Notification&gt;, Decoder&lt;Notification&gt; {
      public CustomSerializer(VerifiableProperties verifiableProperties) {
          /* This constructor must be present for successful compile. */
      }
      @Override
      public byte[] toBytes(Notification o) {
          return new byte[0];
      }
  
      @Override
      public Notification fromBytes(byte[] bytes) {
          return null;
      }
  }
  </code></pre>
  
  <p>Can someone tell me what is the issue ? is this the right way ? </p>