title: how to consume an object from producer to consumer in kafka?
body: <p>following are the classes for producer and consumer. When am able to produce the data and not able to consume it with the following code. Can any one help me here? Did I do any wrong in the coding? My aim is to read the <code>CustomMessage</code> object from the consumer and store the data in DB.</p>
  <p>In my cmd prompt I have opened 5 instances 1 for zookeeper, 1 for kafka, 1 for producer and 1 for consumer. I really don't understand. Do I need to keep all the instances up when I run the producer and consumer classes?</p>
  <p>Any pointers here will be really helpful.</p>
  <p>Thanks in advance.</p>
  <pre><code>producer class:::
              
      package com.kafka.test.demo;
      
      import java.io.IOException;
      import java.util.Properties;
      
      import javax.xml.parsers.ParserConfigurationException;
      
      import org.apache.kafka.clients.producer.KafkaProducer;
      import org.apache.kafka.clients.producer.ProducerRecord;
      import org.xml.sax.SAXException;
      
      public class KafkaaProducer {
          public static void main(String[] args) throws ParserConfigurationException, SAXException, IOException {
              Properties props = new Properties();
  //customMessage is a pojo object which should be send to the consumer..
              CustomMessage  customMessage= new CustomMessage();
              customMessage.setMessage(&quot;hello kafka&quot;);
              customMessage.setFan(&quot;1234213123&quot;);
              customMessage.setSourceSystem(&quot;Dmap&quot;);
              customMessage.setStatus(&quot;Unenrolled&quot;);
              customMessage.setMessageTyep(&quot;Simple Message&quot;);
              customMessage.setCreatedTime(&quot;5&quot;);
              customMessage.setProcessedTime(&quot;6&quot;);
              customMessage.setRetryCount(&quot;3&quot;);
              props.put(&quot;metadata.broker.list&quot;, &quot;localhost:9092&quot;);
              props.put(&quot;serializer.class&quot;, &quot;kafka.serializer.StringEncoder&quot;);
              props.put(&quot;request.required.acks&quot;, &quot;1&quot;);
              props.put(&quot;bootstrap.servers&quot;, &quot;localhost:9092,localhost:9093&quot;);
              //CustomMessageSerializer
  props.put(&quot;key.serializer&quot;,&quot;com.kafka.test.demo.CustomMessageSerializer&quot;); 
              props.put(&quot;value.serializer&quot;, &quot;com.kafka.test.demo.CustomMessageSerializer&quot;);
              try {
                  KafkaProducer&lt;String, CustomMessage&gt; producer = new KafkaProducer&lt;String, CustomMessage&gt;(props);
                  producer.send(new ProducerRecord&lt;String, CustomMessage&gt;(&quot;NewMessageTopic&quot;, &quot;customMessage&quot;,customMessage));
                  //producer.send(new ProducerRecord&lt;String, CustomMessage&gt;(&quot;NewMessageTopic&quot;, customMessage));
                  System.out.println(&quot;Message &quot; + &quot;&quot; + &quot; sent !!&quot;);
              } catch (Exception e) {
                  e.printStackTrace();
              }
          }
      }
      
      consumer class::
      package com.kafka.test.demo;
      
      import java.net.UnknownHostException;
      import java.util.Collections;
      import java.util.Properties;
      
      import org.apache.kafka.clients.consumer.ConsumerRecord;
      import org.apache.kafka.clients.consumer.ConsumerRecords;
      import org.apache.kafka.clients.consumer.KafkaConsumer;
      
      import com.mongodb.BasicDBObject;
      import com.mongodb.DB;
      import com.mongodb.DBCollection;
      import com.mongodb.DBObject;
      import com.mongodb.MongoClient;
      
      public class KafkaaConsumer {
          public static void main(String[] args) throws InterruptedException {
              Properties props = new Properties();
              props.put(&quot;zookeeper.connect&quot;, &quot;localhost:2181&quot;);
              props.put(&quot;group.id&quot;, &quot;testgroup&quot;);
              props.put(&quot;zookeeper.session.timeout.ms&quot;, &quot;4000&quot;);
              props.put(&quot;zookeeper.sync.time.ms&quot;, &quot;300&quot;);
              props.put(&quot;rebalance.backoff.ms&quot;, &quot;40000&quot;);
              props.put(&quot;bootstrap.servers&quot;, &quot;localhost:9092,localhost:9093&quot;);
              props.put(&quot;value.deserializer&quot;, &quot;com.kafka.test.demo.CustomMessageDeserializer&quot;);
              props.put(&quot;key.deserializer&quot;, &quot;com.kafka.test.demo.CustomMessageDeserializer&quot;);
              //perisitMessage();
              try{
                  KafkaConsumer&lt;String,CustomMessage&gt; consumer = new KafkaConsumer&lt;String, CustomMessage&gt;(props);
                  consumer.subscribe(Collections.singletonList(&quot;NewMessageTopic&quot;));
                  while (true) {
                      ConsumerRecords&lt;String, CustomMessage&gt; messages = consumer.poll(100);
                      for (ConsumerRecord&lt;String, CustomMessage&gt; message : messages) {
                        System.out.println(&quot;Message received &quot; + message);
                      }
                    perisitMessage();
                  }
              } catch (Exception e) {
                  e.printStackTrace();
              }
          }
      
          private static void perisitMessage() {
              // TODO Auto-generated method stub
              CustomMessage  customMessage = new CustomMessage();
              customMessage.setMessage(&quot;hello kafka&quot;);
              customMessage.setFan(&quot;1234213123&quot;);
              customMessage.setSourceSystem(&quot;Dmap&quot;);
              customMessage.setStatus(&quot;Unenrolled&quot;);
              customMessage.setMessageTyep(&quot;Simple Message&quot;);
              customMessage.setCreatedTime(&quot;5&quot;);
              customMessage.setProcessedTime(&quot;6&quot;);
              customMessage.setRetryCount(&quot;3&quot;);
              try {
                   MongoClient mongoClient = new MongoClient( &quot;localhost&quot; , 27017 );
                   DB db = mongoClient.getDB(&quot;DeviceTrack&quot;);
                   DBCollection msgCollection = db.getCollection(&quot;messages&quot;);
                   BasicDBObject document = new BasicDBObject();
                   document.put(&quot;message&quot;, customMessage.getMessage());
                   document.put(&quot;fan&quot;, customMessage.getFan());
                   document.put(&quot;SourceSystem&quot;, customMessage.getSourceSystem());
                   document.put(&quot;RetryCount&quot;, customMessage.getRetryCount());
                   document.put(&quot;ProcessedTime&quot;, customMessage.getProcessedTime());
                   document.put(&quot;CreatedTime&quot;, customMessage.getCreatedTime());
                   document.put(&quot;MessageTyep&quot;, customMessage.getMessageTyep());
                   document.put(&quot;Status&quot;, customMessage.getStatus());
                   msgCollection.insert(document);
                   System.out.println(&quot;Inserted in the data in DB succesfully&quot;);
      
              } catch (UnknownHostException e) {
                  // TODO Auto-generated catch block
                  e.printStackTrace();
              }
          }
      }
      
  package com.kafka.test.demo;
  
  import java.util.Map;
  
  import org.apache.kafka.common.serialization.Deserializer;
  
  import com.fasterxml.jackson.databind.ObjectMapper;
  
  public class CustomMessageDeserializer implements Deserializer {
  
      public Object deserialize(String arg0, byte[] arg1) {
          ObjectMapper mapper = new ObjectMapper();
          System.out.println(&quot;arg1&quot;+arg1);
          CustomMessage message = null;
          try {
              message = mapper.readValue(arg1, CustomMessage.class);
          } catch (Exception e) {
  
              e.printStackTrace();
          }
          System.out.println(&quot;&quot;+message);
          return message;
      }
  
  
  
      public void close() {
          // TODO Auto-generated method stub
  
      }
  
      public void configure(Map arg0, boolean arg1) {
          // TODO Auto-generated method stub
  
      }
  
  }
          
      package com.kafka.test.demo;
      
      import java.util.Map;
      
      import org.apache.kafka.common.serialization.Serializer;
      
      import com.fasterxml.jackson.databind.ObjectMapper;
      
      public class CustomMessageSerializer implements Serializer {
      
          public byte[] serialize(String arg0, Object arg1) {
              byte[] retVal = null;
              ObjectMapper objectMapper = new ObjectMapper();
              try {
                  retVal = objectMapper.writeValueAsString(arg1).getBytes();
              } catch (Exception e) {
                  e.printStackTrace();
              }
              System.out.println(&quot;value ::::::&quot;+retVal);
              return retVal;
          }
      
          public void close() {
              // TODO Auto-generated method stub
      
          }
      
          public void configure(Map arg0, boolean arg1) {
              // TODO Auto-generated method stub
      
          }
      }
      
  package com.kafka.test.demo;
  
  public class CustomMessage {
      
      private String messageId;
      private String parentMsgId;
      private String fan;
      private String message;
      private String sourceSystem;
      private String status;
      private String messageTyep;
      private String createdTime;
      private String processedTime;
      private String retryCount;
      
      /**
       * @return the messageId
       */
      public String getMessageId() {
          return messageId;
      }
      /**
       * @param messageId the messageId to set
       */
      public void setMessageId(String messageId) {
          this.messageId = messageId;
      }
      /**
       * @return the parentMsgId
       */
      public String getParentMsgId() {
          return parentMsgId;
      }
      /**
       * @param parentMsgId the parentMsgId to set
       */
      public void setParentMsgId(String parentMsgId) {
          this.parentMsgId = parentMsgId;
      }
      /**
       * @return the fan
       */
      public String getFan() {
          return fan;
      }
      /**
       * @param fan the fan to set
       */
      public void setFan(String fan) {
          this.fan = fan;
      }
      /**
       * @return the message
       */
      public String getMessage() {
          return message;
      }
      /**
       * @param message the message to set
       */
      public void setMessage(String message) {
          this.message = message;
      }
      /**
       * @return the sourceSystem
       */
      public String getSourceSystem() {
          return sourceSystem;
      }
      /**
       * @param sourceSystem the sourceSystem to set
       */
      public void setSourceSystem(String sourceSystem) {
          this.sourceSystem = sourceSystem;
      }
      /**
       * @return the status
       */
      public String getStatus() {
          return status;
      }
      /**
       * @param status the status to set
       */
      public void setStatus(String status) {
          this.status = status;
      }
      /**
       * @return the messageTyep
       */
      public String getMessageTyep() {
          return messageTyep;
      }
      /**
       * @param messageTyep the messageTyep to set
       */
      public void setMessageTyep(String messageTyep) {
          this.messageTyep = messageTyep;
      }
      /**
       * @return the createdTime
       */
      public String getCreatedTime() {
          return createdTime;
      }
      /**
       * @param createdTime the createdTime to set
       */
      public void setCreatedTime(String createdTime) {
          this.createdTime = createdTime;
      }
      /**
       * @return the processedTime
       */
      public String getProcessedTime() {
          return processedTime;
      }
      /**
       * @param processedTime the processedTime to set
       */
      public void setProcessedTime(String processedTime) {
          this.processedTime = processedTime;
      }
      /**
       * @return the retryCount
       */
      public String getRetryCount() {
          return retryCount;
      }
      /**
       * @param retryCount the retryCount to set
       */
      public void setRetryCount(String retryCount) {
          this.retryCount = retryCount;
      }
  }
  </code></pre>