camel-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Willem Jiang <willem.ji...@gmail.com>
Subject Re: Auto-commit and camel-kafka
Date Mon, 06 Jul 2015 02:55:39 GMT
Hi Michael,

I think we don’t take the consideration of committing the offset during the shutdown processing.
Please feel free to create a JIRA[1] and send out a patch of it.

[1]https://issues.apache.org/jira/browse/CAMEL

--  
Willem Jiang


Blog: http://willemjiang.blogspot.com (English)  
http://jnn.iteye.com (Chinese)
Twitter: willemjiang  
Weibo: 姜宁willem



On July 5, 2015 at 6:42:31 AM, Michael J. Kitchin (mcoyote.jr@gmail.com) wrote:
> Hi there,
>  
> These are questions re: the official camel-kafka integration. Since the
> issues touch on both camel and kafka I'm sending to both users lists in
> search of answers.
>  
> - - - - -
>  
> I have a simple, inonly, point-to-point, synchronous camel route (a)
> consuming from kafka topic (consumer), (b) running the resulting exchanges
> (messages) through a processor chain and (c) forwarding the outcome on to
> somewhere else.
>  
> If the runtime dies while exchanges are in this pipeline, I want things to
> pick up where they left off when restarted. I'm not picky about seeing the
> same data more than once (as long as it's recent), I just can't lose
> anything.
>  
> In brief, everything's working great except this failure/recovery part --
> in-flight exchanges are getting lost and there is no, apparent re-delivery
> on restart.
>  
> I think this has to do with consumer auto-commit, which is the default
> behavior. My reading of the kafka and camel-kafka docs suggests disabling
> auto-commit will give me what I want, but when I try it I'm not seeing
> re-delivery kick off when I restart.
>  
> After disabling auto-commit, II walked through this in the debugger, FWIW.
> If interested, you can see the code, here:
> -
> https://github.com/apache/camel/blob/master/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java
 
>  
> ...Specifically:
> - Processing is underway line 148: processor.process(exchange);
> - Camel triggers a shutdown and this returns, as normal without any
> exchange exception check
> - The barrier await hits at line
> 165: berrier.await(endpoint.getBarrierAwaitTimeoutMs(),
> TimeUnit.MILLISECONDS);
> - This aligns with the rest of the streams and triggers the barrier action,
> which in turn performs the commit at line 193: consumer.commitOffsets();
>  
> Since any exception from line 148 is suppressed and there's no subsequent
> interrupted() or exchange exception check it looks like there's no way to
> to signal not to commit and in-flight exchanges are a guaranteed loss.
>  
> Does this sound correct?
>  
> If so, barring a change from the maintainers I figure I might fork this
> code and optionally bypass the consumer.commitOffsets(); during shutdown
> or when an exchange carries an exception after the
> processor.process(exchange) call.
>  
> Thoughts?
>  
> Please let me know if I may provide any additional information. Thanks.
>  
> -Regards,
> MjK
>  
> - - - - -
>  
> Michael J. Kitchin
> Senior Software Engineer
> Operational Systems, Inc.
> 4450 Arapahoe Avenue, Suite 100
> Boulder, CO 80303
>  
> Phone: 719-271-6476
> Email: michael.kitchin@opsysinc.com
> Web: www.opsysinc.com
>  


Mime
View raw message