[ 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)