[jira] [Commented] (KAFKA-2350) Add KafkaConsumer pause capability

["Gwen Shapira (JIRA)" <ji...@apache.org>]

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

Gwen Shapira commented on KAFKA-2350:
-------------------------------------

Cool feature :)

Can you clarify: " If you skip poll() ... then a rebalance will be triggered "

When does a delay count as skipping? Are we obligated to do the next poll() immediately after the first one ended?
I expect to use the consumer to do something like: 
"poll until I get N messages, write those messages elsewhere, poll again". 
If the "write messages elsewhere" takes longer than expected (DB is busy kinda scenario), the consumer will lose the partitions?

(sorry if I missed important discussion elsewhere, feel free to refer me to another JIRA or thread)

> Add KafkaConsumer pause capability
> ----------------------------------
>
>                 Key: KAFKA-2350
>                 URL: https://issues.apache.org/jira/browse/KAFKA-2350
>             Project: Kafka
>          Issue Type: Improvement
>            Reporter: Jason Gustafson
>            Assignee: Jason Gustafson
>
> There are some use cases in stream processing where it is helpful to be able to pause consumption of a topic. For example, when joining two topics, you may need to delay processing of one topic while you wait for the consumer of the other topic to catch up. The new consumer currently doesn't provide a nice way to do this. If you skip poll() or if you unsubscribe, then a rebalance will be triggered and your partitions will be reassigned.
> One way to achieve this would be to add two new methods to KafkaConsumer:
> {code}
> void pause(String... topics);
> void unpause(String... topics);
> {code}
> When a topic is paused, a call to KafkaConsumer.poll will not initiate any new fetches for that topic. After it is unpaused, fetches will begin again.



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