[
https://issues.apache.org/jira/browse/KAFKA-4486?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15729122#comment-15729122
]
ASF GitHub Bot commented on KAFKA-4486:
---------------------------------------
GitHub user enothereska opened a pull request:
https://github.com/apache/kafka/pull/2225
KAFKA-4486: Don't commit offsets on exception
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/enothereska/kafka KAFKA-4486-exception-commit
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/kafka/pull/2225.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 #2225
----
commit fcd0e1edbd06dc47a64b72d57d155391b7680238
Author: Eno Thereska <eno.there...@gmail.com>
Date: 2016-12-07T16:09:20Z
Don't commit offsets on exception
----
> Kafka Streams - exception in process still commits offsets
> ----------------------------------------------------------
>
> Key: KAFKA-4486
> URL: https://issues.apache.org/jira/browse/KAFKA-4486
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 0.10.1.0
> Environment: Java 8
> Reporter: Joel Lundell
> Assignee: Eno Thereska
>
> I'm building a streams application and would like to be able to control the
> commits manually using ProcessorContext#commit() from an instance of
> org.apache.kafka.streams.processor.Processor.
> My use case is that I want to read messages from a topic and push them to AWS
> SQS and I need to be able to guarantee that all messages reach the queue at
> least once. I also want to use SQS batching support so my approach at the
> moment is that in Processor#process i'm saving X records in a data structure
> and when I have a full batch I send it off and if successful i commit. If I
> for any reason can't deliver the records I don't want the offsets being
> committed so that when processing works again I can start processing from the
> last successful record.
> When I was trying out the error handling I noticed that if I create a
> Processor and in the process method always throw an exception that will
> trigger StreamThread#shutdownTaskAndState which calls
> StreamThread#commitOffsets and next time I run the application it starts as
> if the previous "record" was successfully processed.
> Is there a way to achieve what I'm looking for?
> I found a similar discussion in
> https://issues.apache.org/jira/browse/KAFKA-3491 but that issue is still open.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
