Producing and consuming JSON in KAFKA
<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>