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
