@@ -0,0 +1,3549 @@
+package com.mapr.kafka.serializer.json;%0A%0Aimport com.fasterxml.jackson.core.JsonProcessingException;%0Aimport com.fasterxml.jackson.databind.JsonNode;%0Aimport com.fasterxml.jackson.databind.ObjectMapper;%0Aimport org.apache.kafka.clients.consumer.ConsumerConfig;%0Aimport org.apache.kafka.clients.consumer.ConsumerRecord;%0Aimport org.apache.kafka.clients.consumer.ConsumerRecords;%0Aimport org.apache.kafka.clients.consumer.KafkaConsumer;%0Aimport org.apache.kafka.common.errors.WakeupException;%0A%0Aimport java.util.Arrays;%0Aimport java.util.Properties;%0Aimport java.util.Scanner;%0A%0A/**%0A * Created by Narmatha on 06/24/17.%0A */%0Apublic class Consumer %7B%0A    private static Scanner in;%0A%0A    public static void main(String%5B%5D argv)throws Exception%7B%0A        if (argv.length != 2) %7B%0A            System.err.printf(%22Usage: %25s %3CtopicName%3E %3CgroupId%3E%5Cn%22,%0A                    com.mapr.kafka.simple.Consumer.class.getSimpleName());%0A            System.exit(-1);%0A        %7D%0A        in = new Scanner(System.in);%0A        String topicName = argv%5B0%5D;%0A        String groupId = argv%5B1%5D;%0A%0A        ConsumerThread consumerRunnable = new ConsumerThread(topicName,groupId);%0A        consumerRunnable.start();%0A        String line = %22%22;%0A        while (!line.equals(%22exit%22)) %7B%0A            line = in.next();%0A        %7D%0A        consumerRunnable.getKafkaConsumer().wakeup();%0A        System.out.println(%22Stopping consumer .....%22);%0A        consumerRunnable.join();%0A    %7D%0A%0A    private static class ConsumerThread extends Thread%7B%0A        private String topicName;%0A        private String groupId;%0A        private KafkaConsumer%3CString,JsonNode%3E kafkaConsumer;%0A%0A        public ConsumerThread(String topicName, String groupId)%7B%0A            this.topicName = topicName;%0A            this.groupId = groupId;%0A        %7D%0A        public void run() %7B%0A            Properties configProperties = new Properties();%0A            configProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, %22localhost:9092%22);%0A            configProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, %22org.apache.kafka.common.serialization.ByteArrayDeserializer%22);%0A            configProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, %22org.apache.kafka.connect.json.JsonDeserializer%22);%0A            configProperties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);%0A            configProperties.put(ConsumerConfig.CLIENT_ID_CONFIG, %22simple%22);%0A%0A%0A            //Figure out where to start processing messages from%0A            kafkaConsumer = new KafkaConsumer%3CString, JsonNode%3E(configProperties);%0A            kafkaConsumer.subscribe(Arrays.asList(topicName));%0A            ObjectMapper mapper = new ObjectMapper();%0A%0A            //Start processing messages%0A            try %7B%0A                while (true) %7B%0A                    ConsumerRecords%3CString, JsonNode%3E records = kafkaConsumer.poll(100);%0A                    for (ConsumerRecord%3CString, JsonNode%3E record : records) %7B%0A                        JsonNode jsonNode = record.value();%0A                        System.out.println(mapper.treeToValue(jsonNode,Contact.class));%0A                    %7D%0A                %7D%0A            %7Dcatch(WakeupException ex)%7B%0A                System.out.println(%22Exception caught %22 + ex.getMessage());%0A            %7D catch (JsonProcessingException e) %7B%0A                e.printStackTrace();%0A            %7D finally%7B%0A                kafkaConsumer.close();%0A                System.out.println(%22After closing KafkaConsumer%22);%0A            %7D%0A        %7D%0A        public KafkaConsumer%3CString,JsonNode%3E getKafkaConsumer()%7B%0A            return this.kafkaConsumer;%0A        %7D%0A    %7D%0A%7D%0A

Create consumer.java