Thanks for the suggestion, Ismael. I updated the KIP. -Jason
On Wed, Jan 6, 2016 at 6:57 AM, Ismael Juma <ism...@juma.me.uk> wrote: > Thanks Jason. I read the KIP and it makes sense to me. A minor suggestion: > in the "Ensuring Fair Consumption" section, there are 3 paragraphs with 2 > examples (2 partitions with 100 max.poll.records and 3 partitions with 30 > max.poll.records). I think you could simplify this by using one of the > examples in the 3 paragraphs. > > Ismael > > On Tue, Jan 5, 2016 at 7:32 PM, Jason Gustafson <ja...@confluent.io> > wrote: > > > I've updated the KIP with some implementation details. I also added more > > discussion on the heartbeat() alternative. The short answer for why we > > rejected this API is that it doesn't seem to work well with offset > commits. > > This would tend to make correct usage complicated and difficult to > explain. > > Additionally, we don't see any clear advantages over having a way to set > > the max records. For example, using max.records=1 would be equivalent to > > invoking heartbeat() on each iteration of the message processing loop. > > > > Going back to the discussion on whether we should use a configuration > value > > or overload poll(), I'm leaning toward the configuration option mainly > for > > compatibility and to keep the KafkaConsumer API from getting any more > > complex. Also, as others have mentioned, it seems reasonable to want to > > tune this setting in the same place that the session timeout and > heartbeat > > interval are configured. I still feel a little uncomfortable with the > need > > to do a lot of configuration tuning to get the consumer working for a > > particular environment, but hopefully the defaults are conservative > enough > > that most users won't need to. However, if it remains a problem, then we > > could still look into better options for managing the size of batches > > including overloading poll() with a max records argument or possibly by > > implementing a batch scaling algorithm internally. > > > > -Jason > > > > > > On Mon, Jan 4, 2016 at 12:18 PM, Jason Gustafson <ja...@confluent.io> > > wrote: > > > > > Hi Cliff, > > > > > > I think we're all agreed that the current contract of poll() should be > > > kept. The consumer wouldn't wait for max messages to become available > in > > > this proposal; it would only sure that it never returns more than max > > > messages. > > > > > > -Jason > > > > > > On Mon, Jan 4, 2016 at 11:52 AM, Cliff Rhyne <crh...@signal.co> wrote: > > > > > >> Instead of a heartbeat, I'd prefer poll() to return whatever messages > > the > > >> client has. Either a) I don't care if I get less than my max message > > >> limit > > >> or b) I do care and will set a larger timeout. Case B is less common > > than > > >> A and is fairly easy to handle in the application's code. > > >> > > >> On Mon, Jan 4, 2016 at 1:47 PM, Gwen Shapira <g...@confluent.io> > wrote: > > >> > > >> > 1. Agree that TCP window style scaling will be cool. I'll try to > think > > >> of a > > >> > good excuse to use it ;) > > >> > > > >> > 2. I'm very concerned about the challenges of getting the timeouts, > > >> > hearbeats and max messages right. > > >> > > > >> > Another option could be to expose "heartbeat" API to consumers. If > my > > >> app > > >> > is still processing data but is still alive, it could initiate a > > >> heartbeat > > >> > to signal its alive without having to handle additional messages. > > >> > > > >> > I don't know if this improves more than it complicates though :( > > >> > > > >> > On Mon, Jan 4, 2016 at 11:40 AM, Jason Gustafson < > ja...@confluent.io> > > >> > wrote: > > >> > > > >> > > Hey Gwen, > > >> > > > > >> > > I was thinking along the lines of TCP window scaling in order to > > >> > > dynamically find a good consumption rate. Basically you'd start > off > > >> > > consuming say 100 records and you'd let it increase until the > > >> consumption > > >> > > took longer than half the session timeout (for example). You > /might/ > > >> be > > >> > > able to achieve the same thing using pause/resume, but it would > be a > > >> lot > > >> > > trickier since you have to do it at the granularity of partitions. > > But > > >> > > yeah, database write performance doesn't always scale in a > > predictable > > >> > > enough way to accommodate this, so I'm not sure how useful it > would > > >> be in > > >> > > practice. It might also be more difficult to implement since it > > >> wouldn't > > >> > be > > >> > > as clear when to initiate the next fetch. With a static setting, > the > > >> > > consumer knows exactly how many records will be returned on the > next > > >> call > > >> > > to poll() and can send fetches accordingly. > > >> > > > > >> > > On the other hand, I do feel a little wary of the need to tune the > > >> > session > > >> > > timeout and max messages though since these settings might depend > on > > >> the > > >> > > environment that the consumer is deployed in. It wouldn't be a big > > >> deal > > >> > if > > >> > > the impact was relatively minor, but getting them wrong can cause > a > > >> lot > > >> > of > > >> > > rebalance churn which could keep the consumer from making any > > >> progress. > > >> > > It's not a particularly graceful failure. > > >> > > > > >> > > -Jason > > >> > > > > >> > > On Mon, Jan 4, 2016 at 10:49 AM, Gwen Shapira <g...@confluent.io> > > >> wrote: > > >> > > > > >> > > > I can't speak to all use-cases, but for the database one, I > think > > >> > > > pause-resume will be necessary in any case, and therefore > dynamic > > >> batch > > >> > > > sizes are not needed. > > >> > > > > > >> > > > Databases are really unexpected regarding response times - load > > and > > >> > > locking > > >> > > > can affect this. I'm not sure there's a good way to know you are > > >> going > > >> > > into > > >> > > > rebalance hell before it is too late. So if I were writing code > > that > > >> > > > updates an RDBMS based on Kafka, I'd pick a reasonable batch > size > > >> (say > > >> > > 5000 > > >> > > > records), and basically pause, batch-insert all records, commit > > and > > >> > > resume. > > >> > > > > > >> > > > Does that make sense? > > >> > > > > > >> > > > On Mon, Jan 4, 2016 at 10:37 AM, Jason Gustafson < > > >> ja...@confluent.io> > > >> > > > wrote: > > >> > > > > > >> > > > > Gwen and Ismael, > > >> > > > > > > >> > > > > I agree the configuration option is probably the way to go, > but > > I > > >> was > > >> > > > > wondering whether there would be cases where it made sense to > > let > > >> the > > >> > > > > consumer dynamically set max messages to adjust for downstream > > >> > > slowness. > > >> > > > > For example, if the consumer is writing consumed records to > > >> another > > >> > > > > database, and that database is experiencing heavier than > > expected > > >> > load, > > >> > > > > then the consumer could halve its current max messages in > order > > to > > >> > > adapt > > >> > > > > without risking rebalance hell. It could then increase max > > >> messages > > >> > as > > >> > > > the > > >> > > > > load on the database decreases. It's basically an easier way > to > > >> > handle > > >> > > > flow > > >> > > > > control than we provide with pause/resume. > > >> > > > > > > >> > > > > -Jason > > >> > > > > > > >> > > > > On Mon, Jan 4, 2016 at 9:46 AM, Gwen Shapira < > g...@confluent.io > > > > > >> > > wrote: > > >> > > > > > > >> > > > > > The wiki you pointed to is no longer maintained and fell out > > of > > >> > sync > > >> > > > with > > >> > > > > > the code and protocol. > > >> > > > > > > > >> > > > > > You may want to refer to: > > >> > > > > > > > >> > > > > > > > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > > https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol > > >> > > > > > > > >> > > > > > On Mon, Jan 4, 2016 at 4:38 AM, Jens Rantil < > > >> jens.ran...@tink.se> > > >> > > > wrote: > > >> > > > > > > > >> > > > > > > Hi guys, > > >> > > > > > > > > >> > > > > > > I realized I never thanked yall for your input - thanks! > > >> > > > > > > Jason: I apologize for assuming your stance on the issue! > > >> Feels > > >> > > like > > >> > > > we > > >> > > > > > all > > >> > > > > > > agreed on the solution. +1 > > >> > > > > > > > > >> > > > > > > Follow-up: Jason made a point about defining prefetch and > > >> > fairness > > >> > > > > > > behaviour in the KIP. I am now working on putting that > down > > in > > >> > > > writing. > > >> > > > > > To > > >> > > > > > > do be able to do this I think I need to understand the > > current > > >> > > > prefetch > > >> > > > > > > behaviour in the new consumer API (0.9) a bit better. Some > > >> > specific > > >> > > > > > > questions: > > >> > > > > > > > > >> > > > > > > - How does a specific consumer balance incoming > messages > > >> from > > >> > > > > multiple > > >> > > > > > > partitions? Is the consumer simply issuing Multi-Fetch > > >> > > requests[1] > > >> > > > > for > > >> > > > > > > the > > >> > > > > > > consumed assigned partitions of the relevant topics? Or > > is > > >> the > > >> > > > > > consumer > > >> > > > > > > fetching from one partition at a time and balancing > > between > > >> > them > > >> > > > > > > internally? That is, is the responsibility of partition > > >> > > balancing > > >> > > > > (and > > >> > > > > > > fairness) on the broker side or consumer side? > > >> > > > > > > - Is the above documented somewhere? > > >> > > > > > > > > >> > > > > > > [1] > > >> > > > > > > > > >> > > > > > > > > >> > > > > > > > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > > https://cwiki.apache.org/confluence/display/KAFKA/Writing+a+Driver+for+Kafka > > >> > > > > > > , > > >> > > > > > > see "Multi-Fetch". > > >> > > > > > > > > >> > > > > > > Thanks, > > >> > > > > > > Jens > > >> > > > > > > > > >> > > > > > > On Wed, Dec 23, 2015 at 2:44 AM, Ismael Juma < > > >> ism...@juma.me.uk> > > >> > > > > wrote: > > >> > > > > > > > > >> > > > > > > > On Wed, Dec 23, 2015 at 1:24 AM, Gwen Shapira < > > >> > g...@confluent.io > > >> > > > > > >> > > > > > wrote: > > >> > > > > > > > > > >> > > > > > > > > Given the background, it sounds like you'll generally > > want > > >> > each > > >> > > > > call > > >> > > > > > to > > >> > > > > > > > > poll() to return the same number of events (which is > the > > >> > number > > >> > > > you > > >> > > > > > > > planned > > >> > > > > > > > > on having enough memory / time for). It also sounds > like > > >> > tuning > > >> > > > the > > >> > > > > > > > number > > >> > > > > > > > > of events will be closely tied to tuning the session > > >> timeout. > > >> > > > That > > >> > > > > > is - > > >> > > > > > > > if > > >> > > > > > > > > I choose to lower the session timeout for some > reason, I > > >> will > > >> > > > have > > >> > > > > to > > >> > > > > > > > > modify the number of records returning too. > > >> > > > > > > > > > > >> > > > > > > > > If those assumptions are correct, I think a > > configuration > > >> > makes > > >> > > > > more > > >> > > > > > > > sense. > > >> > > > > > > > > 1. We are unlikely to want this parameter to be change > > at > > >> the > > >> > > > > > lifetime > > >> > > > > > > of > > >> > > > > > > > > the consumer > > >> > > > > > > > > 2. The correct value is tied to another configuration > > >> > > parameter, > > >> > > > so > > >> > > > > > > they > > >> > > > > > > > > will be controlled together. > > >> > > > > > > > > > > >> > > > > > > > > > >> > > > > > > > I was thinking the same thing. > > >> > > > > > > > > > >> > > > > > > > Ismael > > >> > > > > > > > > > >> > > > > > > > > >> > > > > > > > > >> > > > > > > > > >> > > > > > > -- > > >> > > > > > > Jens Rantil > > >> > > > > > > Backend engineer > > >> > > > > > > Tink AB > > >> > > > > > > > > >> > > > > > > Email: jens.ran...@tink.se > > >> > > > > > > Phone: +46 708 84 18 32 > > >> > > > > > > Web: www.tink.se > > >> > > > > > > > > >> > > > > > > Facebook <https://www.facebook.com/#!/tink.se> Linkedin > > >> > > > > > > < > > >> > > > > > > > > >> > > > > > > > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > > http://www.linkedin.com/company/2735919?trk=vsrp_companies_res_photo&trkInfo=VSRPsearchId%3A1057023381369207406670%2CVSRPtargetId%3A2735919%2CVSRPcmpt%3Aprimary > > >> > > > > > > > > > >> > > > > > > Twitter <https://twitter.com/tink> > > >> > > > > > > > > >> > > > > > > > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > >> > > >> > > >> -- > > >> Cliff Rhyne > > >> Software Engineering Lead > > >> e: crh...@signal.co > > >> signal.co > > >> ________________________ > > >> > > >> Cut Through the Noise > > >> > > >> This e-mail and any files transmitted with it are for the sole use of > > the > > >> intended recipient(s) and may contain confidential and privileged > > >> information. Any unauthorized use of this email is strictly > prohibited. > > >> ©2015 Signal. All rights reserved. > > >> > > > > > > > > >
