Hey John, I know I'm a bit late to this party but just for the record, I don't think it's *totally *unreasonable for a user to take up the "poll on max timeout and assume some records will be returned" approach. And I also can imagine plenty of manually-assigned consumers implemented doing exactly that.
It's too bad that we're hitting this during the 2.8 release. If we were having this discussion in the context of 3.0, then I'd say go for it, since it's a breaking change that would just require some modification to the applications' poll loop. A good analogy here seems to be spurious wakeups -- you generally assume that a waiting thread has woken up due to a notify event in another thread, but the docs always make it very clear up front that this can happen "spuriously" and therefore you need to recheck whatever condition you were waiting on before assuming the thread should proceed. Since we *didn't* document this possibility up front in the case of poll(), it seems unfair to suddenly change the behavior in a supposedly non-breaking release. Imagine how many programs would break if spurious wakeups were suddenly introduced in a release, rather than warned about from the get-go (not a perfect analogy, far more programs rely on wait/notify than on poll() returning records, but I think the point still stands. For the record, I also agree with Ismael that a config doesn't feel ideal. There are already enough configs to present a steep learning curve, so I would avoid adding one more wherever possible. And it does indeed seem possible to avoid here, since it's really just a boolean flag (rather than a semi-unbounded space, eg max.poll.interval.ms, or a constant value, eg group.id, where a config does feel appropriate). Given all that, I would personally advocate for the pollOptions overload. The obvious advantages here are: 1) it's more future-proof, in that we can avoid having a similar discussion if/when we want to consider other semantics changes to poll which some users may want while others would not 2) it leaves the door open to using poll with either semantics in a single consumer. I doubt that's going to be very common in terms of the specific option we're discussing here, but it may be more useful for other options we may add in the future Just my 2 cents. But if the pollOptions proposal would really add so much additional work that it would cause the 2.8 release to be significantly delayed, then that's worth taking into account as well. On Wed, Feb 10, 2021 at 9:35 AM John Roesler <vvcep...@apache.org> wrote: > Hello again, all. > > I have submitted the PR: > https://github.com/apache/kafka/pull/10096 > > Ismael chimed in on the PR review to indicate that the > config approach may not be desirable. > > How strongly do we feel that the behavior change is > unacceptable? It seems like most of the people involved felt > the behavior change is ok (although the docs were wrong). > > The arguments against the behavior change were plausible, > but hypothetical. > > Can everyone take a look at the PR and weigh in on whether > the complexity of an extra config option is really worth it > in this case? > > I have to confess I'm currently leaning more toward dropping > the config and going back to the behavior change, while > correcting the docs and the system test. > > While we are wavering on this point, the system tests > continue to fail, and the 2.8.0 release is blocked. We > should aim to make a call today. > > Thanks all, > -John > > On Fri, 2021-02-05 at 15:31 -0800, Guozhang Wang wrote: > > Thanks everyone for chiming in here! I'd also prefer the config approach > if > > compared with API changes. > > > > On Fri, Feb 5, 2021 at 3:18 PM Bill Bejeck <bbej...@gmail.com> wrote: > > > > > I meant to chime in earlier. > > > > > > I also like the `PollOptions` idea, but I have to agree that the config > > > option would be the least disruptive approach. > > > > > > Thanks, > > > Bill > > > > > > On Fri, Feb 5, 2021 at 6:12 PM John Roesler <vvcep...@apache.org> > wrote: > > > > > > > Thanks, all! > > > > > > > > It seems that the config I proposed is a solution that > > > > everyone can be happy with, so I will go ahead with a PR to > > > > fix that. > > > > > > > > I'll update the KIP after a round of PR reviews, in case > > > > there are new concerns that arise. > > > > > > > > Thanks, > > > > -John > > > > > > > > On Fri, 2021-02-05 at 15:07 -0800, Matthias J. Sax wrote: > > > > > Thanks for providing more details. > > > > > > > > > > Adding a config might be the way a least resistance... I am fine > with > > > > that. > > > > > > > > > > -Matthias > > > > > > > > > > On 2/4/21 9:42 AM, Chia-Ping Tsai wrote: > > > > > > > vvvvvvvvvvvvvvvvvvv > > > > > > > long_poll.mode: return_on_records|return_on_response > > > > > > > > > > > > This idea LGTM. It not only makes minimum changes to current > behavior > > > > but also works for KIP-695. > > > > > > > > > > > > On 2021/02/04 16:07:11, John Roesler <vvcep...@apache.org> > wrote: > > > > > > > Hi Matthias, Chia-Ping, and Tom, > > > > > > > > > > > > > > Thanks for the thoughtful replies! > > > > > > > > > > > > > > Re: poll(~forever~) to block indefinitely on records: > > > > > > > Thanks for your dilligence, Chia-Ping. While I wouldn't > > > > > > > personally recommend for anyone to write code that blocks > > > > > > > forever on I/O, I do agree this is something that "real > > > > > > > people" may want to do. > > > > > > > > > > > > > > Just a note for the record, this approach should only be > > > > > > > used in conjunction with a manual assignment. If people are > > > > > > > using a group subscription, they're setting themselves up to > > > > > > > get kicked out of the group when there is low volume of > > > > > > > updates on the topic. And then, when they get kicked out, > > > > > > > they will never know it because they're just going to be > > > > > > > blocked in `poll()` the whole time. > > > > > > > > > > > > > > However, if you don't participate in a group and just: > > > > > > > 1 assign(partitions) > > > > > > > 2 poll(forever), > > > > > > > you should indeed expect to return from poll only when you > > > > > > > have records. > > > > > > > > > > > > > > This possibility is the turning point for me. I'd like to > > > > > > > alter my proposal to an opt-in config, detailed below. > > > > > > > > > > > > > > Re: Javadoc: > > > > > > > Thanks for pointing that out. It does seem like, if we do > > > > > > > decide to change behavior, we should adjust the Javadoc to > > > > > > > say so. That was an oversight on my part, and I daresay that > > > > > > > if I had done that initially, it would have saved Rajini > > > > > > > from having to dig into the code to pinpoint the cause of > > > > > > > those test failures. > > > > > > > > > > > > > > Re: PollOptions: > > > > > > > I actually like this option quite a bit. It seems like this > > > > > > > would be warranted if we expect someone to want to use the > > > > > > > same Consumer instance in both "return on metadata or > > > > > > > records" and "return on only records" mode. Otherwise, we > > > > > > > might as well introduce a new config. > > > > > > > > > > > > > > It also seems like the behavior I proposed in this KIP is > > > > > > > somewhat "advanced", so I could certainly see leaving it off > > > > > > > by default and offering an opt-in config. > > > > > > > > > > > > > > How does everyone feel about this opt-in config: > > > > > > > > > > > > > > vvvvvvvvvvvvvvvvvvv > > > > > > > long_poll.mode: return_on_records|return_on_response > > > > > > > > > > > > > > doc: > > > > > > > * return_on_records: (default) a call to > > > > > > > Consumer#poll(timeout) will block up to the timeout and > > > > > > > return early if records are received. > > > > > > > * return_on_response: a call to Consumer#poll(timeout) will > > > > > > > block up to the timeout and return early if any fetch > > > > > > > response is received. Use this option to get updates from > > > > > > > Consumer#metadata() even if Consumer#records() is empty. > > > > > > > ^^^^^^^^^^^^^^^^^^^ > > > > > > > > > > > > > > Thanks, > > > > > > > John > > > > > > > > > > > > > > On Thu, 2021-02-04 at 08:44 +0000, Tom Bentley wrote: > > > > > > > > Hi, > > > > > > > > > > > > > > > > The Javadoc for KafkaConsumer#poll() includes the following: > > > > > > > > > > > > > > > > * This method returns immediately if there are records > available. > > > > *Otherwise, > > > > > > > > > it will await the passed timeout.* > > > > > > > > > * If the timeout expires, an empty record set will be > returned. > > > > Note that > > > > > > > > > this method may block beyond the > > > > > > > > > * timeout in order to execute custom {@link > > > > ConsumerRebalanceListener} > > > > > > > > > callbacks. > > > > > > > > > > > > > > > > > > > > > > > > > In other words: If the method returns before the timeout > there > > > > must be > > > > > > > > records in the method result. After the timeout has passed > there > > > > may be no > > > > > > > > records. It might block for longer than the timeout. So I > think > > > > returning > > > > > > > > with empty records before at least the given timeout has > passed > > > > breaks that > > > > > > > > contract. > > > > > > > > > > > > > > > > A not-much-prettier alternative to adding a new > > > > > > > > pollForRecordsOrMetadata(Duration) method could be > overloading > > > > poll() to > > > > > > > > take an additional parameter which controlled whether an > early > > > > return with > > > > > > > > empty records was allowed. Or a `poll(PollOptions)`. In the > long > > > > run it > > > > > > > > could be a mistake to include in the method name exactly what > > > > might cause > > > > > > > > an early empty return. > > > > > > > > > > > > > > > > Kind regards, > > > > > > > > > > > > > > > > Tom > > > > > > > > > > > > > > > > > > > > > > > > On Thu, Feb 4, 2021 at 5:08 AM Chia-Ping Tsai < > > > chia7...@apache.org> > > > > wrote: > > > > > > > > > > > > > > > > > Thanks for your sharing Matthias. I agree that is indeed an > > > > anti-pattern > > > > > > > > > to assume poll() returns data or not. > > > > > > > > > > > > > > > > > > However, I check all usages of poll() in code base. There > is an > > > > > > > > > interesting use case - poll(a bigger timeout) - it implies > that > > > > callers > > > > > > > > > want to block poll()(forever) unless there are available > data. > > > > > > > > > > > > > > > > > > [1] > > > > > > > > > > > > > > > > > https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/tools/ConsoleConsumer.scala#L443 > > > > > > > > > [2] > > > > > > > > > > > > > > > > > https://github.com/apache/kafka/blob/trunk/tools/src/main/java/org/apache/kafka/tools/VerifiableConsumer.java#L232 > > > > > > > > > > > > > > > > > > Hence, I start to worry client code like aforementioned > cases > > > > get broken > > > > > > > > > due to behavior change :( > > > > > > > > > > > > > > > > > > On 2021/02/03 22:59:09, "Matthias J. Sax" < > mj...@apache.org> > > > > wrote: > > > > > > > > > > Thanks for your email John. > > > > > > > > > > > > > > > > > > > > I agree that it seems to be an anti-pattern to write code > > > that > > > > makes > > > > > > > > > > assumptions if poll() returns data or not. Thus, we > should > > > > fix-forward > > > > > > > > > > the system test from my point of view. > > > > > > > > > > > > > > > > > > > > From my understanding, the impact of KIP-695 is that we > might > > > > return > > > > > > > > > > early from poll() (ie, before the timeout passed) with no > > > > data, only if > > > > > > > > > > an empty fetch request comes back and there is no other > fetch > > > > request > > > > > > > > > > that did return data. Thus, for most cases, poll() should > > > > still return > > > > > > > > > > early and provide data. -- Thus, I have no concerns with > the > > > > slight > > > > > > > > > > behavior change. > > > > > > > > > > > > > > > > > > > > Would be good to get input from others about this > question > > > > though. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > -Matthias > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > On 2/3/21 10:06 AM, John Roesler wrote: > > > > > > > > > > > Hello again all, > > > > > > > > > > > > > > > > > > > > > > I'm resurrecting this thread to discuss an issue that > has > > > > > > > > > > > come up after merging the code for this KIP. > > > > > > > > > > > > > > > > > > > > > > The issue is that some of the system tests need to be > > > > > > > > > > > updated in the same way that this integration test > needed > > > to > > > > > > > > > > > be updated: > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://github.com/apache/kafka/pull/9836/files#diff-735dcc2179315ebd78a7c75fd21b70b0ae81b90f3d5ec761740bc80abeae891fR1875-R1888 > > > > > > > > > > > > > > > > > > > > > > This issue was reported here: > > > > > > > > > > > https://issues.apache.org/jira/browse/KAFKA-12268 > > > > > > > > > > > and there is some preliminary discussion here: > > > > > > > > > > > https://github.com/apache/kafka/pull/10022 > > > > > > > > > > > > > > > > > > > > > > First, let me offer my apologies for failing to catch > this > > > > > > > > > > > before the merge. I'm sorry that it became Rajini's > work to > > > > > > > > > > > track down the cause of the failure, when it was my > > > > > > > > > > > responsibility to ensure the feature was merged safely. > > > > > > > > > > > > > > > > > > > > > > To recap the situation: > > > > > > > > > > > Consumer#poll(Duration) will now return before the > duration > > > > > > > > > > > expires even if there are no records returned if there > is > > > > > > > > > > > some returned metadata. > > > > > > > > > > > > > > > > > > > > > > This behavior was important for KIP-695. In the > situation > > > > > > > > > > > where we get no records back for some partition, > Streams > > > > > > > > > > > needs to have the freshest possible information about > > > > > > > > > > > whether there are no new records on the broker, or > whether > > > > > > > > > > > there are records on the broker that we still need to > > > fetch. > > > > > > > > > > > If that's not clear, the KIP contains the full story. > > > > > > > > > > > > > > > > > > > > > > It's definitely a behavior change, but our rationale > was > > > > > > > > > > > that it's an acceptable behavior change. Our big > > > alternative > > > > > > > > > > > is to add a _new_ method to Consumer to > > > > > > > > > > > pollForRecordsOrMetadata(Duration) or something. > > > > > > > > > > > > > > > > > > > > > > It seems unreliable to expect the broker to return a > > > > > > > > > > > particular record within a particular timeout in > general, > > > > > > > > > > > which is what these tests are doing. The broker can > decide > > > > > > > > > > > for several reasons not to return data for a > partition, but > > > > > > > > > > > return data for another partition instead. > > > > > > > > > > > > > > > > > > > > > > It seems like the only case where you might reasonably > try > > > > > > > > > > > to rely on that is in a test, where you first write a > > > record > > > > > > > > > > > to a partition, then you assign only that one > partition to > > > a > > > > > > > > > > > consumer, then you poll on the consumer, expecting it > to > > > > > > > > > > > return the data you just wrote. > > > > > > > > > > > > > > > > > > > > > > So the $10 question here is whether we should support > this > > > > > > > > > > > apparently artificial (testing-only) use case to the > point > > > > > > > > > > > where it's worth adding a whole new method to the > Consumer > > > > > > > > > > > interface. > > > > > > > > > > > > > > > > > > > > > > Thanks all, > > > > > > > > > > > John > > > > > > > > > > > > > > > > > > > > > > On Thu, 2020-12-17 at 13:18 -0600, John Roesler wrote: > > > > > > > > > > > > Thanks Jason, > > > > > > > > > > > > > > > > > > > > > > > > We would only return the metadata for the latest > fetches. > > > > > > > > > > > > So, if someone wanted to use this to lazily maintain > a > > > > > > > > > > > > client-side metadata map for all partitions, they'd > have > > > to > > > > > > > > > > > > store it separately and merge in new updates as they > > > > arrive. > > > > > > > > > > > > > > > > > > > > > > > > This way: > > > > > > > > > > > > 1. We don't need to increase the complexity of the > client > > > > by > > > > > > > > > > > > storing that metadata > > > > > > > > > > > > 2. Users will be able to treat all returned metadata > as > > > > > > > > > > > > "fresh" without having to reason about the > timestamps. > > > > > > > > > > > > 3. All parts of the returned ConsumerRecords object > have > > > > the > > > > > > > > > > > > same lifecycle: all the data and metadata are the > results > > > > of > > > > > > > > > > > > the most recent round of fetch responses that had not > > > been > > > > > > > > > > > > previously polled. > > > > > > > > > > > > > > > > > > > > > > > > Does that seem sensible to you? I'll update the KIP > to > > > > > > > > > > > > clarify this. > > > > > > > > > > > > > > > > > > > > > > > > Thanks, > > > > > > > > > > > > -John > > > > > > > > > > > > > > > > > > > > > > > > On Wed, 2020-12-16 at 10:29 -0800, Jason Gustafson > wrote: > > > > > > > > > > > > > Hi John, > > > > > > > > > > > > > > > > > > > > > > > > > > Just one question. It wasn't very clear to me > exactly > > > > when the > > > > > > > > > metadata > > > > > > > > > > > > > would be returned in `ConsumerRecords`. Would we > > > > /always/ include the > > > > > > > > > > > > > metadata for all partitions that are assigned, or > would > > > > it be based > > > > > > > > > on the > > > > > > > > > > > > > latest fetches? > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks, > > > > > > > > > > > > > Jason > > > > > > > > > > > > > > > > > > > > > > > > > > On Fri, Dec 11, 2020 at 4:07 PM John Roesler < > > > > vvcep...@apache.org> > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks, Guozhang! > > > > > > > > > > > > > > > > > > > > > > > > > > > > All of your feedback sounds good to me. I’ll > update > > > > the KIP when I > > > > > > > > > am able. > > > > > > > > > > > > > > > > > > > > > > > > > > > > 3) I believe it is the position after the fetch, > but > > > I > > > > will > > > > > > > > > confirm. I > > > > > > > > > > > > > > think omitting position may render beginning and > end > > > > offsets > > > > > > > > > useless as > > > > > > > > > > > > > > well, which leaves only lag. That would be fine > with > > > > me, but it > > > > > > > > > also seems > > > > > > > > > > > > > > nice to supply this extra metadata since it is > well > > > > defined and > > > > > > > > > probably > > > > > > > > > > > > > > handy for others. Therefore, I’d go the route of > > > > specifying the > > > > > > > > > exact > > > > > > > > > > > > > > semantics and keeping it. > > > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks for the review, > > > > > > > > > > > > > > John > > > > > > > > > > > > > > > > > > > > > > > > > > > > On Fri, Dec 11, 2020, at 17:36, Guozhang Wang > wrote: > > > > > > > > > > > > > > > Hello John, > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks for the updates! I've made a pass on > the KIP > > > > and also the > > > > > > > > > POC PR, > > > > > > > > > > > > > > > here are some minor comments: > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > 1) nit: "receivedTimestamp" -> it seems the > > > metadata > > > > keep getting > > > > > > > > > > > > > > updated, > > > > > > > > > > > > > > > and we do not create a new object but just > update > > > > the values > > > > > > > > > in-place, so > > > > > > > > > > > > > > > maybe calling it `lastUpdateTimstamp` is > better? > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > 2) It will be great to verify in javadocs that > the > > > > new API > > > > > > > > > > > > > > > "ConsumerRecords#metadata(): > Map<TopicPartition, > > > > Metadata>" may > > > > > > > > > return a > > > > > > > > > > > > > > > superset of TopicPartitions than the existing > API > > > > that returns the > > > > > > > > > data > > > > > > > > > > > > > > by > > > > > > > > > > > > > > > partitions, in case users assume their map > > > > key-entries would > > > > > > > > > always be > > > > > > > > > > > > > > the > > > > > > > > > > > > > > > same. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > 3) The "position()" API of the call needs > better > > > > clarification: is > > > > > > > > > it the > > > > > > > > > > > > > > > current position AFTER the records are > returned, or > > > > is it BEFORE > > > > > > > > > the > > > > > > > > > > > > > > > records are returned? Personally I'd suggest > we do > > > > not include it > > > > > > > > > if it > > > > > > > > > > > > > > is > > > > > > > > > > > > > > > not used anywhere yet just to avoid possible > > > > misuage, but I'm fine > > > > > > > > > if you > > > > > > > > > > > > > > > like to keep it still; in that case just > clarify > > > its > > > > semantics. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Other than that，I'm +1 on the KIP as well ! > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Guozhang > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > On Fri, Dec 11, 2020 at 8:15 AM Walker Carlson > < > > > > > > > > > wcarl...@confluent.io> > > > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks for the KIP! > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > +1 (non-binding) > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > walker > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > On Wed, Dec 9, 2020 at 11:40 AM Bruno > Cadonna < > > > > br...@confluent.io > > > > > > > > > > > > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks for the KIP, John! > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > +1 (non-binding) > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Best, > > > > > > > > > > > > > > > > > Bruno > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > On 08.12.20 18:03, John Roesler wrote: > > > > > > > > > > > > > > > > > > Hello all, > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > There hasn't been much discussion on > KIP-695 > > > > so far, so I'd > > > > > > > > > > > > > > > > > > like to go ahead and call for a vote. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > As a reminder, the purpose of KIP-695 to > > > > improve on the > > > > > > > > > > > > > > > > > > "task idling" feature we introduced in > > > > KIP-353. This KIP > > > > > > > > > > > > > > > > > > will allow Streams to offer deterministic > > > time > > > > semantics in > > > > > > > > > > > > > > > > > > join-type topologies. For example, it > makes > > > > sure that > > > > > > > > > > > > > > > > > > when you join two topics, that we > collate the > > > > topics by > > > > > > > > > > > > > > > > > > timestamp. That was always the intent > with > > > > task idling (KIP- > > > > > > > > > > > > > > > > > > 353), but it turns out the previous > mechanism > > > > couldn't > > > > > > > > > > > > > > > > > > provide the desired semantics. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > The details are here: > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/x/JSXZCQ > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks, > > > > > > > > > > > > > > > > > > -John > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > -- > > > > > > > > > > > > > > > -- Guozhang > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > >
