title: Why doesn't a new consumer group id start from the beginning
body: <p>I have a kafka 0.10 cluster with several topics that have messages produced to them.</p>
  
  <p>When I subscribe to the topics with a KafkaConsumer and a new group Id I get no records returned, but if I subscribe to the topics with a ConsumerRebalanceListener that seeks to the beginning with the same group Id, then I get the records in the topic.</p>
  
  <pre><code>@Grab('org.apache.kafka:kafka-clients:0.10.0.0')
  
  import org.apache.kafka.clients.consumer.KafkaConsumer
  import org.apache.kafka.clients.consumer.ConsumerRecords
  import org.apache.kafka.clients.consumer.ConsumerRecord
  import org.apache.kafka.clients.consumer.ConsumerRebalanceListener
  import org.apache.kafka.common.TopicPartition
  import org.apache.kafka.common.PartitionInfo
  
  Properties props = new Properties()
  props.with {
      put("bootstrap.servers","***********:9091")
      put("group.id","script-test-noseek")
      put("enable.auto.commit","true")
      put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer")
      put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer")
      put("session.timeout.ms",30000)
  }
  
  KafkaConsumer consumer = new KafkaConsumer(props)
  def topicMap = [:]
  consumer.listTopics().each { topic, partitioninfo -&gt;
          topicMap[topic] = 0
  }
  
  topicMap.each {topic, count -&gt;
      def stopTime = new Date().time + 30_000
      def stop = false
  
      println "Starting topic: $topic"
      consumer.subscribe([topic])
      //consumer.subscribe([topic], new CRListener(consumer:consumer))
      while(!stop) {
          ConsumerRecords&lt;String, String&gt; records = consumer.poll(5_000)
          topicMap[topic] += records.size()
          consumer.commitAsync()
          if ( new Date().time &gt; stopTime || records.size() == 0) {
              stop = true
          }
      }    
      consumer.unsubscribe()
  }
  
  def total = 0
  println "------------------- Results -----------------------"
  topicMap.each { k,v -&gt;
      if ( v &gt; 0 ) {
          println "Topic: ${k.padRight(64,' ')} Records: ${v}"
      }
      total += v
  } 
  println "==================================================="
  println "Total: ${total}"
  def dummy = "Process End"
  
  class CRListener implements ConsumerRebalanceListener {
      KafkaConsumer consumer
      void onPartitionsAssigned(java.util.Collection partitions) {
          consumer.seekToBeginning(partitions)
      }
      void onPartitionsRevoked(java.util.Collection partitions) {
          consumer.commitSync()
      }
  }
  </code></pre>
  
  <p>The code is Groovy 2.4.x. And I've masked the bootstrap server.
  If I uncomment the consumer subscribe line with the listener, it does what I expect it to do. But as it is I get no results.</p>
  
  <p>Assume that I change the group Id for each run, just so as to not be picking up where another execution leaves off.</p>
  
  <p>I can't see what I'm doing wrong. Any help would be appreciated.</p>