title: Producing and consuming JSON in KAFKA
body: <p>We are going to deploy Apache Kafka 2.10 on our project and communicate via JSON objects between producer and consumer.</p>
  
  <p>So far I suppose I need to:</p>
  
  <ol>
  <li>Implement a custom serializer to convert JSON into byte array</li>
  <li>implement a custom deserializer to convert byte array into JSON object</li>
  <li>Produce the message</li>
  <li>Read the message in Consumer class</li>
  </ol>
  
  <p>Regarding first point, I believe it should be something like this:</p>
  
  <pre><code>@Override
  public byte[] serialize(String topic, T data) {
      if (data == null)
          return null;
      try {
          ObjectMapper mapper = new ObjectMapper();
          return mapper.writeValueAsBytes(data);
      } catch (Exception e) {
          throw new SerializationException("Error serializing JSON message", e);
      }
  }
  </code></pre>
  
  <p>Where <code>T data</code> could be passed as string <code>"{\"key\" : \"value\"}"</code>.</p>
  
  <p>However by now there are problems with 2-4 points. I tried this in custom deserializer:</p>
  
  <pre><code>@Override
  public JsonNode deserialize(String topic, byte[] bytes) {
      if (bytes == null)
          return null;
  
      JsonNode data;
      try {
          objectMapper = new ObjectMapper();
          data = objectMapper.readTree(bytes);
      } catch (Exception e) {
          throw new SerializationException(e);
      }
      return data;
  }
  </code></pre>
  
  <p>and in my Consumer I tried:</p>
  
  <pre><code>    KafkaConsumer&lt;String, TextNode&gt; consumer = new KafkaConsumer&lt;String, TextNode&gt;(messageConsumer.properties);
      consumer.subscribe(Arrays.asList(messageConsumer.topicName));
      int i = 0;
      while (true) {
          ConsumerRecords&lt;String, TextNode&gt; records = consumer.poll(100);
          for (ConsumerRecord&lt;String, TextNode&gt; record : records) {
              System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value().asText());
          }
      }
  </code></pre>
  
  <p>I thought this will produce a proper original json string, but all I got invoking <code>record.value().asText()</code> was some hash string <code>"IntcImtleVwiIDogXCJ2YWx1ZVwiIH0i"</code>.</p>
  
  <p>Any advice or example of communicating via JSON in kafka would be greatly appreciated.</p>