Hey Ewen, This makes sense. People usually do not want to stop consuming when committing offsets.
One corner case about async commit with retries I am thinking is that it is possible that two offset commits interleave with each other and that might create problem. Like you said maybe we can cancel the previous one. Another thing is that whether the future mechanism will only be applied to auto commit or it will also be used in manual commit? Because in new consumer we allow user to provide an offset map for offset commit. Simply canceling a previous pending offset commit does not seem to be ideal in this case because the two commits could be for different partitions. Thanks. Jiangjie (Becket) Qin On 4/14/15, 4:31 PM, "Ewen Cheslack-Postava" <e...@confluent.io> wrote: >I'd like to get some feedback on changing the offset commit API in the new >consumer. Since this is user-facing API I wanted to make sure this gets >better visibility than the JIRA ( >https://issues.apache.org/jira/browse/KAFKA-2123) might. > >The motivation is to make it possible to do async commits but be able to >tell when the commit completes/fails. I'm suggesting changing the API from > >void commit(Map offsets, CommitType) > >to > >Future<Void> commit(Map<TopicPartition, Long> offsets, >ConsumerCommitCallback callback); > >which matches the approach used for the producer. The >ConsumerCommitCallback only has one method: > >public void onCompletion(Exception exception); > >This enables a few different use cases: > >* Blocking commit via Future.get(), and blocking with timeouts via >Future.get(long, TimeUnit) >* See exceptions via the future (see discussion of retries below) >* Callback-based notification so you can keep processing messages and only >take action if something goes wrong, takes too long, etc. This is the use >case that motivated >* Fire and forget commits via a shorthand commit() API and ignoring the >resulting future. > >One big difference between this and the producer API is that there isn't >any result (except maybe an exception) from commitOffsets. This leads to >the somewhat awkward Future<Void> signature. I personally prefer that to >the sync/async flag, especially since it also provides a non-blocking >interface for checking whether the commit is complete. > >I posted a WIP patch to the JIRA. In the progress of making it I found a >few issues that might be worth discussing: > >1. Retries. In the old approach, this was trivial since it only applied to >synchronous calls, so we could just loop until the request was successful. >Do we want to start introducing a retries mechanism here, and should it >apply to all types of requests or are we going to end up with a couple of >different retry settings for specific cases, like offset commit. The WIP >patch allows errors to bubble up through the Future on the first failure, >which right now can cause some tests to fail transiently (e.g. consumer >bounce test). > >I think some sort of retry mechanism, even if it's an internal constant >rather than configurable, is probably the right solution, but I want to >figure out how broadly they should apply. I think adding them only for >offset commits isn't hard. > >2. The Future implementation is a bit weird because the consumer doesn't >have a dedicated IO thread. My only concern is that this could lead to >some >unintuitive results based on the current implementation because the way >this works is to just run poll() in the thread calling Future.get(), but >it >mutes all non-coordinator nodes which means other processing is >effectively >paused. If you're processing offset commits in a separate thread from your >main consumer thread that's calling poll(), you might just end up bocking >the main thread while waiting on the Future. Then again, I'm not sure the >other nodes really even need to be muted -- maybe Jay or Guozhang have >ideas on this? > >3. Should the future be cancellable? This probably isn't hard to >implement, >but I'm not sure we should even bother. On the one hand it could be nice, >especially if you have an old commit request that you want to superseded >by >a new one with updated offsets. On the other hand, if the request has >already been sent out, cancelling it won't accomplish anything. I think >the >only case this is useful is when there are retries. > >Thoughts? > >-- >Thanks, >Ewen
