Why doesn't a new consumer group id start from the beginning
<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>