[ https://issues.apache.org/jira/browse/KAFKA-2350?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14645343#comment-14645343 ]
ASF GitHub Bot commented on KAFKA-2350: --------------------------------------- GitHub user hachikuji opened a pull request: https://github.com/apache/kafka/pull/100 KAFKA-2350; KafkaConsumer pause/resume API You can merge this pull request into a Git repository by running: $ git pull https://github.com/hachikuji/kafka KAFKA-2350 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/kafka/pull/100.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 #100 ---- commit a82b60a48f47a24b55d5bffff07ddb1a22fbcf52 Author: Jason Gustafson <ja...@confluent.io> Date: 2015-07-29T00:58:10Z KAFKA-2350; KafkaConsumer pause/resume API ---- > 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 calls to poll() or if you unsubscribe, then > a rebalance will be triggered and your partitions will be reassigned to > another consumer. The desired behavior is instead that you keep the partition > assigned and simply > One way to achieve this would be to add two new methods to KafkaConsumer: > {code} > void pause(TopicPartition... partitions); > void resume(TopicPartition... partitions); > {code} > Here is the expected behavior of pause/resume: > * When a partition is paused, calls to KafkaConsumer.poll will not initiate > any new fetches for that partition. > * After the partition is resumed, fetches will begin again. > * While a partition is paused, seek() and position() can still be used to > advance or query the current position. > * Rebalance does not preserve pause/resume state. -- This message was sent by Atlassian JIRA (v6.3.4#6332)