title: Why consumer hangs while consuming messages from Kafka on DC/OS using Client API for Java?
body: <p>I installed Kafka on DC/OS (Mesos) cluster on AWS. Enabled three brokers and created a topic called "topic1".</p>
  
  <pre><code>dcos kafka topic create topic1 --partitions 3 --replication 3
  </code></pre>
  
  <p>Then I wrote a Producer class to send messages and a Consumer class to receive them.</p>
  
  <pre><code>public class Producer {
      public static void sendMessage(String msg) throws InterruptedException, ExecutionException {
          Map&lt;String, Object&gt; producerConfig = new HashMap&lt;&gt;();
          System.out.println("setting Producerconfig.");
          producerConfig.put("bootstrap.servers", 
                  "172.16.20.207:9946,172.16.20.234:9125,172.16.20.36:9636");
  
          ByteArraySerializer serializer = new ByteArraySerializer();
          System.out.println("Creating KafkaProcuder");
          KafkaProducer&lt;byte[], byte[]&gt; kafkaProducer = new KafkaProducer&lt;&gt;(producerConfig, serializer, serializer);
          for (int i = 0; i &lt; 100; i++) {
              String msgstr = msg + i;
              byte[] message = msgstr.getBytes();
              ProducerRecord&lt;byte[], byte[]&gt; record = new ProducerRecord&lt;&gt;("topic1", message);
              System.out.println("Sent:" + msgstr);
              kafkaProducer.send(record);
          }
          kafkaProducer.close();
      }
  
      public static void main(String[] args) throws InterruptedException, ExecutionException {
          sendMessage("Kafka test message 2/27 3:32");
      }
  
  }
  
  public class Consumer {
      public static String getMessage() {
          Map&lt;String, Object&gt; consumerConfig = new HashMap&lt;&gt;();
          consumerConfig.put("bootstrap.servers", 
                  "172.16.20.207:9946,172.16.20.234:9125,172.16.20.36:9636");
          consumerConfig.put("group.id", "dj-group");
          consumerConfig.put("enable.auto.commit", "true");
          consumerConfig.put("auto.offset.reset", "earliest");
          ByteArrayDeserializer deserializer = new ByteArrayDeserializer();
          KafkaConsumer&lt;byte[], byte[]&gt; kafkaConsumer = new KafkaConsumer&lt;&gt;(consumerConfig, deserializer, deserializer);
  
          kafkaConsumer.subscribe(Arrays.asList("topic1"));
          while (true) {
              ConsumerRecords&lt;byte[], byte[]&gt; records = kafkaConsumer.poll(100);
              System.out.println(records.count() + " of records received.");
              for (ConsumerRecord&lt;byte[], byte[]&gt; record : records) {
                  System.out.println(Arrays.toString(record.value()));
              }
          }
      }
  
      public static void main(String[] args) {
          getMessage();
      }
  }
  </code></pre>
  
  <p>First I ran <code>Producer</code> on the cluster to send messages to <code>topic1</code>. However when I ran <code>Consumer</code>, it couldn't receive anything, just hang. </p>
  
  <p><code>Producer</code> is working since I was able to get all the messages by running the shell script that came with Kafka install </p>
  
  <pre><code>./bin/kafka-console-consumer.sh --zookeeper master.mesos:2181/dcos-service-kafka --topic topic1 --from-beginning
  </code></pre>
  
  <p>But why can't I receive with <code>Consumer</code>? This <a href="https://github.com/SOHU-Co/kafka-node/issues/11" rel="noreferrer">post</a> suggests group.id with old offset might be a possible cause. I only create group.id in the consumer not the producer. How do I config the offset for this group? </p>