title: Kafka consumer in java not consuming messages
body: <p>I am trying to a kafka consumer to get messages which are produced and posted to a topic in Java. My consumer goes as follows. </p>
  
  <p>consumer.java</p>
  
  <pre><code>import java.io.UnsupportedEncodingException;
  import java.nio.ByteBuffer;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.Properties;
  
  import kafka.consumer.Consumer;
  import kafka.consumer.ConsumerConfig;
  import kafka.consumer.ConsumerIterator;
  import kafka.consumer.KafkaStream;
  import kafka.javaapi.consumer.ConsumerConnector;
  import kafka.javaapi.message.ByteBufferMessageSet;
  import kafka.message.MessageAndOffset;
  
  
  
  public class KafkaConsumer extends  Thread {
      final static String clientId = "SimpleConsumerDemoClient";
      final static String TOPIC = " AATest";
      ConsumerConnector consumerConnector;
  
  
      public static void main(String[] argv) throws UnsupportedEncodingException {
          KafkaConsumer KafkaConsumer = new KafkaConsumer();
          KafkaConsumer.start();
      }
  
      public KafkaConsumer(){
          Properties properties = new Properties();
          properties.put("zookeeper.connect","10.200.208.59:2181");
          properties.put("group.id","test-group");      
          ConsumerConfig consumerConfig = new ConsumerConfig(properties);
          consumerConnector = Consumer.createJavaConsumerConnector(consumerConfig);
      }
  
      @Override
      public void run() {
          Map&lt;String, Integer&gt; topicCountMap = new HashMap&lt;String, Integer&gt;();
          topicCountMap.put(TOPIC, new Integer(1));
          Map&lt;String, List&lt;KafkaStream&lt;byte[], byte[]&gt;&gt;&gt; consumerMap = consumerConnector.createMessageStreams(topicCountMap);
          KafkaStream&lt;byte[], byte[]&gt; stream =  consumerMap.get(TOPIC).get(0);
          System.out.println(stream);
          ConsumerIterator&lt;byte[], byte[]&gt; it = stream.iterator();
          while(it.hasNext())
              System.out.println("from it");
              System.out.println(new String(it.next().message()));
  
      }
  
      private static void printMessages(ByteBufferMessageSet messageSet) throws UnsupportedEncodingException {
          for(MessageAndOffset messageAndOffset: messageSet) {
              ByteBuffer payload = messageAndOffset.message().payload();
              byte[] bytes = new byte[payload.limit()];
              payload.get(bytes);
              System.out.println(new String(bytes, "UTF-8"));
          }
      }
  }
  </code></pre>
  
  <p>When I run the above code I am getting nothing in the console wheres the java producer program behind the screen is posting data continously under the 'AATest' topic. Also the in the zookeeper console I am getting the following lines when I try running the above consumer.java</p>
  
  <pre><code>[2015-04-30 15:57:31,284] INFO Accepted socket connection from /10.200.208.59:51780 (org.apache.zookeeper.
  server.NIOServerCnxnFactory)
  [2015-04-30 15:57:31,284] INFO Client attempting to establish new session at /10.200.208.59:51780 (org.apa
  che.zookeeper.server.ZooKeeperServer)
  [2015-04-30 15:57:31,315] INFO Established session 0x14d09cebce30007 with negotiated timeout 6000 for clie
  nt /10.200.208.59:51780 (org.apache.zookeeper.server.ZooKeeperServer)
  </code></pre>
  
  <p>Also when I run a separate console-consumer pointing to the AATest topic, I am getting all the data produced by the producer to that topic. </p>
  
  <p>Both consumer and broker are in the same machine whereas the producer is in different machine. This actually resembles <a href="https://stackoverflow.com/questions/25070003/kafka-java-consumer-never-receives-any-messages">this question</a>. But going through it dint help me. Please help me.</p>