title: Kafka 0.10 Java consumer not reading message from topic
body: <p>I have a simple java producer like below</p>
  
  <pre><code>public class Producer 
  {
      private final static String TOPIC = "my-example-topi8";
      private final static String BOOTSTRAP_SERVERS = "localhost:8092";
  
      public static void main( String[] args ) throws Exception {
          Producer&lt;String, byte[]&gt; producer = createProducer();
          for(int i=0;i&lt;3000;i++) {
              String msg = "Test Message-" + i;
              final ProducerRecord&lt;String, byte[]&gt; record = new ProducerRecord&lt;String, byte[]&gt;(TOPIC, "key" + i, msg.getBytes());
              producer.send(record).get();
              System.out.println("Sent message " + msg);
          }
          producer.close();
      }
  
      private static Producer&lt;String, byte[]&gt; createProducer() {
          Properties props = new Properties();
          props.put("metadata.broker.list", BOOTSTRAP_SERVERS);
          props.put("bootstrap.servers", BOOTSTRAP_SERVERS);
          props.put("client.id", "AppFromJava");
          props.put("serializer.class", "kafka.serializer.DefaultEncoder");
          props.put("key.serializer.class", "kafka.serializer.StringEncoder");
          props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
          props.put("compression.codec", "snappy");
          props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
          return new KafkaProducer&lt;String, byte[]&gt;(props);
      }
  }
  </code></pre>
  
  <p>I am trying to read data as below</p>
  
  <pre><code>public class Consumer 
  {
      private final static String TOPIC = "my-example-topi8";
      private final static String BOOTSTRAP_SERVERS = "localhost:8092";
  
      public static void main( String[] args ) throws Exception {
          Consumer&lt;String, byte[]&gt; consumer = createConsumer();
          start(consumer);
      }
  
      static void start(Consumer&lt;String, byte[]&gt; consumer) throws InterruptedException {
          final int giveUp = 10;   
          int noRecordsCount = 0;
          int stopCount = 1000;
  
          while (true) {
              final ConsumerRecords&lt;String, byte[]&gt; consumerRecords = consumer.poll(1000);
              if (consumerRecords.count()==0) {
                  noRecordsCount++;
                  if (noRecordsCount &gt; giveUp) break;
                  else continue;
              }
  
  
              consumerRecords.forEach(record -&gt; {
                 // Process the record System.out.printf("\nConsumer Record:(%s, %s, %s)", record.key(), new String(record.value()), record.topic());
              });
  
              consumer.commitSync();
              break;
          }
          consumer.close();
          System.out.println("DONE");
      }
  
      private static Consumer&lt;String, byte[]&gt; createConsumer() {
          final Properties props = new Properties();
          props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                                      BOOTSTRAP_SERVERS);
          props.put(ConsumerConfig.GROUP_ID_CONFIG,
                                      "KafkaExampleConsumer");
          props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                  StringDeserializer.class.getName());
          props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                  ByteArrayDeserializer.class.getName());
          props.put(ConsumerConfig.CLIENT_ID_CONFIG, "1234");
          props.put("enable.auto.commit", "false");
  
          // Create the consumer using props.
          final Consumer&lt;String, byte[]&gt; consumer = new KafkaConsumer(props);
          consumer.subscribe(Collections.singletonList(TOPIC));
          return consumer;
      }
  }
  </code></pre>
  
  <p>But the consumer is not reading any message from kafka. If I add the below at the very <code>start()</code></p>
  
  <pre><code>consumer.poll(0);
  consumer.seekToBeginning(consumer.assignment());
  </code></pre>
  
  <p>Then the consumer starts reading from the topic. But then each time the consumer is restarted it is reading message from the start of the topic which I don't want. If I add the below config while starting Consumer</p>
  
  <pre><code>props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  </code></pre>
  
  <p>then it reads message from the topic but if the consumer gets restarted before processing all the message then it does not read the unprocessed message.</p>
  
  <p>Can someone let me know what is going wrong and how can I fix this?</p>
  
  <p>Kafka broker and zookeeper is running with the default configuration.</p>