[
https://issues.apache.org/jira/browse/KAFKA-2350?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14992357#comment-14992357
]
Guozhang Wang commented on KAFKA-2350:
--------------------------------------
Could this be resolved by calling
{code}
consumer.seek(record.offset - 1)
consumer.pause(...)
{code}
So that we will only committing up to the previous message. Note that the
previous message does not need to be existed: if let's say we fetch a compacted
topic with message offsets (0, 5, 10), and we got a error on message with
offset 10, seeking and restarting at offset 9 will let us to fetch message 10
again.
> Add KafkaConsumer pause capability
> ----------------------------------
>
> Key: KAFKA-2350
> URL: https://issues.apache.org/jira/browse/KAFKA-2350
> Project: Kafka
> Issue Type: Improvement
> Components: consumer
> Reporter: Jason Gustafson
> Assignee: Jason Gustafson
> Fix For: 0.9.0.0
>
>
> 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)
