I like all these ideas.
Our convention is to keep method names declarative so it should probably be
subscribe(List<String> topics, Callback c)
assign(List<TopicPartition)
The javadoc would obviously have to clarify the relationship between a
subscribed topic and assigned partitions. Presumably unsubscribe/unassign
are unnecessary since this is just a matter of subscribing to the empty
list.
-Jay
On Fri, Jul 31, 2015 at 11:29 AM, Jason Gustafson <ja...@confluent.io>
wrote:
> I was thinking a little bit this morning about the subscription API and I
> have a few ideas on how to address some of the concerns about intuitiveness
> and exception handling.
>
> 1. Split the current notion of topic/partition subscription into
> subscription of topics and assignment of partitions. These concepts are
> pretty fundamentally different and I think at least some of the confusion
> about when subscriptions() can be used is caused by the fact that we
> overload the term. If instead that method is renamed to assignment(), then
> we are communicating to users that it is possible to have a subscription
> without an active assignment, which is not obvious with the current API.
> The code in fact already separates these concepts internally, so this would
> just expose it to the user.
>
> 2. Merge rebalance callback into a subscription callback and add method a
> way to handle errors. The consumer's current rebalance callback is
> basically invoked when a subscription "succeeds," so it seems a little
> weird to also provide a callback on subscription. Perhaps we can just take
> the rebalance callback out of configuration and have the user provide it on
> subscribe(). We can add a method to the callback to handle errors (e.g. for
> non-existing topics). Since the callback is provided at subscribe time, it
> should be clearer to the user that the assignment will not be ready
> immediately when subscribe returns. It's also arguably a little more
> natural to set this callback at subscription time rather than when the
> consumer is constructed.
>
> 3. Get rid of the additive subscribe methods and just use setSubscription
> which would clear the old subscription. After you start providing callbacks
> to subscribe, then the implementation starts to get tricky if each call to
> subscribe provides a separate callback. Instead, as Jay suggested, we could
> just provide a way to set the full list of subscriptions at once, and then
> there is only one callback to maintain.
>
> With these points, the API might look something like this:
>
> void setSubscription(List<String> topics, RebalanceCallback callback);
> void setAssignment(List<TopicPartition> partitions);
> List<String> subscription();
> List<TopicPartition> assignment();
>
> interface RebalanceCallback {
> void onAssignment(List<TopicPartition> partitions);
> void onRevocation(List<TopicPartition> partitions);
>
> // handle non-existing topics, etc.
> void onError(Exception e);
> }
>
> Any thoughts?
>
> -Jason
>
>
>
> On Thu, Jul 30, 2015 at 11:59 AM, Jay Kreps <j...@confluent.io> wrote:
>
>> Hey Becket,
>>
>> Yeah the high-level belief here is that it is possible to give something
>> as high level as the existing "high level" consumer, but this is not likely
>> to be the end-all be-all of high-level interfaces for processing streams of
>> messages. For example neither of these interfaces handles the threading
>> model for the processing, which obviously is a fairly low-level
>> implementation detail left to the user in you proposal, the current code,
>> as well as the existing scala consumer.
>>
>> There will be many of these: the full-fledged stream processing
>> frameworks like Storm/Spark, scalaz streams, the RxJava stuff, a more
>> traditional message queue like "processor" interface, not to mention the
>> stuff we're trying to do with KIP-28. For these frameworks it will be quite
>> weird to add a bunch of new threads since they will want to dictate the
>> threading model.
>>
>> What will be a major failure though is if this client isn't low-level
>> enough and we need to introduce another layer underneath. This would happen
>> either because we dictate too much to make it usable for various
>> applications, frameworks, or use cases. This is the concern with dictating
>> threading and processing models.
>>
>> So to summarize the goal is to subsume the existing APIs, which I think
>> we all agree this does, and be a foundation on which to build other
>> abstractions.
>>
>> WRT KIP-28, I think it is quite general and if we do that right it will
>> subsume a lot of the higher level processing and will give a full threaded
>> processing model to the user.
>>
>>
>> -Jay
>>
>>
>> On Wed, Jul 29, 2015 at 6:25 PM, Jiangjie Qin <j...@linkedin.com> wrote:
>>
>>> Thanks for the comments Jason and Jay.
>>>
>>> Jason, I had the same concern for producer's callback as well before,
>>> but it seems to be fine from some callbacks I wrote - user can always pass
>>> in object in the constructor if necessary for synchronization.
>>>
>>> Jay, I agree that the current API might be fine for people who wants to
>>> wrap it up. But I thought the new consumer was supposed to be a combination
>>> of old high and low level consumer, which means it should be able to be
>>> used as is, just like producer. If KafkaConsumer is designed to be wrapped
>>> up for use, then the question becomes whether Kafka will provide a decent
>>> wrapper or not? Neha mentioned that KIP-28 will address the users who only
>>> care about data. Would that be the wrapper provided by Kafka? I am not sure
>>> if that is sufficient though because the processor is highly abstracted,
>>> and might only meet the static data stream requirement as I listed in the
>>> grid. For users who need something from the other grids, are we going to
>>> have another wrapper? Or are we expecting all the user to write their own
>>> wrapper for KafkaConsumer? Some other comments are in line.
>>>
>>> Thanks,
>>>
>>> Jiangjie (Becket) Qin
>>>
>>> On Wed, Jul 29, 2015 at 3:16 PM, Jay Kreps <j...@confluent.io> wrote:
>>>
>>>> Some comments on the proposal:
>>>>
>>>> I think we are conflating a number of things that should probably be
>>>> addressed individually because they are unrelated. My past experience is
>>>> that this always makes progress hard. The more we can pick apart these
>>>> items the better:
>>>>
>>>> 1. threading model
>>>> 2. blocking vs non-blocking semantics
>>>> 3. missing apis
>>>> 4. missing javadoc and other api surprises
>>>> 5. Throwing exceptions.
>>>>
>>>> The missing APIs are getting added independently. Some like your
>>>> proposed offsetByTime where things we agreed to hold off on for the first
>>>> release and do when we'd thought it through. If there are uses for it now
>>>> we can accelerate. I think each of these is really independent, we know
>>>> there are things that need to be added but lumping them all into one
>>>> discussion will be confusing.
>>>>
>>>> WRT throwing exceptions the policy is to throw exceptions that are
>>>> unrecoverable and handle and log other exceptions that are transient. That
>>>> policy makes sense if you go through the thought exercise of "what will the
>>>> user do if i throw this exception to them" if they have no other rational
>>>> response but to retry (and if failing to anticipate and retry with that
>>>> exception will kill their program) . You can argue whether the topic not
>>>> existing is transient or not, unfortunately the way we did auto-creation
>>>> makes it transient if you are in "auto create mode" and non-transient
>>>> otherwise (ick!). In any case this is an orthogonal discussion to
>>>> everything else. I think the policy is right and if we don't conform to it
>>>> in some way that is really an independent bug/discussion.
>>>>
>>> Agreed we can discuss about them separately.
>>>
>>>>
>>>> I suggest we focus on threading and the current event-loop style of api
>>>> design since I think that is really the crux.
>>>>
>>>> The analogy between the producer threading model and the consumer model
>>>> actually doesn't work for me. The goal of the producer is actually to take
>>>> requests from many many user threads and shove them into a single buffer
>>>> for batching. So the threading model isn't the 1:1 threads you describe it
>>>> is N:1.The goal of the consumer is to support single-threaded processing.
>>>> This is what drives the difference. Saying that the producer has N:1
>>>> threads therefore for the consumer should have 1:1 threads instead of just
>>>> 1 thread doesn't make sense any more then an analogy to the brokers
>>>> threading model would--the problem we're solving is totally different.
>>>>
>>> I think the ultimate goal for producer and consumer are still allowing
>>> user to send/receive data in parallel. In producer we picked the solution
>>> of one-producer-serving-multiple-threads, and in consumer we picked
>>> multiple-single-threaded-consumers instead of
>>> single-consumer-serving-multiple threads. And we believe people can always
>>> implement the latter with the former. I think this is a reasonable
>>> decision. However, there are also reasonable concerns over the
>>> multiple-single-threaded-consumers solution which is that the single-thread
>>> might have to be a dedicate polling thread in many cases which pushes user
>>> towards the other solution - i.e. implementing a
>>> single-thread-consumer-serving-multiple-threads wrapper. From what we hear,
>>> it seems to be a quite common concern for most of the users we talked to.
>>> Plus the adoption bar of the consumer will be much higher because user will
>>> have to understand some of the details of the things they don't care as
>>> listed in the grid.
>>> The analogy between producer/consumer is intended to show that a
>>> separate polling thread will solve the concerns we have.
>>>
>>>>
>>>>
>>> I think ultimately though what you need to think about is, does an event
>>>> loop style of API make sense? That is the source of all the issues you
>>>> describe. This style of API is incredibly prevalent from unix select to
>>>> GUIs to node.js. It's a great way to model multiple channels of messages
>>>> coming in. It is a fantastic style for event processing. Programmers
>>>> understand this style of api though I would agree it is unusual compared to
>>>> blocking apis. But it is is a single threaded processing model. The current
>>>> approach is basically a pure event loop with some convenience methods that
>>>> are effectively "poll until X is complete".
>>>>
>>>> I think basically all the confusion you are describing comes from not
>>>> documenting/expecting an event loop. The "if you don't call poll nothing
>>>> happens" point is basically this. It's an event loop. You have to loop. You
>>>> can't not call poll. The docs don't cover this right now, perhaps. I think
>>>> if they do it's not unreasonable behavior.
>>>>
>>> I'm not sure if I understand the event-loop correctly and honestly I did
>>> not think about it clearly before. My understanding is that an even-loop
>>> model means a single listener thread, but there can be multiple event
>>> generator threads. The downside is that the listener thread has to be fast
>>> and very careful about blocking. If we look at the consumer, the current
>>> model is the caller thread itself act as both event generator and listener.
>>> As a generator, it generates different task by calling the convenience
>>> methods. As a listener, it listens to the messages on broker and also the
>>> tasks generated by itself. So in our proposal, we are not changing the
>>> event-loop model here just separated the event generator and event
>>> listener. It looks to me that the underlying execution thread follows the
>>> event-loop model, the special thing might be it is not only listening to
>>> the messages from broker, but also listening to the tasks from the user
>>> thread. This is essentially the thing a consumer has to do - interact with
>>> both server and user.
>>>
>>>>
>>>> If we want to move away from an event loop I'm not sure *any* aspect
>>>> of the current event loop style of api makes sense any more. I am not
>>>> totally married to event loops, but i do think what we have gives an
>>>> elegant way of implementing any higher level abstractions that would fully
>>>> implement the user's parallelism model. I don't want to go rethink
>>>> everything but I do think a half-way implementation that is event loop +
>>>> background threads is likely going to be icky.
>>>>
>>> We brought this up before to change the consumer.poll() to
>>> consumer.consume(). And did not do so simply because we wanted to less
>>> change in API... I might be crazy but can we think of the proposed model as
>>> processing thread + event-loop instead, rather than event-loop + background
>>> thread?
>>>
>>>>
>>>> WRT making it configurable whether liveness means "actually consuming"
>>>> or "background thread running" I would suggest that that is really the
>>>> worst outcome. These type of "modes" that are functionally totally
>>>> different are just awful from a documentation, testing, usability, etc pov.
>>>> I would strongly prefer we pick either of these, document it, and make it
>>>> work well rather than trying to do both.
>>>>
>>> Previously I thought this was the major benefit we wanted from a single
>>> threaded model, personally I don't have a strong preference on this. So I
>>> am OK with either way.
>>>
>>>>
>>>> -Jay
>>>>
>>>>
>>>>
>>>>
>>>>
>>>>
>>>>
>>>>
>>>> On Wed, Jul 29, 2015 at 1:20 PM, Neha Narkhede <n...@confluent.io>
>>>> wrote:
>>>>
>>>>> Works now. Thanks Becket!
>>>>>
>>>>> On Wed, Jul 29, 2015 at 1:19 PM, Jiangjie Qin <j...@linkedin.com>
>>>>> wrote:
>>>>>
>>>>>> Ah... My bad, forgot to change the URL link for pictures.
>>>>>> Thanks for the quick response, Neha. It should be fixed now, can you
>>>>>> try again?
>>>>>>
>>>>>> Jiangjie (Becket) Qin
>>>>>>
>>>>>> On Wed, Jul 29, 2015 at 1:10 PM, Neha Narkhede <n...@confluent.io>
>>>>>> wrote:
>>>>>>
>>>>>>> Thanks Becket. Quick comment - there seem to be a bunch of images
>>>>>>> that the wiki refers to, but none loaded for me. Just making sure if its
>>>>>>> just me or can everyone not see the pictures?
>>>>>>>
>>>>>>> On Wed, Jul 29, 2015 at 12:00 PM, Jiangjie Qin <j...@linkedin.com>
>>>>>>> wrote:
>>>>>>>
>>>>>>>> I agree with Ewen that a single threaded model will be tricky to
>>>>>>>> implement the same conventional semantic of async or Future. We just
>>>>>>>> drafted the following wiki which explains our thoughts in LinkedIn on
>>>>>>>> the
>>>>>>>> new consumer API and threading model.
>>>>>>>>
>>>>>>>>
>>>>>>>> https://cwiki.apache.org/confluence/display/KAFKA/New+consumer+API+change+proposal
>>>>>>>>
>>>>>>>> We were trying to see:
>>>>>>>> 1. If we can use some kind of methodology to help us think about
>>>>>>>> what API we want to provide to user for different use cases.
>>>>>>>> 2. What is the pros and cons of current single threaded model. Is
>>>>>>>> there a way that we can maintain the benefits while solve the issues
>>>>>>>> we are
>>>>>>>> facing now with single threaded model.
>>>>>>>>
>>>>>>>> Thanks,
>>>>>>>>
>>>>>>>> Jiangjie (Becket) Qin
>>>>>>>>
>>>>>>>> On Tue, Jul 28, 2015 at 10:28 PM, Ewen Cheslack-Postava <
>>>>>>>> e...@confluent.io> wrote:
>>>>>>>>
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> On Tue, Jul 28, 2015 at 5:18 PM, Guozhang Wang <wangg...@gmail.com
>>>>>>>>> > wrote:
>>>>>>>>>
>>>>>>>>>> I think Ewen has proposed these APIs for using callbacks along
>>>>>>>>>> with returning future in the commit calls, i.e. something similar to:
>>>>>>>>>>
>>>>>>>>>> public Future<void> commit(ConsumerCommitCallback callback);
>>>>>>>>>>
>>>>>>>>>> public Future<void> commit(Map<TopicPartition, Long> offsets,
>>>>>>>>>> ConsumerCommitCallback callback);
>>>>>>>>>>
>>>>>>>>>> At that time I was slightly intending not to include the Future
>>>>>>>>>> besides adding the callback mainly because of the implementation
>>>>>>>>>> complexity
>>>>>>>>>> I feel it could introduce along with the retry settings after looking
>>>>>>>>>> through the code base. I would happy to change my mind if we could
>>>>>>>>>> propose
>>>>>>>>>> a prototype implementation that is simple enough.
>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>> One of the reasons that interface ended up being difficult (or
>>>>>>>>> maybe impossible) to make work reasonably is because the consumer was
>>>>>>>>> thread-safe at the time. That made it impossible to know what should
>>>>>>>>> be
>>>>>>>>> done when Future.get() is called -- should the implementation call
>>>>>>>>> poll()
>>>>>>>>> itself, or would the fact that the user is calling get() imply that
>>>>>>>>> there's
>>>>>>>>> a background thread running the poll() loop and we just need to wait
>>>>>>>>> for it?
>>>>>>>>>
>>>>>>>>> The consumer is no longer thread safe, but I think the same
>>>>>>>>> problem remains because the expectation with Futures is that they are
>>>>>>>>> thread safe. Which means that even if the consumer isn't thread safe,
>>>>>>>>> I
>>>>>>>>> would expect to be able to hand that Future off to some other thread,
>>>>>>>>> have
>>>>>>>>> the second thread call get(), and then continue driving the poll loop
>>>>>>>>> in my
>>>>>>>>> thread (which in turn would eventually resolve the Future).
>>>>>>>>>
>>>>>>>>> I quite dislike the sync/async enum. While both operations commit
>>>>>>>>> offsets, their semantics are so different that overloading a single
>>>>>>>>> method
>>>>>>>>> with both is messy. That said, I don't think we should consider this
>>>>>>>>> an
>>>>>>>>> inconsistency wrt the new producer API's use of Future because the
>>>>>>>>> two APIs
>>>>>>>>> have a much more fundamental difference that justifies it: they have
>>>>>>>>> completely different threading and execution models.
>>>>>>>>>
>>>>>>>>> I think a Future-based API only makes sense if you can guarantee
>>>>>>>>> the operations that Futures are waiting on will continue to make
>>>>>>>>> progress
>>>>>>>>> regardless of what the thread using the Future does. The producer API
>>>>>>>>> makes
>>>>>>>>> that work by processing asynchronous requests in a background thread.
>>>>>>>>> The
>>>>>>>>> new consumer does not, and so it becomes difficult/impossible to
>>>>>>>>> implement
>>>>>>>>> the Future correctly. (Or, you have to make assumptions which break
>>>>>>>>> other
>>>>>>>>> use cases; if you want to support the simple use case of just making a
>>>>>>>>> commit() synchronous by calling get(), the Future has to call poll()
>>>>>>>>> internally; but if you do that, then if any user ever wants to add
>>>>>>>>> synchronization to the consumer via some external mechanism, then the
>>>>>>>>> implementation of the Future's get() method will not be subject to
>>>>>>>>> that
>>>>>>>>> synchronization and things will break).
>>>>>>>>>
>>>>>>>>> -Ewen
>>>>>>>>>
>>>>>>>>>
>>>>>>>>>> Guozhang
>>>>>>>>>>
>>>>>>>>>> On Tue, Jul 28, 2015 at 4:03 PM, Neha Narkhede <n...@confluent.io
>>>>>>>>>> > wrote:
>>>>>>>>>>
>>>>>>>>>>> Hey Adi,
>>>>>>>>>>>
>>>>>>>>>>> When we designed the initial version, the producer API was still
>>>>>>>>>>> changing. I thought about adding the Future and then just didn't
>>>>>>>>>>> get to it.
>>>>>>>>>>> I agree that we should look into adding it for consistency.
>>>>>>>>>>>
>>>>>>>>>>> Thanks,
>>>>>>>>>>> Neha
>>>>>>>>>>>
>>>>>>>>>>> On Tue, Jul 28, 2015 at 1:51 PM, Aditya Auradkar <
>>>>>>>>>>> aaurad...@linkedin.com> wrote:
>>>>>>>>>>>
>>>>>>>>>>>> Great discussion everyone!
>>>>>>>>>>>>
>>>>>>>>>>>> One general comment on the sync/async API's on the new consumer.
>>>>>>>>>>>> I think the producer tackles sync vs async API's well. For
>>>>>>>>>>>> API's that can either be sync or async, can we simply return a
>>>>>>>>>>>> future? That
>>>>>>>>>>>> seems more elegant for the API's that make sense either in both
>>>>>>>>>>>> flavors.
>>>>>>>>>>>> From the users perspective, it is more consistent with the new
>>>>>>>>>>>> producer.
>>>>>>>>>>>> One easy example is the commit call with the CommitType enum.. we
>>>>>>>>>>>> can make
>>>>>>>>>>>> that call always async and users can block on the future if they
>>>>>>>>>>>> want to
>>>>>>>>>>>> make sure their offsets are committed.
>>>>>>>>>>>>
>>>>>>>>>>>> Aditya
>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>> On Mon, Jul 27, 2015 at 2:06 PM, Onur Karaman <
>>>>>>>>>>>> okara...@linkedin.com> wrote:
>>>>>>>>>>>>
>>>>>>>>>>>>> Thanks for the great responses, everyone!
>>>>>>>>>>>>>
>>>>>>>>>>>>> To expand a tiny bit on my initial post: while I did bring up
>>>>>>>>>>>>> old high level consumers, the teams we spoke to were actually not
>>>>>>>>>>>>> the types
>>>>>>>>>>>>> of services that simply wanted an easy way to get
>>>>>>>>>>>>> ConsumerRecords. We spoke
>>>>>>>>>>>>> to infrastructure teams that I would consider to be closer to the
>>>>>>>>>>>>> "power-user" end of the spectrum and would want KafkaConsumer's
>>>>>>>>>>>>> level of
>>>>>>>>>>>>> granularity. Some would use auto group management. Some would use
>>>>>>>>>>>>> explicit
>>>>>>>>>>>>> group management. All of them would turn off auto offset commits.
>>>>>>>>>>>>> Yes, the
>>>>>>>>>>>>> Samza team had prior experience with the old SimpleConsumer, but
>>>>>>>>>>>>> this is
>>>>>>>>>>>>> the first kafka consumer being used by the Databus team. So I
>>>>>>>>>>>>> don't really
>>>>>>>>>>>>> think the feedback received was about the simpler times or wanting
>>>>>>>>>>>>> additional higher-level clients.
>>>>>>>>>>>>>
>>>>>>>>>>>>> - Onur
>>>>>>>>>>>>>
>>>>>>>>>>>>> On Mon, Jul 27, 2015 at 1:41 PM, Jason Gustafson <
>>>>>>>>>>>>> ja...@confluent.io> wrote:
>>>>>>>>>>>>>
>>>>>>>>>>>>>> I think if we recommend a longer session timeout, then we
>>>>>>>>>>>>>> should expose the heartbeat frequency in configuration since
>>>>>>>>>>>>>> this generally
>>>>>>>>>>>>>> controls how long normal rebalances will take. I think it's
>>>>>>>>>>>>>> currently
>>>>>>>>>>>>>> hard-coded to 3 heartbeats per session timeout. It could also be
>>>>>>>>>>>>>> nice to
>>>>>>>>>>>>>> have an explicit LeaveGroup request to implement clean shutdown
>>>>>>>>>>>>>> of a
>>>>>>>>>>>>>> consumer. Then the coordinator doesn't have to wait for the
>>>>>>>>>>>>>> timeout to
>>>>>>>>>>>>>> reassign partitions.
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> -Jason
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> On Mon, Jul 27, 2015 at 1:25 PM, Jay Kreps <j...@confluent.io>
>>>>>>>>>>>>>> wrote:
>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> Hey Kartik,
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> Totally agree we don't want people tuning timeouts in the
>>>>>>>>>>>>>>> common case.
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> However there are two ways to avoid this:
>>>>>>>>>>>>>>> 1. Default the timeout high
>>>>>>>>>>>>>>> 2. Put the heartbeat in a separate thread
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> When we were doing the consumer design we discussed this
>>>>>>>>>>>>>>> tradeoff and I think the conclusion we came to was that
>>>>>>>>>>>>>>> defaulting to a
>>>>>>>>>>>>>>> high timeout was actually better. This means it takes a little
>>>>>>>>>>>>>>> longer to
>>>>>>>>>>>>>>> detect a failure, but usually that is not a big problem and
>>>>>>>>>>>>>>> people who want
>>>>>>>>>>>>>>> faster failure detection can tune it down. This seemed better
>>>>>>>>>>>>>>> than having
>>>>>>>>>>>>>>> the failure detection not really cover the consumption and just
>>>>>>>>>>>>>>> be a
>>>>>>>>>>>>>>> background ping. The two reasons where (a) you still have the
>>>>>>>>>>>>>>> GC problem
>>>>>>>>>>>>>>> even for the background thread, (b) consumption is in some
>>>>>>>>>>>>>>> sense a better
>>>>>>>>>>>>>>> definition of an active healthy consumer and a lot of problems
>>>>>>>>>>>>>>> crop up when
>>>>>>>>>>>>>>> you have an inactive consumer with an active background thread
>>>>>>>>>>>>>>> (as today).
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> When we had the discussion I think what we realized was that
>>>>>>>>>>>>>>> most people who were worried about the timeout where imagining
>>>>>>>>>>>>>>> a very low
>>>>>>>>>>>>>>> default (500ms) say. But in fact just setting this to 60
>>>>>>>>>>>>>>> seconds or higher
>>>>>>>>>>>>>>> as a default would be okay, this adds to the failure detection
>>>>>>>>>>>>>>> time but
>>>>>>>>>>>>>>> only apps that care about this need to tune. This should
>>>>>>>>>>>>>>> largely eliminate
>>>>>>>>>>>>>>> false positives since after all if you disappear for 60 seconds
>>>>>>>>>>>>>>> that
>>>>>>>>>>>>>>> actually starts to be more of a true positive, even if you come
>>>>>>>>>>>>>>> back... :-)
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> -Jay
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> On Mon, Jul 27, 2015 at 1:05 PM, Kartik Paramasivam <
>>>>>>>>>>>>>>> kparamasi...@linkedin.com> wrote:
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> adding the open source alias. This email started off as a
>>>>>>>>>>>>>>>> broader discussion around the new consumer. I was zooming
>>>>>>>>>>>>>>>> into only the
>>>>>>>>>>>>>>>> aspect of poll() being the only mechanism for driving the
>>>>>>>>>>>>>>>> heartbeats.
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> Yes the lag is the effect of the problem (not the
>>>>>>>>>>>>>>>> problem). Monitoring the lag is important as it is the
>>>>>>>>>>>>>>>> primary way to tell
>>>>>>>>>>>>>>>> if the application is wedged. There might be other metrics
>>>>>>>>>>>>>>>> which can
>>>>>>>>>>>>>>>> possibly capture the same essence. Yes the lag is at the
>>>>>>>>>>>>>>>> consumer group
>>>>>>>>>>>>>>>> level, but you can tell that one of the consumers is messed up
>>>>>>>>>>>>>>>> if one of
>>>>>>>>>>>>>>>> the partitions in the application start generating lag and
>>>>>>>>>>>>>>>> others are good
>>>>>>>>>>>>>>>> for e.g.
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> Monitoring aside, I think the main point of concern is that
>>>>>>>>>>>>>>>> in the old consumer most customers don't have to worry about
>>>>>>>>>>>>>>>> unnecessary
>>>>>>>>>>>>>>>> rebalances and most of the things that they do in their app
>>>>>>>>>>>>>>>> doesn't have an
>>>>>>>>>>>>>>>> impact on the session timeout.. (i.e. the only thing that
>>>>>>>>>>>>>>>> causes
>>>>>>>>>>>>>>>> rebalances is when the GC is out of whack). For the handful
>>>>>>>>>>>>>>>> of customers
>>>>>>>>>>>>>>>> who are impacted by GC related rebalances, i would imagine
>>>>>>>>>>>>>>>> that all of them
>>>>>>>>>>>>>>>> would really want us to make the system more resilient. I
>>>>>>>>>>>>>>>> agree that the
>>>>>>>>>>>>>>>> GC problem can't be solved easily in the java client, however
>>>>>>>>>>>>>>>> it appears
>>>>>>>>>>>>>>>> that now we would be expecting the consuming applications to
>>>>>>>>>>>>>>>> be even more
>>>>>>>>>>>>>>>> careful with ongoing tuning of the timeouts. At LinkedIn, we
>>>>>>>>>>>>>>>> have seen
>>>>>>>>>>>>>>>> that most kafka applications don't have much of a clue about
>>>>>>>>>>>>>>>> configuring
>>>>>>>>>>>>>>>> the timeouts and just end up calling the Kafka team when their
>>>>>>>>>>>>>>>> application
>>>>>>>>>>>>>>>> sees rebalances.
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> The other side effect of poll driving the heartbeats is
>>>>>>>>>>>>>>>> that we have to make sure that people don't set a poll timeout
>>>>>>>>>>>>>>>> that is
>>>>>>>>>>>>>>>> larger than the session timeout. If we had a notion of
>>>>>>>>>>>>>>>> implicit
>>>>>>>>>>>>>>>> heartbeats then we could also automatically make this work for
>>>>>>>>>>>>>>>> consumers by
>>>>>>>>>>>>>>>> sending hearbeats at the appropriate interval even though the
>>>>>>>>>>>>>>>> customers
>>>>>>>>>>>>>>>> want to do a long poll.
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> We could surely work around this in LinkedIn if either we
>>>>>>>>>>>>>>>> have the Pause() api or an explicit HeartBeat() api on the
>>>>>>>>>>>>>>>> consumer.
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> Would love to hear how other people think about this
>>>>>>>>>>>>>>>> subject ?
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> Thanks
>>>>>>>>>>>>>>>> Kartik
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> On Sat, Jul 25, 2015 at 7:41 PM, Neha Narkhede <
>>>>>>>>>>>>>>>> n...@confluent.io> wrote:
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> Agree with the dilemma you are pointing out, which is that
>>>>>>>>>>>>>>>>> there are many ways the application's message processing
>>>>>>>>>>>>>>>>> could fail and we
>>>>>>>>>>>>>>>>> wouldn't be able to model all of those in the consumer's
>>>>>>>>>>>>>>>>> failure detection
>>>>>>>>>>>>>>>>> mechanism. So we should try to model as much of it as we can
>>>>>>>>>>>>>>>>> so the
>>>>>>>>>>>>>>>>> consumer's failure detection is meaningful.
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> Point being that the only absolute way to really detect
>>>>>>>>>>>>>>>>>> that an app is healthy is to monitor lag. If the lag
>>>>>>>>>>>>>>>>>> increases then for
>>>>>>>>>>>>>>>>>> sure something is wrong.
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> The lag is merely the effect of the problem, not the
>>>>>>>>>>>>>>>>> problem itself. Lag is also a consumer group level concept
>>>>>>>>>>>>>>>>> and the problem
>>>>>>>>>>>>>>>>> we have is being able to detect failures at the level of
>>>>>>>>>>>>>>>>> individual
>>>>>>>>>>>>>>>>> consumer instances.
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> As you pointed out, a consumer that poll() is a stronger
>>>>>>>>>>>>>>>>> indicator of whether the consumer is alive or not. The
>>>>>>>>>>>>>>>>> dilemma then is who
>>>>>>>>>>>>>>>>> defines what a healthy poll() frequency is. No one else but
>>>>>>>>>>>>>>>>> the application
>>>>>>>>>>>>>>>>> owner can define what a "normal" processing latency is for
>>>>>>>>>>>>>>>>> their
>>>>>>>>>>>>>>>>> application. Now the question is what's the easiest way for
>>>>>>>>>>>>>>>>> the user to
>>>>>>>>>>>>>>>>> define this without having to tune and fine tune this too
>>>>>>>>>>>>>>>>> often. The
>>>>>>>>>>>>>>>>> heartbeat interval certainly does not have to be *exactly*
>>>>>>>>>>>>>>>>> 99tile of processing latency but could be in the ballpark +
>>>>>>>>>>>>>>>>> an error delta.
>>>>>>>>>>>>>>>>> The error delta is the application owner's acceptable risk
>>>>>>>>>>>>>>>>> threshold during
>>>>>>>>>>>>>>>>> which they would be ok if the application remains part of the
>>>>>>>>>>>>>>>>> group despite
>>>>>>>>>>>>>>>>> being dead. It is ultimately a tradeoff between operational
>>>>>>>>>>>>>>>>> ease and more
>>>>>>>>>>>>>>>>> accurate failure detection.
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> With quotas the write latencies to kafka could range from
>>>>>>>>>>>>>>>>>> a few milliseconds all the way to a tens of seconds.
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> This is actually no different from the GC problem. Most
>>>>>>>>>>>>>>>>> most of the times, the normal GC falls in the few ms range
>>>>>>>>>>>>>>>>> and there are
>>>>>>>>>>>>>>>>> many applications even at LinkedIn for which the max GC falls
>>>>>>>>>>>>>>>>> in the
>>>>>>>>>>>>>>>>> multiple seconds range. Note that it also can't be predicted,
>>>>>>>>>>>>>>>>> so has to be
>>>>>>>>>>>>>>>>> an observed value. One way or the other, you have to observe
>>>>>>>>>>>>>>>>> what this
>>>>>>>>>>>>>>>>> acceptable "max" is for your application and then set the
>>>>>>>>>>>>>>>>> appropriate
>>>>>>>>>>>>>>>>> timeouts.
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> Since this is not something that can be automated, this is
>>>>>>>>>>>>>>>>> a config that the application owner has to set based on the
>>>>>>>>>>>>>>>>> expected
>>>>>>>>>>>>>>>>> behavior of their application. Not wanting to do that leads
>>>>>>>>>>>>>>>>> to ending up
>>>>>>>>>>>>>>>>> with bad consumption semantics where the application process
>>>>>>>>>>>>>>>>> continues to
>>>>>>>>>>>>>>>>> be part of a group owning partitions but not consuming since
>>>>>>>>>>>>>>>>> it has halted
>>>>>>>>>>>>>>>>> due to a problem. The fact that the design requires them to
>>>>>>>>>>>>>>>>> express that in
>>>>>>>>>>>>>>>>> poll() frequency or not doesn't change the fact that the
>>>>>>>>>>>>>>>>> application owner
>>>>>>>>>>>>>>>>> has to go through the process of measuring and then defining
>>>>>>>>>>>>>>>>> this "max".
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> The reverse where they don't do this and the application
>>>>>>>>>>>>>>>>> remains in the group despite being dead is super confusing
>>>>>>>>>>>>>>>>> and frustrating
>>>>>>>>>>>>>>>>> too. So the due diligence up front is actually worth. And as
>>>>>>>>>>>>>>>>> long as the
>>>>>>>>>>>>>>>>> poll() latency and processing latency can be monitored, it
>>>>>>>>>>>>>>>>> should be easy
>>>>>>>>>>>>>>>>> to tell the reason for a rebalance, whether that is valid or
>>>>>>>>>>>>>>>>> not and how
>>>>>>>>>>>>>>>>> that should be tuned.
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> As for the wrapper, KIP-28 is the wrapper in open source
>>>>>>>>>>>>>>>>> that will hide this complexity and I agree that LI is
>>>>>>>>>>>>>>>>> unblocked since you
>>>>>>>>>>>>>>>>> can do this in TrackerConsumer in the meantime.
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> Thanks,
>>>>>>>>>>>>>>>>> Neha
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> On Sat, Jul 25, 2015 at 4:30 PM, Kartik Paramasivam <
>>>>>>>>>>>>>>>>> kparamasi...@linkedin.com> wrote:
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> For commit(), I think it should hopefully be an easier
>>>>>>>>>>>>>>>>>> discussion, so maybe we can follow up when we meet up next.
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> As far as the heartbeat is concerned, I think the points
>>>>>>>>>>>>>>>>>> you discuss are all very valid.
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> GC pauses impacting the heartbeats is a real issue.
>>>>>>>>>>>>>>>>>> However there are a smaller percentage of memory hungry apps
>>>>>>>>>>>>>>>>>> that get hit
>>>>>>>>>>>>>>>>>> by it.
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> The broader issue whereby even if the heartbeats are
>>>>>>>>>>>>>>>>>> healthy, the app might not be behaving correctly is also
>>>>>>>>>>>>>>>>>> real. If the app
>>>>>>>>>>>>>>>>>> is calling poll() then the probability that the app is
>>>>>>>>>>>>>>>>>> healthy is surely
>>>>>>>>>>>>>>>>>> higher. But this again isn't an absolute measure that the
>>>>>>>>>>>>>>>>>> app is
>>>>>>>>>>>>>>>>>> processing correctly.
>>>>>>>>>>>>>>>>>> In other cases the app might have even died in which case
>>>>>>>>>>>>>>>>>> this discussion is moot.
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> Point being that the only absolute way to really detect
>>>>>>>>>>>>>>>>>> that an app is healthy is to monitor lag. If the lag
>>>>>>>>>>>>>>>>>> increases then for
>>>>>>>>>>>>>>>>>> sure something is wrong.
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> The proposal seems to be that the application needs to
>>>>>>>>>>>>>>>>>> tune their session timeout based on the 99tile of the time
>>>>>>>>>>>>>>>>>> they take to
>>>>>>>>>>>>>>>>>> process events after every poll. This turns out is a
>>>>>>>>>>>>>>>>>> nontrivial thing to
>>>>>>>>>>>>>>>>>> do for an application todo. To start with when an
>>>>>>>>>>>>>>>>>> application is new their
>>>>>>>>>>>>>>>>>> data is going to be based on tests that they have done on
>>>>>>>>>>>>>>>>>> synthetic data.
>>>>>>>>>>>>>>>>>> This often times doesn't represent what they will see in
>>>>>>>>>>>>>>>>>> production. Once
>>>>>>>>>>>>>>>>>> the app is in production their processing latencies will
>>>>>>>>>>>>>>>>>> potentially vary
>>>>>>>>>>>>>>>>>> over time. It is extremely unlikely that the application
>>>>>>>>>>>>>>>>>> owner does a
>>>>>>>>>>>>>>>>>> careful job of monitoring the 99tile of latencies over time
>>>>>>>>>>>>>>>>>> and readjust
>>>>>>>>>>>>>>>>>> the settings. Often times the latencies vary because of
>>>>>>>>>>>>>>>>>> variance is other
>>>>>>>>>>>>>>>>>> services that are called by the consumer as part of
>>>>>>>>>>>>>>>>>> processing the events.
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> Case in point would be a simple app which reads events
>>>>>>>>>>>>>>>>>> and writes to Kafka. With quotas the write latencies to
>>>>>>>>>>>>>>>>>> kafka could range
>>>>>>>>>>>>>>>>>> from a few milliseconds all the way to a tens of seconds.
>>>>>>>>>>>>>>>>>> As the scale of
>>>>>>>>>>>>>>>>>> processing for an app increasing the app or that 'user'
>>>>>>>>>>>>>>>>>> could now get
>>>>>>>>>>>>>>>>>> quotaed. Instead of slowing down gracefully unless the
>>>>>>>>>>>>>>>>>> application owner
>>>>>>>>>>>>>>>>>> has carefully tuned the timeout, now we are looking at a
>>>>>>>>>>>>>>>>>> potential outage
>>>>>>>>>>>>>>>>>> where the app could get hit by constant rebalances.
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> If we expose the pause() Api then It is possible for us
>>>>>>>>>>>>>>>>>> to take care of this in the linkedin wrapper. Whereby we
>>>>>>>>>>>>>>>>>> would keep
>>>>>>>>>>>>>>>>>> calling poll on a separate thread periodically and enqueue
>>>>>>>>>>>>>>>>>> the messages.
>>>>>>>>>>>>>>>>>> When the queue is full we would call pause().
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> In essence we can work around it in LinkedIn, however I
>>>>>>>>>>>>>>>>>> think it is vastly better if we address this in the Api as
>>>>>>>>>>>>>>>>>> every major
>>>>>>>>>>>>>>>>>> customer will eventually be pained by it.
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> Kartik
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> On Jul 24, 2015, at 10:08 PM, Jay Kreps <j...@confluent.io>
>>>>>>>>>>>>>>>>>> wrote:
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> Hey guys,
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> Happy to discuss. I agree there may be some rough edges
>>>>>>>>>>>>>>>>>> and now is definitely the time to clean them up.
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> I'm pretty reluctant to change the threading model or
>>>>>>>>>>>>>>>>>> undergo a big api redesign at this point beyond the group
>>>>>>>>>>>>>>>>>> management stuff
>>>>>>>>>>>>>>>>>> we've discussed in the context of Samza/copycat which is
>>>>>>>>>>>>>>>>>> already a big
>>>>>>>>>>>>>>>>>> effort.
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> Overall I agree that we have done a poor job of
>>>>>>>>>>>>>>>>>> documenting which apis block and which don't and when people
>>>>>>>>>>>>>>>>>> are surprised
>>>>>>>>>>>>>>>>>> because we haven't labeled something that will be
>>>>>>>>>>>>>>>>>> unintuitive. But the
>>>>>>>>>>>>>>>>>> overall style of poll/select-based apis is quite common in
>>>>>>>>>>>>>>>>>> programming
>>>>>>>>>>>>>>>>>> going back to unix select so I don't think it's beyond
>>>>>>>>>>>>>>>>>> people if explained
>>>>>>>>>>>>>>>>>> well (after all we need to mix sync and async apis and if we
>>>>>>>>>>>>>>>>>> don't say
>>>>>>>>>>>>>>>>>> which is which any scheme will be confusing).
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> For what it's worth the experience with this api has
>>>>>>>>>>>>>>>>>> actually been about 1000x better than the issues people had
>>>>>>>>>>>>>>>>>> around
>>>>>>>>>>>>>>>>>> intuitiveness with the high-level api. The crazy blocking
>>>>>>>>>>>>>>>>>> iterator,
>>>>>>>>>>>>>>>>>> impossible internal queue sizing, baroque threading model,
>>>>>>>>>>>>>>>>>> etc have all
>>>>>>>>>>>>>>>>>> caused endless amounts of anger. Not to mention that that
>>>>>>>>>>>>>>>>>> client
>>>>>>>>>>>>>>>>>> effectively disqualifies about 50% of the use cases people
>>>>>>>>>>>>>>>>>> want to try to
>>>>>>>>>>>>>>>>>> use it for (plus I regularly hear people tell me they've
>>>>>>>>>>>>>>>>>> heard not to use
>>>>>>>>>>>>>>>>>> it at all for various reasons ranging from data loss to lack
>>>>>>>>>>>>>>>>>> of features).
>>>>>>>>>>>>>>>>>> It's important to have that context when people need to
>>>>>>>>>>>>>>>>>> switch and they say
>>>>>>>>>>>>>>>>>> "oh the old way was so simple and the new way complex!" :-)
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> Let me give some context related to your points, based on
>>>>>>>>>>>>>>>>>> our previous discussions:
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> For commit, let's discuss, that is easy either way.
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> The motivation for avoiding additional threading was
>>>>>>>>>>>>>>>>>> two-fold. First this client is really intended to be the
>>>>>>>>>>>>>>>>>> lowest level
>>>>>>>>>>>>>>>>>> client. There are many, many possible higher level
>>>>>>>>>>>>>>>>>> processing abstractions.
>>>>>>>>>>>>>>>>>> One thing we found to be a big problem with the high-level
>>>>>>>>>>>>>>>>>> client was that
>>>>>>>>>>>>>>>>>> it coupled things everyone must have--failover, etc--with
>>>>>>>>>>>>>>>>>> things that are
>>>>>>>>>>>>>>>>>> different in each use case like the appropriate threading
>>>>>>>>>>>>>>>>>> model. If you do
>>>>>>>>>>>>>>>>>> this you need to also maintain a thread free low-level
>>>>>>>>>>>>>>>>>> consumer api for
>>>>>>>>>>>>>>>>>> people to get around whatever you have done.
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> The second reason was that the internal threading in the
>>>>>>>>>>>>>>>>>> client became quite complex. The answer with threading is
>>>>>>>>>>>>>>>>>> always that "it
>>>>>>>>>>>>>>>>>> won't be complex this time", but it always is.
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> For the heartbeat you correctly describe the downside to
>>>>>>>>>>>>>>>>>> coupling heartbeat with poll--the contract is that the
>>>>>>>>>>>>>>>>>> application must
>>>>>>>>>>>>>>>>>> regularly consume to be considered an active consumer. This
>>>>>>>>>>>>>>>>>> allows the
>>>>>>>>>>>>>>>>>> possibility of false positive failure detections. However
>>>>>>>>>>>>>>>>>> it's important to
>>>>>>>>>>>>>>>>>> understand the downside of the alternative. If you do
>>>>>>>>>>>>>>>>>> background polling a
>>>>>>>>>>>>>>>>>> consumer is considered active as long as it isn't shutdown.
>>>>>>>>>>>>>>>>>> This leads to
>>>>>>>>>>>>>>>>>> all kinds of active consumers that aren't consuming because
>>>>>>>>>>>>>>>>>> they have
>>>>>>>>>>>>>>>>>> leaked or otherwise stopped but are still claiming
>>>>>>>>>>>>>>>>>> partitions and
>>>>>>>>>>>>>>>>>> heart-beating. This failure mode is actually far far worse.
>>>>>>>>>>>>>>>>>> If you allow
>>>>>>>>>>>>>>>>>> false positives the user sees the frequent rebalances and
>>>>>>>>>>>>>>>>>> knows they aren't
>>>>>>>>>>>>>>>>>> consuming frequently enough to be considered active but if
>>>>>>>>>>>>>>>>>> you allows false
>>>>>>>>>>>>>>>>>> negatives you end up having weeks go by before someone
>>>>>>>>>>>>>>>>>> notices that a
>>>>>>>>>>>>>>>>>> partition has been unconsumed the whole time at which point
>>>>>>>>>>>>>>>>>> the data is
>>>>>>>>>>>>>>>>>> gone. Plus of course even if you do this you still have
>>>>>>>>>>>>>>>>>> regular false
>>>>>>>>>>>>>>>>>> positives anyway from GC pauses (as now). We discussed this
>>>>>>>>>>>>>>>>>> in some depth
>>>>>>>>>>>>>>>>>> at the time and decided that it is better to have the
>>>>>>>>>>>>>>>>>> liveness notion tied
>>>>>>>>>>>>>>>>>> to *actual* consumption which is the actual definition
>>>>>>>>>>>>>>>>>> of liveness.
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> Cheers,
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> -Jay
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> On Fri, Jul 24, 2015 at 5:35 PM, Onur Karaman <
>>>>>>>>>>>>>>>>>> okara...@linkedin.com> wrote:
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> Hi Confluent Team.
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> There has recently been a lot of open source activity
>>>>>>>>>>>>>>>>>>> regarding the new KafkaConsumer:
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> https://issues.apache.org/jira/browse/KAFKA-2123
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> https://issues.apache.org/jira/browse/KAFKA-2350
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> https://issues.apache.org/jira/browse/KAFKA-2359
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> http://mail-archives.apache.org/mod_mbox/kafka-users/201507.mbox/%3ccaauywg_pwbs3hsevnp5rccmpvqbaamap+zgn8fh+woelvt_...@mail.gmail.com%3E
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> We’ve explained the KafkaConsumer API to the Databus,
>>>>>>>>>>>>>>>>>>> Samza, and some other teams and we got similar feedback.
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> To summarize the feedback we received from other teams:
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> 1.
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> The current behavior is not intuitive. For example,
>>>>>>>>>>>>>>>>>>> KafkaConsumer.poll drives everything. The other methods
>>>>>>>>>>>>>>>>>>> like subscribe,
>>>>>>>>>>>>>>>>>>> unsubscribe, seek, commit(async) don’t do anything
>>>>>>>>>>>>>>>>>>> without a
>>>>>>>>>>>>>>>>>>> KafkaConsumer.poll call.
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> 1.
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> The semantics of a commit() call should be
>>>>>>>>>>>>>>>>>>> consistent between sync and async operations. Currently,
>>>>>>>>>>>>>>>>>>> sync commit is a
>>>>>>>>>>>>>>>>>>> blocking call which actually sends out an
>>>>>>>>>>>>>>>>>>> OffsetCommitRequest and waits for
>>>>>>>>>>>>>>>>>>> the response upon the user’s KafkaConsumer.commit call.
>>>>>>>>>>>>>>>>>>> However, the async
>>>>>>>>>>>>>>>>>>> commit is a nonblocking call which just queues up the
>>>>>>>>>>>>>>>>>>> OffsetCommitRequest.
>>>>>>>>>>>>>>>>>>> The request itself is later sent out in the next poll.
>>>>>>>>>>>>>>>>>>> The teams we talked
>>>>>>>>>>>>>>>>>>> to found this misleading.
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> 1.
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> Heartbeats are dependent on user application
>>>>>>>>>>>>>>>>>>> behavior (i.e. user applications calling poll). This can
>>>>>>>>>>>>>>>>>>> be a big problem
>>>>>>>>>>>>>>>>>>> as we don’t control how different applications behave.
>>>>>>>>>>>>>>>>>>> For example, we
>>>>>>>>>>>>>>>>>>> might have an application which reads from Kafka and
>>>>>>>>>>>>>>>>>>> writes to Espresso. If
>>>>>>>>>>>>>>>>>>> Espresso is slow for whatever reason, then in rebalances
>>>>>>>>>>>>>>>>>>> could happen.
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> Generally speaking, we feel that the current
>>>>>>>>>>>>>>>>>>> KafkaConsumer API design is more of a wrapping around the
>>>>>>>>>>>>>>>>>>> old simple
>>>>>>>>>>>>>>>>>>> consumer, i.e. in old consumer we ask users to deal with
>>>>>>>>>>>>>>>>>>> raw protocols and
>>>>>>>>>>>>>>>>>>> error handlings while in KafkaConsumer we do that for
>>>>>>>>>>>>>>>>>>> users. However, for
>>>>>>>>>>>>>>>>>>> old high level consumer users (which are the majority of
>>>>>>>>>>>>>>>>>>> users), the
>>>>>>>>>>>>>>>>>>> experience is a noticeable regression. The old high level
>>>>>>>>>>>>>>>>>>> consumer
>>>>>>>>>>>>>>>>>>> interface is simple and easy to use for end user, while
>>>>>>>>>>>>>>>>>>> KafkaConsumer
>>>>>>>>>>>>>>>>>>> requires users to be aware of many underlying details and
>>>>>>>>>>>>>>>>>>> is becoming
>>>>>>>>>>>>>>>>>>> prohibitive for users to adopt. This is hinted by the
>>>>>>>>>>>>>>>>>>> javadoc growing
>>>>>>>>>>>>>>>>>>> bigger and bigger.
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> We think it's getting to the point where we should take
>>>>>>>>>>>>>>>>>>> a step back and look at the big picture.
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> The current state of KafkaConsumer is that it's
>>>>>>>>>>>>>>>>>>> single-threaded. There's one big KafkaConsumer.poll called
>>>>>>>>>>>>>>>>>>> by the user
>>>>>>>>>>>>>>>>>>> which pretty much drives everything:
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> - data fetches
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> - heartbeats
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> - join groups (new consumer joining a group, topic
>>>>>>>>>>>>>>>>>>> subscription changes, reacting to group rebalance)
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> - async offset commits
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> - executing callbacks
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> Given that the selector's poll is being driven by the
>>>>>>>>>>>>>>>>>>> end user, this ends up making us educate users on NIO and
>>>>>>>>>>>>>>>>>>> the consequences
>>>>>>>>>>>>>>>>>>> of not calling KafkaConsumer.poll frequently enough:
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> - Coordinator will mark the consumer dead
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> - async commits won't send
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> - callbacks won't fire
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> More generally speaking, there are many surprises with
>>>>>>>>>>>>>>>>>>> the current KafkaConsumer implementation.
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> Here's what we consider to be the goals of KafkaConsumer:
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> - NIO
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> - ability to commit, manipulate offsets, and consume
>>>>>>>>>>>>>>>>>>> messages
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> - a way to subscribe to topics(auto group management) or
>>>>>>>>>>>>>>>>>>> partitions(explicit group management)
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> - no surprises in the user experience
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> The last point is the big one that we think we aren't
>>>>>>>>>>>>>>>>>>> hitting. We think the most important example is that there
>>>>>>>>>>>>>>>>>>> should be no
>>>>>>>>>>>>>>>>>>> requirement from the end user to consistently
>>>>>>>>>>>>>>>>>>> KafkaConsumer.poll in order
>>>>>>>>>>>>>>>>>>> for all of the above tasks to happen. We think it would be
>>>>>>>>>>>>>>>>>>> better to split
>>>>>>>>>>>>>>>>>>> those tasks into tasks that should not rely on
>>>>>>>>>>>>>>>>>>> KafkaConsumer.poll and tasks
>>>>>>>>>>>>>>>>>>> that should rely on KafkaConsumer.poll.
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> Tasks that should not rely on KafkaConsumer.poll:
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> - heartbeats
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> - join groups
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> - commits
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> - executing callbacks
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> Only data fetches should rely on KafkaConsumer.poll
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> This would help reduce the amount of surprises to the
>>>>>>>>>>>>>>>>>>> end user.
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> We’ve sketched out a proposal and we’ll send it out to
>>>>>>>>>>>>>>>>>>> you guys early next week. We’d like to meet up with you at
>>>>>>>>>>>>>>>>>>> LinkedIn on *July
>>>>>>>>>>>>>>>>>>> 31, 2015* so we can talk about it before proposing it
>>>>>>>>>>>>>>>>>>> to open source.
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> Regards,
>>>>>>>>>>>>>>>>>>> LinkedIn Kafka Dev Team
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> --
>>>>>>>>>>>>>>>>> Thanks,
>>>>>>>>>>>>>>>>> Neha
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>> --
>>>>>>>>>>> Thanks,
>>>>>>>>>>> Neha
>>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>> --
>>>>>>>>>> -- Guozhang
>>>>>>>>>>
>>>>>>>>>
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> --
>>>>>>>>> Thanks,
>>>>>>>>> Ewen
>>>>>>>>>
>>>>>>>>
>>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>> --
>>>>>>> Thanks,
>>>>>>> Neha
>>>>>>>
>>>>>>
>>>>>>
>>>>>
>>>>>
>>>>> --
>>>>> Thanks,
>>>>> Neha
>>>>>
>>>>
>>>>
>>>
>>
>
