[
https://issues.apache.org/jira/browse/KAFKA-3960?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15381888#comment-15381888
]
ASF GitHub Bot commented on KAFKA-3960:
---------------------------------------
GitHub user 13h3r opened a pull request:
https://github.com/apache/kafka/pull/1629
KAFKA-3960 - Committed offset not set after first assign
Fixes https://issues.apache.org/jira/browse/KAFKA-3960
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/13h3r/kafka kafka-3960
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/kafka/pull/1629.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #1629
----
commit 7708b0c39f3dc5ceb7e74422aba3708acfea57a9
Author: Alexey Romanchuk <[email protected]>
Date: 2016-07-18T08:27:46Z
Fix KAFKA-3960 - Committed offset not set after first assign
----
> Committed offset not set after first assign
> -------------------------------------------
>
> Key: KAFKA-3960
> URL: https://issues.apache.org/jira/browse/KAFKA-3960
> Project: Kafka
> Issue Type: Bug
> Components: clients
> Affects Versions: 0.10.0.0
> Reporter: Alexey Romanchuk
> Priority: Blocker
>
> Committed offset did not set after first assign. Here it is minimal example
> (scala):
> {code}
> 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()
> {code}
> and the result is
> {noformat}
> 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=''}
> -----------
> {noformat}
> Pay attention to
> {noformat}
> topic-1
> Position - 1262864347
> Committed - null
> {noformat}
> There is no committed offset fetched from broker, but it is. Looks like we
> should set {{needsFetchCommittedOffsets}} to {{true}} during assign the
> partition
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
