title: Kafka - Why fresh groupId doesn't return all messages in topic when setting AUTO_OFFSET_RESET_CONFIG as "latest"
body: <p>I try to implement a very simple Kafka (0.9.0.1) consumer in scala (code below).</p>
  
  <p>For my understanding, Kafka (or better say the Zookeeper) stores for each groupId the offset of the last consumed message for a giving topic. So given the following scenario:</p>
  
  <ol>
  <li>Consumer with <code>groupId1</code> which Yesterday consumed the only 5
  messages in a topic. Now last consumed message has offset 4 (considering the 
  first message with offset 0)</li>
  <li>During the night 2 new messages arrive to the topic</li>
  <li>Today I restart the consumer, with the same <code>groupId1</code>, there will
  be two options:</li>
  </ol>
  
  <p>Option 1: The consumer will read the last 2 new messages which arrived during the night if I set the following property as <code>"latest"</code>:</p>
  
  <pre><code>props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")
  </code></pre>
  
  <p>Option 2: The consumer will read all the 7 messages in the topic if I set the following property as <code>"earliest"</code>:</p>
  
  <pre><code>props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
  </code></pre>
  
  <p><strong>Problem:</strong> For some reason, if I change the groupId of the consumer to <code>groupId2</code>, that is a new groupId for the given topic, so it never consumed any message before and its latest offset should be 0. I was expecting that by setting</p>
  
  <pre><code>props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")
  </code></pre>
  
  <p>The consumer will read during the first execution all the messages stored in the topic (the equivalent of having earliest). And then for following executions it will consume just the new ones. However this is not what happens.</p>
  
  <p>If I set a new <code>groupId</code> and keep <code>AUTO_OFFSET_RESET_CONFIG</code> as <code>latest</code>, the consumer is not able to read any message. What I need to do then is for the first run set <code>AUTO_OFFSET_RESET_CONFIG</code> as <code>earliest</code>, and once there is already an offset different to 0 for the groupID I can move to <code>latest</code>.</p>
  
  <p>Is this how it should be working my consumer? Is there a better solution than switching the <code>AUTO_OFFSET_RESET_CONFIG</code>after the first time I run the consumer?</p>
  
  <p>Below is the code I am using as a simple consumer:</p>
  
  <pre><code>class KafkaTestings {
  
    val brokers = "listOfBrokers"
    val groupId = "anyGroupId"
    val topic = "anyTopic"
  
    val props = createConsumerConfig(brokers, groupId)
  
    def createConsumerConfig(brokers: String, groupId: String): Properties = {
      val props = new Properties()
      props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
      props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId)
      props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
      props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000")
      props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000")
      props.put(ConsumerConfig.CLIENT_ID_CONFIG, "12321")
      props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")
      props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
      props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
      props
    }
  
    def run() = {
      consumer.subscribe(Collections.singletonList(this.topic))
  
      Executors.newSingleThreadExecutor.execute(    new Runnable {
        override def run(): Unit = {
  
          while (true) {
            val records = consumer.poll(1000)
  
            for (record &lt;- records) {
              println("Record: "+record.value)
            }
  
          }
  
        }
      })
    }
  }
  
  object ScalaConsumer extends App {
    val testConsumer = new KafkaTestings()
    testConsumer.run()
  } 
  </code></pre>
  
  <p><a href="https://www.programcreek.com/scala/kafka.consumer.SimpleConsumer" rel="nofollow noreferrer">This</a> was used as a reference to write this simple consumer</p>