title: Kafka Serialization of an object
body: <p>I started playing with Kafka. I've set an a zookeeper configuration, and I managed to send and consume String messages. 
  Now I am trying to pass an Object (in java), but from some reason, when parsing the Message in the consumer I have header issues. I tried several serialization options (using Decoder/Encoder), and all of the return the same header issue. </p>
  
  <p>Here is my code
  The producer:</p>
  
  <pre><code>        Properties props = new Properties();
          props.put("zk.connect", "localhost:2181");
          props.put("serializer.class", "com.inneractive.reporter.kafka.EventsDataSerializer");
          ProducerConfig config = new ProducerConfig(props);
          Producer&lt;Long, EventDetails&gt; producer = new Producer&lt;Long, EventDetails&gt;(config);
          ProducerData&lt;Long, EventDetails&gt; data = new ProducerData&lt;Long, EventDetails&gt;("test3", 1, Arrays.asList(new EventDetails());
          try {
             producer.send(data);
          } finally {
             producer.close();
          }
  </code></pre>
  
  <p>And the consumer:</p>
  
  <pre><code>        Properties props = new Properties();
          props.put("zk.connect", "localhost:2181");
          props.put("zk.connectiontimeout.ms", "1000000");
          props.put("groupid", "test_group");
  
          // Create the connection to the cluster
          ConsumerConfig consumerConfig = new ConsumerConfig(props);
          ConsumerConnector consumerConnector = Consumer.createJavaConsumerConnector(consumerConfig);
  
          // create 4 partitions of the stream for topic “test”, to allow 4 threads to consume
          Map&lt;String, List&lt;KafkaMessageStream&lt;EventDetails&gt;&gt;&gt; topicMessageStreams =
                  consumerConnector.createMessageStreams(ImmutableMap.of("test3", 4), new EventsDataSerializer());
          List&lt;KafkaMessageStream&lt;EventDetails&gt;&gt; streams = topicMessageStreams.get("test3");
  
          // create list of 4 threads to consume from each of the partitions
          ExecutorService executor = Executors.newFixedThreadPool(4);
  
          // consume the messages in the threads
          for (final KafkaMessageStream&lt;EventDetails&gt; stream: streams) {
              executor.submit(new Runnable() {
                  public void run() {
                      for(EventDetails event: stream) {
                          System.err.println("********** Got message" + event.toString());        
                      }
                  }
              });
          }
  </code></pre>
  
  <p>and my Serializer:</p>
  
  <pre><code>public  class EventsDataSerializer implements Encoder&lt;EventDetails&gt;, Decoder&lt;EventDetails&gt; {
      public Message toMessage(EventDetails eventDetails) {
          try {
              ObjectMapper mapper = new ObjectMapper(new SmileFactory());
              byte[] serialized = mapper.writeValueAsBytes(eventDetails);
              return new Message(serialized);
  } catch (IOException e) {
              e.printStackTrace();
              return null;   // TODO
          }
  }
      public EventDetails toEvent(Message message) {
          EventDetails event = new EventDetails();
  
          ObjectMapper mapper = new ObjectMapper(new SmileFactory());
          try {
              //TODO handle error
              return mapper.readValue(message.payload().array(), EventDetails.class);
          } catch (IOException e) {
              e.printStackTrace();
              return null;
          }
  
      }
  }
  </code></pre>
  
  <p>And this is the error I get:</p>
  
  <pre><code>org.codehaus.jackson.JsonParseException: Input does not start with Smile format header (first byte = 0x0) and parser has REQUIRE_HEADER enabled: can not parse
   at [Source: N/A; line: -1, column: -1]
  </code></pre>
  
  <p>When I worked with <code>MessagePack</code> and with plain writing to a <code>ObjectOutputStream</code> I got a similiar header issue. I also tried to add the payload CRC32 to the message, but that didn't help as well.</p>
  
  <p>What am I doing wrong here?</p>