title: Kafka Consumer does not receive messages
body: <p>I am a newbie in Kafka. I read many instructions on the Internet to make a Kafka Producer and Kafka Consumer. I did the former successfully which can send messages to Kafka cluster. However, I did not complete with the latter one. Please kindly help me to solve this problem. I saw my problem likes some posts on StackOverflow but I want to describe more clearly.
  I run Kafka and Zookeeper on Ubuntu server on Virtual Box. Use the simplest configuration (almost defaults) with 1 Kafka cluster and 1 Zookeeper cluster.</p>
  
  <p>1.When I use the command line of Kafka for producer and consumer, like: </p>
  
  <pre><code>* Case 1: It works. I can see the word: Hello, World on the screen
  
  $~/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic helloKafka --partitions 1 --replication-factor 1.
  $echo "Hello, World" | ~/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic TutorialTopic &gt; /dev/null.
  $~/kafka/bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic TutorialTopic --from-beginning.
  </code></pre>
  
  <p>2.When I use the Producer and command line of Kafka for consumer, like: </p>
  
  <pre><code>* Case 2: It works. I can see the messages which sent from the Producer on the screen
  
  $~/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic helloKafka --partitions 1 --replication-factor 1.
  $~/kafka/bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic TutorialTopic --from-beginning.
  $java -cp target/Kafka_Producer_Program-0.0.1-SNAPSHOT.jar AddLab_Producer
  </code></pre>
  
  <p>3.When I use the Producer and the Consumer, like: </p>
  
  <pre><code>* Case 3: Only Producer works perfectly. The Consumer runs but does not shows any messages. 
  
  $~/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic helloKafka --partitions 1 --replication-factor 1.
  $java -cp target/Kafka_Producer_Program-0.0.1-SNAPSHOT.jar AddLab_Producer
  $java -cp target/Kafka_Consumer_Program-0.0.1-SNAPSHOT.jar AddLab_Consumer
  </code></pre>
  
  <p>This is my code of the Producer and Consumer. Actually, I copied them from some instructions website of Kafka.</p>
  
  <p>*Producer program</p>
  
  <pre><code>import java.util.Properties;
  import java.util.concurrent.ExecutionException;
  import org.apache.kafka.clients.producer.KafkaProducer;
  import org.apache.kafka.clients.producer.ProducerConfig;
  import org.apache.kafka.clients.producer.ProducerRecord;
  import org.apache.kafka.common.serialization.StringSerializer;
  public class AddLab_Producer {
      public static void main(String args[]) throws InterruptedException, ExecutionException {
          Properties props = new Properties();
          props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
          props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
          props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  
          KafkaProducer&lt;String, String&gt; producer = new KafkaProducer&lt;String, String&gt;(props);
  
          boolean sync = false;
          String topic = args[0];
          String key = "mykey";
  
          for (int i = 1; i &lt;= 3; i++) {
              String value = args[1] + " " + i;
              ProducerRecord&lt;String, String&gt; producerRecord = new ProducerRecord&lt;String, String&gt;(topic, value);
              if (sync) {
                  producer.send(producerRecord).get();
              } else {
                  producer.send(producerRecord);
              }
          }
          producer.close();
      }
  }
  </code></pre>
  
  <p>*Consumer program</p>
  
  <pre><code>import kafka.consumer.Consumer;
  import kafka.consumer.ConsumerConfig;
  import kafka.consumer.ConsumerIterator;
  import kafka.consumer.KafkaStream;
  import kafka.javaapi.consumer.ConsumerConnector;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.Properties;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  
  public class AddLab_Consumer {
  
      public static class KafkaPartitionConsumer implements Runnable {
  
          private int tnum ;
          private KafkaStream kfs ;
  
          public KafkaPartitionConsumer(int id, KafkaStream ks) {
              tnum = id ;
              kfs = ks ;
          }   
          public void run() {
              // TODO Auto-generated method stub
              System.out.println("This is thread " + tnum) ;
  
              ConsumerIterator&lt;byte[], byte[]&gt; it = kfs.iterator();
                  int i = 1 ;
                  while (it.hasNext()) {
                      System.out.println(tnum + " " + i + ": " + new String(it.next().message()));
                      ++i ;
                  }       
          }
      }
  
      public static class MultiKafka {    
          public void run() {
          }   
      }
  
      public static void main(String[] args) {
  
          Properties props = new Properties();
          props.put("zookeeper.connect", "localhost:2181");
          props.put("group.id", "mygroupid2");
          props.put("zookeeper.session.timeout.ms", "413");
          props.put("zookeeper.sync.time.ms", "203");
          props.put("auto.commit.interval.ms", "1000");
  
          ConsumerConfig cf = new ConsumerConfig(props) ;    
          ConsumerConnector consumer = Consumer.createJavaConsumerConnector(cf) ;      
          String topic = "mytopic" ;     
          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 = consumer.createMessageStreams(topicCountMap);
          List&lt;KafkaStream&lt;byte[], byte[]&gt;&gt; streams = consumerMap.get(topic);
  
          ExecutorService executor = Executors.newFixedThreadPool(1); ;
  
          int threadnum = 0 ;     
          for(KafkaStream&lt;byte[],byte[]&gt; stream  : streams) { 
              executor.execute(new KafkaPartitionConsumer(threadnum,stream));
              ++threadnum ;
          }
      }
  }
  </code></pre>
  
  <p>*My POM.xml file</p>
  
  <pre><code>&lt;project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"&gt;
      &lt;modelVersion&gt;4.0.0&lt;/modelVersion&gt;
      &lt;groupId&gt;HelloJava&lt;/groupId&gt;
      &lt;artifactId&gt;HelloJava&lt;/artifactId&gt;
      &lt;version&gt;0.0.1-SNAPSHOT&lt;/version&gt;
  
      &lt;dependencies&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;org.apache.kafka&lt;/groupId&gt;
              &lt;artifactId&gt;kafka_2.10&lt;/artifactId&gt;
              &lt;version&gt;0.9.0.0&lt;/version&gt;
          &lt;/dependency&gt;
  
          &lt;dependency&gt;
              &lt;groupId&gt;junit&lt;/groupId&gt;
              &lt;artifactId&gt;junit&lt;/artifactId&gt;
              &lt;version&gt;3.8.1&lt;/version&gt;
              &lt;scope&gt;test&lt;/scope&gt;
          &lt;/dependency&gt;
      &lt;/dependencies&gt;
  
      &lt;build&gt;
          &lt;sourceDirectory&gt;src&lt;/sourceDirectory&gt;
          &lt;plugins&gt;
              &lt;plugin&gt;
                  &lt;groupId&gt;org.apache.maven.plugins&lt;/groupId&gt;
                  &lt;artifactId&gt;maven-shade-plugin&lt;/artifactId&gt;
                  &lt;version&gt;1.4&lt;/version&gt;
                  &lt;configuration&gt;
                      &lt;createDependencyReducedPom&gt;true&lt;/createDependencyReducedPom&gt;
                      &lt;filters&gt;
                          &lt;filter&gt;
                              &lt;artifact&gt;*:*&lt;/artifact&gt;
                              &lt;excludes&gt;
                                  &lt;exclude&gt;META-INF/*.SF&lt;/exclude&gt;
                                  &lt;exclude&gt;META-INF/*.DSA&lt;/exclude&gt;
                                  &lt;exclude&gt;META-INF/*.RSA&lt;/exclude&gt;
                              &lt;/excludes&gt;
                          &lt;/filter&gt;
                      &lt;/filters&gt;
                  &lt;/configuration&gt;
                  &lt;executions&gt;
                      &lt;execution&gt;
                          &lt;phase&gt;package&lt;/phase&gt;
                          &lt;goals&gt;
                              &lt;goal&gt;shade&lt;/goal&gt;
                          &lt;/goals&gt;
                      &lt;/execution&gt;
                  &lt;/executions&gt;
              &lt;/plugin&gt;
          &lt;/plugins&gt;
      &lt;/build&gt;
  &lt;/project&gt;
  </code></pre>
  
  <p>I am very appreciate about your help. Thank you very much.</p>
  
  <p><a href="http://i.stack.imgur.com/kPIDb.jpg" rel="nofollow">Screen of Consumer. It seem runs but can not receive any messages from Producer</a></p>