You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@kafka.apache.org by "Ed Tyrrill (JIRA)" <ji...@apache.org> on 2016/09/13 20:10:20 UTC
[jira] [Commented] (KAFKA-2359) New consumer - partitions auto
assigned only on poll
[ https://issues.apache.org/jira/browse/KAFKA-2359?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15488276#comment-15488276 ]
Ed Tyrrill commented on KAFKA-2359:
-----------------------------------
Hi Elias,
Calling pause() causes the same IllegalArgumentException as calling the seek* methods. Stack trace:
at org.apache.kafka.clients.consumer.internals.SubscriptionState.assignedState(SubscriptionState.java:231)
at org.apache.kafka.clients.consumer.internals.SubscriptionState.pause(SubscriptionState.java:346)
at org.apache.kafka.clients.consumer.KafkaConsumer.pause(KafkaConsumer.java:1313)
Any plan on fixing this one? I am calling poll() now as a work-around.
Ed
> 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)