[jira] [Commented] (KAFKA-2359) New consumer - partitions auto assigned only on poll

["Ken Gidley (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-2359?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15192034#comment-15192034 ] 

Ken Gidley commented on KAFKA-2359:
-----------------------------------

How about this use case: 

I want to reprocess all the data in my log stream on application startup, using the same consumer group.id.  With 0.9.0.0, I call 'subscribe', then try to call 'seekToBeginning' and it fails with "Caused by: java.lang.IllegalStateException: No current assignment for partition test1". 

It seems weird to have to call poll(), which will read some records, and then throw them away to re-read from the beginning of the topic.

> New consumer - partitions auto assigned only on poll
> ----------------------------------------------------
>
>                 Key: KAFKA-2359
>                 URL: https://issues.apache.org/jira/browse/KAFKA-2359
>             Project: Kafka
>          Issue Type: Bug
>    Affects Versions: 0.9.0.0
>            Reporter: Stevo Slavic
>            Priority: Minor
>
> In the new consumer I encountered unexpected behavior. After constructing {{KafkaConsumer}} instance with configured consumer rebalance callback handler, and subscribing to a topic with "consumer.subscribe(topic)", retrieving subscriptions would return empty set and callback handler would not get called (no partitions ever assigned or revoked), no matter how long instance was up.
> Then I found by inspecting {{KafkaConsumer}} code that partition assignment will only be triggered on first {{poll}}, since {{pollOnce}} has:
> {noformat}
> // ensure we have partitions assigned if we expect to
> if (subscriptions.partitionsAutoAssigned())
>     coordinator.ensurePartitionAssignment();
> {noformat}
> I'm proposing to fix this by including same {{ensurePartitionAssignment}} fragment in {{KafkaConsumer.subscriptions}} accessor as well.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)