[KAFKA-3960] Committed offset not set after first assign

• Type: Bug
• Status: Resolved
• Priority: Blocker
• Resolution: Fixed
• Affects Version/s: 0.10.0.0
• Fix Version/s: 0.10.0.1
• Component/s: clients
• Assignee: Alexey Romanchuk
• Reporter: Alexey Romanchuk
• Reviewer: Jason Gustafson
• Created: 13/Jul/16 10:21
• Updated: 23/Jul/16 08:54
• Resolved: 23/Jul/16 08:53

Description

Committed offset did not set after first assign. Here it is minimal example (scala):

  val props = new Properties()
  props.put("bootstrap.servers", "localhost:9092")
  props.put("client.id", "client1")
  props.put("group.id", "client1")
  props.put("enable.auto.commit", "false")
  props.put("key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer")
  props.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer")

  val consumer = new KafkaConsumer[Array[Byte], Array[Byte]](props)

  import scala.collection.JavaConversions._
  def dumpPositionAndCommitted() = {
    consumer.assignment().foreach { tp =>
      println(tp)
      println(s"Position - ${consumer.position(tp)}")
      println(s"Committed - ${consumer.committed(tp)}")
    }
    println("-----------")
  }

  consumer.assign(Collections.singleton(new TopicPartition("topic", 0)))
  dumpPositionAndCommitted()
  Thread.sleep(3000)
  val ps = Collections.singleton(new TopicPartition("topic", 1)) ++ consumer.assignment()
  consumer.assign(ps)
  dumpPositionAndCommitted()
  Thread.sleep(3000)
  dumpPositionAndCommitted()

and the result is

Position - 1211046445
Committed - OffsetAndMetadata{offset=1211046445, metadata=''}
-----------
topic-1
Position - 1262864347
Committed - null
topic-0
Position - 1211046445
Committed - OffsetAndMetadata{offset=1211046445, metadata=''}
-----------
topic-1
Position - 1262864347
Committed - null
topic-0
Position - 1211046445
Committed - OffsetAndMetadata{offset=1211046445, metadata=''}
-----------

Pay attention to

topic-1
Position - 1262864347
Committed - null

There is no committed offset fetched from broker, but it is. Looks like we should set needsFetchCommittedOffsets to true during assign the partition

Issue Links

• links to: GitHub Pull Request #1629