Thanks Anna, you are right on both points. I updated the KIP. -Jason
On Thu, Jul 26, 2018 at 2:08 PM, Anna Povzner <a...@confluent.io> wrote: > Hi Jason, > > Thanks for the update. I agree with the current proposal. > > Two minor comments: > 1) In “API Changes” section, first paragraph says that “users can catch the > more specific exception type and use the new `seekToNearest()` API defined > below.”. Since LogTruncationException “will include the partitions that > were truncated and the offset of divergence”., shouldn’t the client use > seek(offset) to seek to the offset of divergence in response to the > exception? > 2) In “Protocol Changes” section, OffsetsForLeaderEpoch subsection says > “Note > that consumers will send a sentinel value (-1) for the current epoch and > the broker will simply disregard that validation.”. Is that still true with > MetadataResponse containing leader epoch? > > Thanks, > Anna > > On Wed, Jul 25, 2018 at 1:44 PM Jason Gustafson <ja...@confluent.io> > wrote: > > > Hi All, > > > > I have made some updates to the KIP. As many of you know, a side project > of > > mine has been specifying the Kafka replication protocol in TLA. You can > > check out the code here if you are interested: > > https://github.com/hachikuji/kafka-specification. In addition to > > uncovering > > a couple unknown bugs in the replication protocol (e.g. > > https://issues.apache.org/jira/browse/KAFKA-7128), this has helped me > > validate the behavior in this KIP. In fact, the original version I > proposed > > had a weakness. I initially suggested letting the leader validate the > > expected epoch at the fetch offset. This made sense for the consumer in > the > > handling of unclean leader election, but it was not strong enough to > > protect the follower in all cases. In order to make advancement of the > high > > watermark safe, for example, the leader actually needs to be sure that > > every follower in the ISR matches its own epoch. > > > > I attempted to fix this problem by treating the epoch in the fetch > request > > slightly differently for consumers and followers. For consumers, it would > > be the expected epoch of the record at the fetch offset, and the leader > > would raise a LOG_TRUNCATION error if the expectation failed. For > > followers, it would be the current epoch and the leader would require > that > > it match its own epoch. This was unsatisfying both because of the > > inconsistency in behavior and because the consumer was left with the > weaker > > fencing that we already knew was insufficient for the replicas. > Ultimately > > I decided that we should make the behavior consistent and that meant that > > the consumer needed to act more like a following replica. Instead of > > checking for truncation while fetching, the consumer should check for > > truncation after leader changes. After checking for truncation, the > > consumer can then use the current epoch when fetching and get the > stronger > > protection that it provides. What this means is that the Metadata API > must > > include the current leader epoch. Given the problems we have had around > > stale metadata and how challenging they have been to debug, I'm convinced > > that this is a good idea in any case and it resolves the inconsistent > > behavior in the Fetch API. The downside is that there will be some > > additional overhead upon leader changes, but I don't think it is a major > > concern since leader changes are rare and the OffsetForLeaderEpoch > request > > is cheap. > > > > This approach leaves the door open for some interesting follow up > > improvements. For example, now that we have the leader epoch in the > > Metadata request, we can implement similar fencing for the Produce API. > And > > now that the consumer can reason about truncation, we could consider > having > > a configuration to expose records beyond the high watermark. This would > let > > users trade lower end-to-end latency for weaker durability semantics. It > is > > sort of like having an acks=0 option for the consumer. Neither of these > > options are included in this KIP, I am just mentioning them as potential > > work for the future. > > > > Finally, based on the discussion in this thread, I have added the > > seekToCommitted API for the consumer. Please take a look and let me know > > what you think. > > > > Thanks, > > Jason > > > > On Fri, Jul 20, 2018 at 2:34 PM, Guozhang Wang <wangg...@gmail.com> > wrote: > > > > > Hi Jason, > > > > > > The proposed API seems reasonable to me too. Could you please also > update > > > the wiki page ( > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP- > > > 320%3A+Allow+fetchers+to+detect+and+handle+log+truncation) > > > with a section say "workflow" on how the proposed API will be co-used > > with > > > others to: > > > > > > 1. consumer callers handling a LogTruncationException. > > > 2. consumer internals for handling a retriable > > UnknownLeaderEpochException. > > > > > > > > > Guozhang > > > > > > > > > On Tue, Jul 17, 2018 at 10:23 AM, Anna Povzner <a...@confluent.io> > > wrote: > > > > > > > Hi Jason, > > > > > > > > > > > > I also like your proposal and agree that > > KafkaConsumer#seekToCommitted() > > > > is > > > > more intuitive as a way to initialize both consumer's position and > its > > > > fetch state. > > > > > > > > > > > > My understanding that KafkaConsumer#seekToCommitted() is purely for > > > > clients > > > > who store their offsets externally, right? And we are still going to > > > > add KafkaConsumer#findOffsets() > > > > in this KIP as we discussed, so that the client can handle > > > > LogTruncationException? > > > > > > > > > > > > Thanks, > > > > > > > > Anna > > > > > > > > > > > > On Thu, Jul 12, 2018 at 3:57 PM Dong Lin <lindon...@gmail.com> > wrote: > > > > > > > > > Hey Jason, > > > > > > > > > > It is a great summary. The solution sounds good. I might have minor > > > > > comments regarding the method name. But we can discuss that minor > > > points > > > > > later after we reach consensus on the high level API. > > > > > > > > > > Thanks, > > > > > Dong > > > > > > > > > > > > > > > On Thu, Jul 12, 2018 at 11:41 AM, Jason Gustafson < > > ja...@confluent.io> > > > > > wrote: > > > > > > > > > > > Hey Anna and Dong, > > > > > > > > > > > > Thanks a lot for the great discussion. I've been hanging back a > bit > > > > > because > > > > > > honestly the best option hasn't seemed clear. I agree with Anna's > > > > general > > > > > > observation that there is a distinction between the position of > the > > > > > > consumer and its fetch state up to that position. If you think > > about > > > > it, > > > > > a > > > > > > committed offset actually represents both of these. The metadata > is > > > > used > > > > > to > > > > > > initialize the state of the consumer application and the offset > > > > > initializes > > > > > > the position. Additionally, we are extending the offset commit in > > > this > > > > > KIP > > > > > > to also include the last epoch fetched by the consumer, which is > > used > > > > to > > > > > > initialize the internal fetch state. Of course if you do an > > arbitrary > > > > > > `seek` and immediately commit offsets, then there won't be a last > > > epoch > > > > > to > > > > > > commit. This seems intuitive since there is no fetch state in > this > > > > case. > > > > > We > > > > > > only commit fetch state when we have it. > > > > > > > > > > > > So if we think about a committed offset as initializing both the > > > > > consumer's > > > > > > position and its fetch state, then the gap in the API is > evidently > > > that > > > > > we > > > > > > don't have a way to initialize the consumer to a committed > offset. > > We > > > > do > > > > > it > > > > > > implicitly of course for offsets stored in Kafka, but since > > external > > > > > > storage is a use case we support, then we should have an explicit > > API > > > > as > > > > > > well. Perhaps something like this: > > > > > > > > > > > > seekToCommitted(TopicPartition, OffsetAndMetadata) > > > > > > > > > > > > In this KIP, we are proposing to allow the `OffsetAndMetadata` > > object > > > > to > > > > > > include the leader epoch, so I think this would have the same > > effect > > > as > > > > > > Anna's suggested `seekToRecord`. But perhaps it is a more natural > > fit > > > > > given > > > > > > the current API? Furthermore, if we find a need for additional > > > metadata > > > > > in > > > > > > the offset commit API in the future, then we will just need to > > modify > > > > the > > > > > > `OffsetAndMetadata` object and we will not need a new `seek` API. > > > > > > > > > > > > With this approach, I think then we can leave the `position` API > as > > > it > > > > > is. > > > > > > The position of the consumer is still just the next expected > fetch > > > > > offset. > > > > > > If a user needs to record additional state based on previous > fetch > > > > > > progress, then they would use the result of the previous fetch to > > > > obtain > > > > > > it. This makes the dependence on fetch progress explicit. I think > > we > > > > > could > > > > > > make this a little more convenience with a helper in the > > > > > `ConsumerRecords` > > > > > > object, but I think that's more of a nice-to-have. > > > > > > > > > > > > Thoughts? > > > > > > > > > > > > By the way, I have been iterating a little bit on the replica > side > > of > > > > > this > > > > > > KIP. My initial proposal in fact did not have strong enough > fencing > > > to > > > > > > protect all of the edge cases. I believe the current proposal > fixes > > > the > > > > > > problems, but I am still verifying the model. > > > > > > > > > > > > Thanks, > > > > > > Jason > > > > > > > > > > > > > > > > > > On Wed, Jul 11, 2018 at 10:45 AM, Dong Lin <lindon...@gmail.com> > > > > wrote: > > > > > > > > > > > > > Hey Anna, > > > > > > > > > > > > > > Thanks much for the explanation. Approach 1 also sounds good to > > > me. I > > > > > > think > > > > > > > findOffsets() is useful for users who don't use automatic > offset > > > > reset > > > > > > > policy. > > > > > > > > > > > > > > Just one more question. Since users who store offsets > externally > > > need > > > > > to > > > > > > > provide leaderEpoch to findOffsets(...), do we need an extra > API > > > for > > > > > user > > > > > > > to get both offset and leaderEpoch, e.g. recordPosition()? > > > > > > > > > > > > > > Thanks, > > > > > > > Dong > > > > > > > > > > > > > > On Wed, Jul 11, 2018 at 10:12 AM, Anna Povzner < > > a...@confluent.io> > > > > > > wrote: > > > > > > > > > > > > > > > Hi Dong, > > > > > > > > > > > > > > > > > > > > > > > > What I called “not covering all use cases” is what you call > > > > > best-effort > > > > > > > > (not guaranteeing some corner cases). I think we are on the > > same > > > > page > > > > > > > here. > > > > > > > > > > > > > > > > > > > > > > > > I wanted to be clear in the API whether the consumer seeks > to a > > > > > > position > > > > > > > > (offset) or to a record (offset, leader epoch). The only > > use-case > > > > of > > > > > > > > seeking to a record is seeking to a committed offset for a > user > > > who > > > > > > > stores > > > > > > > > committed offsets externally. (Unless users find some other > > > reason > > > > to > > > > > > > seek > > > > > > > > to a record.) I thought it was possible to provide this > > > > functionality > > > > > > > with > > > > > > > > findOffset(offset, leader epoch) followed by a seek(offset). > > > > However, > > > > > > you > > > > > > > > are right that this will not handle the race condition where > > > > > > > non-divergent > > > > > > > > offset found by findOffset() could change again before the > > > consumer > > > > > > does > > > > > > > > the first fetch. > > > > > > > > > > > > > > > > > > > > > > > > Regarding position() — if we add position that returns > (offset, > > > > > leader > > > > > > > > epoch), this is specifically a position after a record that > was > > > > > > actually > > > > > > > > consumed or position of a committed record. In which case, I > > > still > > > > > > think > > > > > > > > it’s cleaner to get a record position of consumed message > from > > a > > > > new > > > > > > > helper > > > > > > > > method in ConsumerRecords() or from committed offsets. > > > > > > > > > > > > > > > > > > > > > > > > I think all the use-cases could be then covered with: > > > > > > > > > > > > > > > > (Approach 1) > > > > > > > > > > > > > > > > seekToRecord(offset, leaderEpoch) — this will just > > initialize/set > > > > the > > > > > > > > consumer state; > > > > > > > > > > > > > > > > findOffsets(offset, leaderEpoch) returns {offset, > leaderEpoch} > > > > > > > > > > > > > > > > > > > > > > > > If we agree that the race condition is also a corner case, > > then I > > > > > think > > > > > > > we > > > > > > > > can cover use-cases with: > > > > > > > > > > > > > > > > (Approach 2) > > > > > > > > > > > > > > > > findOffsets(offset, leaderEpoch) returns offset — we still > want > > > > > leader > > > > > > > > epoch as a parameter for the users who store their committed > > > > offsets > > > > > > > > externally. > > > > > > > > > > > > > > > > > > > > > > > > I am actually now leaning more to approach 1, since it is > more > > > > > > explicit, > > > > > > > > and maybe there are more use cases for it. > > > > > > > > > > > > > > > > > > > > > > > > Thanks, > > > > > > > > > > > > > > > > Anna > > > > > > > > > > > > > > > > > > > > > > > > On Tue, Jul 10, 2018 at 3:47 PM Dong Lin < > lindon...@gmail.com> > > > > > wrote: > > > > > > > > > > > > > > > > > Hey Anna, > > > > > > > > > > > > > > > > > > Thanks for the comment. To answer your question, it seems > > that > > > we > > > > > can > > > > > > > > cover > > > > > > > > > all case in this KIP. As stated in "Consumer Handling" > > section, > > > > > > KIP-101 > > > > > > > > > based approach will be used to derive the truncation offset > > > from > > > > > the > > > > > > > > > 2-tuple (offset, leaderEpoch). This approach is best effort > > and > > > > it > > > > > is > > > > > > > > > inaccurate only in very rare scenarios (as described in > > > KIP-279). > > > > > > > > > > > > > > > > > > By using seek(offset, leaderEpoch), consumer will still be > > able > > > > to > > > > > > > follow > > > > > > > > > this best-effort approach to detect log truncation and > > > determine > > > > > the > > > > > > > > > truncation offset. On the other hand, if we use > seek(offset), > > > > > > consumer > > > > > > > > will > > > > > > > > > not detect log truncation in some cases which weakens the > > > > guarantee > > > > > > of > > > > > > > > this > > > > > > > > > KIP. Does this make sense? > > > > > > > > > > > > > > > > > > Thanks, > > > > > > > > > Dong > > > > > > > > > > > > > > > > > > On Tue, Jul 10, 2018 at 3:14 PM, Anna Povzner < > > > a...@confluent.io > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > Sorry, I hit "send" before finishing. Continuing... > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > 2) Hiding most of the consumer handling log truncation > > logic > > > > with > > > > > > > > minimal > > > > > > > > > > exposure in KafkaConsumer API. I was proposing this > path. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Before answering your specific questions… I want to > answer > > to > > > > > your > > > > > > > > > comment > > > > > > > > > > “In general, maybe we should discuss the final solution > > that > > > > > covers > > > > > > > all > > > > > > > > > > cases?”. With current KIP, we don’t cover all cases of > > > consumer > > > > > > > > detecting > > > > > > > > > > log truncation because the KIP proposes a leader epoch > > cache > > > in > > > > > > > > consumer > > > > > > > > > > that does not persist across restarts. Plus, we only > store > > > last > > > > > > > > committed > > > > > > > > > > offset (either internally or users can store externally). > > > This > > > > > has > > > > > > a > > > > > > > > > > limitation that the consumer will not always be able to > > find > > > > > point > > > > > > of > > > > > > > > > > truncation just because we have a limited history (just > one > > > > data > > > > > > > > point). > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > So, maybe we should first agree on whether we accept that > > > > storing > > > > > > > last > > > > > > > > > > committed offset/leader epoch has a limitation that the > > > > consumer > > > > > > will > > > > > > > > not > > > > > > > > > > be able to detect log truncation in all cases? > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks, > > > > > > > > > > > > > > > > > > > > Anna > > > > > > > > > > > > > > > > > > > > On Tue, Jul 10, 2018 at 2:20 PM Anna Povzner < > > > > a...@confluent.io> > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > Hi Dong, > > > > > > > > > > > > > > > > > > > > > > Thanks for the follow up! I finally have much more > clear > > > > > > > > understanding > > > > > > > > > of > > > > > > > > > > > where you are coming from. > > > > > > > > > > > > > > > > > > > > > > You are right. The success of findOffsets()/finding a > > point > > > > of > > > > > > > > > > > non-divergence depends on whether we have enough > entries > > in > > > > the > > > > > > > > > > consumer's > > > > > > > > > > > leader epoch cache. However, I think this is a > > fundamental > > > > > > > limitation > > > > > > > > > of > > > > > > > > > > > having a leader epoch cache that does not persist > across > > > > > consumer > > > > > > > > > > restarts. > > > > > > > > > > > > > > > > > > > > > > If we consider the general case where consumer may or > may > > > not > > > > > > have > > > > > > > > this > > > > > > > > > > > cache, then I see two paths: > > > > > > > > > > > 1) Letting the user to track the leader epoch history > > > > > externally, > > > > > > > and > > > > > > > > > > have > > > > > > > > > > > more exposure to leader epoch and finding point of > > > > > non-divergence > > > > > > > in > > > > > > > > > > > KafkaConsumer API. I understand this is the case you > were > > > > > talking > > > > > > > > > about. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > On Tue, Jul 10, 2018 at 12:16 PM Dong Lin < > > > > lindon...@gmail.com > > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > >> Hey Anna, > > > > > > > > > > >> > > > > > > > > > > >> Thanks much for your detailed explanation and example! > > It > > > > does > > > > > > > help > > > > > > > > me > > > > > > > > > > >> understand the difference between our understanding. > > > > > > > > > > >> > > > > > > > > > > >> So it seems that the solution based on findOffsets() > > > > currently > > > > > > > > focuses > > > > > > > > > > >> mainly on the scenario that consumer has cached > > > leaderEpoch > > > > -> > > > > > > > > offset > > > > > > > > > > >> mapping whereas I was thinking about the general case > > > where > > > > > > > consumer > > > > > > > > > may > > > > > > > > > > >> or > > > > > > > > > > >> may not have this cache. I guess that is why we have > > > > different > > > > > > > > > > >> understanding here. I have some comments below. > > > > > > > > > > >> > > > > > > > > > > >> > > > > > > > > > > >> 3) The proposed solution using findOffsets(offset, > > > > > leaderEpoch) > > > > > > > > > followed > > > > > > > > > > >> by > > > > > > > > > > >> seek(offset) works if consumer has the cached > > leaderEpoch > > > -> > > > > > > > offset > > > > > > > > > > >> mapping. But if we assume consumer has this cache, do > we > > > > need > > > > > to > > > > > > > > have > > > > > > > > > > >> leaderEpoch in the findOffsets(...)? Intuitively, the > > > > > > > > > > findOffsets(offset) > > > > > > > > > > >> can also derive the leaderEpoch using offset just like > > the > > > > > > > proposed > > > > > > > > > > >> solution does with seek(offset). > > > > > > > > > > >> > > > > > > > > > > >> > > > > > > > > > > >> 4) If consumer does not have cached leaderEpoch -> > > offset > > > > > > mapping, > > > > > > > > > which > > > > > > > > > > >> is > > > > > > > > > > >> the case if consumer is restarted on a new machine, > then > > > it > > > > is > > > > > > not > > > > > > > > > clear > > > > > > > > > > >> what leaderEpoch would be included in the FetchRequest > > if > > > > > > consumer > > > > > > > > > does > > > > > > > > > > >> seek(offset). This is the case that motivates the > first > > > > > question > > > > > > > of > > > > > > > > > the > > > > > > > > > > >> previous email. In general, maybe we should discuss > the > > > > final > > > > > > > > solution > > > > > > > > > > >> that > > > > > > > > > > >> covers all cases? > > > > > > > > > > >> > > > > > > > > > > >> > > > > > > > > > > >> 5) The second question in my previous email is related > > to > > > > the > > > > > > > > > following > > > > > > > > > > >> paragraph: > > > > > > > > > > >> > > > > > > > > > > >> "... In some cases, offsets returned from position() > > could > > > > be > > > > > > > actual > > > > > > > > > > >> consumed messages by this consumer identified by > > {offset, > > > > > leader > > > > > > > > > epoch}. > > > > > > > > > > >> In > > > > > > > > > > >> other cases, position() returns offset that was not > > > actually > > > > > > > > consumed. > > > > > > > > > > >> Suppose, the user calls position() for the last > > > offset...". > > > > > > > > > > >> > > > > > > > > > > >> I guess my point is that, if user calls position() for > > the > > > > > last > > > > > > > > offset > > > > > > > > > > and > > > > > > > > > > >> uses that offset in seek(...), then user can probably > > just > > > > > call > > > > > > > > > > >> Consumer#seekToEnd() without calling position() and > > > > seek(...). > > > > > > > > > Similarly > > > > > > > > > > >> user can call Consumer#seekToBeginning() to the seek > to > > > the > > > > > > > earliest > > > > > > > > > > >> position without calling position() and seek(...). > Thus > > > > > > position() > > > > > > > > > only > > > > > > > > > > >> needs to return the actual consumed messages > identified > > by > > > > > > > {offset, > > > > > > > > > > leader > > > > > > > > > > >> epoch}. Does this make sense? > > > > > > > > > > >> > > > > > > > > > > >> > > > > > > > > > > >> Thanks, > > > > > > > > > > >> Dong > > > > > > > > > > >> > > > > > > > > > > >> > > > > > > > > > > >> On Mon, Jul 9, 2018 at 6:47 PM, Anna Povzner < > > > > > a...@confluent.io > > > > > > > > > > > > > > > > wrote: > > > > > > > > > > >> > > > > > > > > > > >> > Hi Dong, > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > >> > Thanks for considering my suggestions. > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > >> > Based on your comments, I realized that my > suggestion > > > was > > > > > not > > > > > > > > > complete > > > > > > > > > > >> with > > > > > > > > > > >> > regard to KafkaConsumer API vs. consumer-broker > > > protocol. > > > > > > While > > > > > > > I > > > > > > > > > > >> propose > > > > > > > > > > >> > to keep KafkaConsumer#seek() unchanged and take > offset > > > > only, > > > > > > the > > > > > > > > > > >> underlying > > > > > > > > > > >> > consumer will send the next FetchRequest() to broker > > > with > > > > > > offset > > > > > > > > and > > > > > > > > > > >> > leaderEpoch if it is known (based on leader epoch > > cache > > > in > > > > > > > > > consumer) — > > > > > > > > > > >> note > > > > > > > > > > >> > that this is different from the current KIP, which > > > > suggests > > > > > to > > > > > > > > > always > > > > > > > > > > >> send > > > > > > > > > > >> > unknown leader epoch after seek(). This way, if the > > > > consumer > > > > > > > and a > > > > > > > > > > >> broker > > > > > > > > > > >> > agreed on the point of non-divergence, which is some > > > > > {offset, > > > > > > > > > > >> leaderEpoch} > > > > > > > > > > >> > pair, the new leader which causes another truncation > > > (even > > > > > > > further > > > > > > > > > > back) > > > > > > > > > > >> > will be able to detect new divergence and restart > the > > > > > process > > > > > > of > > > > > > > > > > finding > > > > > > > > > > >> > the new point of non-divergence. So, to answer your > > > > > question, > > > > > > If > > > > > > > > the > > > > > > > > > > >> > truncation happens just after the user calls > > > > > > > > > > >> > KafkaConsumer#findOffsets(offset, leaderEpoch) > > followed > > > > by > > > > > > > > > > seek(offset), > > > > > > > > > > >> > the user will not seek to the wrong position without > > > > knowing > > > > > > > that > > > > > > > > > > >> > truncation has happened, because the consumer will > get > > > > > another > > > > > > > > > > >> truncation > > > > > > > > > > >> > error, and seek again. > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > >> > I am afraid, I did not understand your second > > question. > > > > Let > > > > > me > > > > > > > > > > >> summarize my > > > > > > > > > > >> > suggestions again, and then give an example to > > hopefully > > > > > make > > > > > > my > > > > > > > > > > >> > suggestions more clear. Also, the last part of my > > > example > > > > > > shows > > > > > > > > how > > > > > > > > > > the > > > > > > > > > > >> > use-case in your first question will work. If it > does > > > not > > > > > > answer > > > > > > > > > your > > > > > > > > > > >> > second question, would you mind clarifying? I am > also > > > > > focusing > > > > > > > on > > > > > > > > > the > > > > > > > > > > >> case > > > > > > > > > > >> > of a consumer having enough entries in the cache. > The > > > case > > > > > of > > > > > > > > > > restarting > > > > > > > > > > >> > from committed offset either stored externally or > > > > internally > > > > > > > will > > > > > > > > > > >> probably > > > > > > > > > > >> > need to be discussed more. > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > >> > Let me summarize my suggestion again: > > > > > > > > > > >> > > > > > > > > > > > >> > 1) KafkaConsumer#seek() and KafkaConsumer#position() > > > > remains > > > > > > > > > unchanged > > > > > > > > > > >> > > > > > > > > > > > >> > 2) New KafkaConsumer#findOffsets() takes {offset, > > > > > leaderEpoch} > > > > > > > > pair > > > > > > > > > > per > > > > > > > > > > >> > topic partition and returns offset per topic > > partition. > > > > > > > > > > >> > > > > > > > > > > > >> > 3) FetchRequest() to broker after > KafkaConsumer#seek() > > > > will > > > > > > > > contain > > > > > > > > > > the > > > > > > > > > > >> > offset set by seek and leaderEpoch that corresponds > to > > > the > > > > > > > offset > > > > > > > > > > based > > > > > > > > > > >> on > > > > > > > > > > >> > leader epoch cache in the consumer. > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > >> > The rest of this e-mail is a long and contrived > > example > > > > with > > > > > > > > several > > > > > > > > > > log > > > > > > > > > > >> > truncations and unclean leader elections to > illustrate > > > the > > > > > API > > > > > > > and > > > > > > > > > > your > > > > > > > > > > >> > first use-case. Suppose we have three brokers. > > > Initially, > > > > > > Broker > > > > > > > > A, > > > > > > > > > B, > > > > > > > > > > >> and > > > > > > > > > > >> > C has one message at offset 0 with leader epoch 0. > > Then, > > > > > > Broker > > > > > > > A > > > > > > > > > goes > > > > > > > > > > >> down > > > > > > > > > > >> > for some time. Broker B becomes a leader with epoch > 1, > > > and > > > > > > > writes > > > > > > > > > > >> messages > > > > > > > > > > >> > to offsets 1 and 2. Broker C fetches offset 1, but > > > before > > > > > > > fetching > > > > > > > > > > >> offset > > > > > > > > > > >> > 2, becomes a leader with leader epoch 2 and writes a > > > > message > > > > > > at > > > > > > > > > offset > > > > > > > > > > >> 2. > > > > > > > > > > >> > Here is the state of brokers at this point: > > > > > > > > > > >> > > > > > > > > > > > >> > > Broker A: > > > > > > > > > > >> > > offset 0, epoch 0 <— leader > > > > > > > > > > >> > > goes down… > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > >> > > Broker B: > > > > > > > > > > >> > > offset 0, epoch 0 > > > > > > > > > > >> > > offset 1, epoch 1 <- leader > > > > > > > > > > >> > > offset 2, epoch 1 > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > >> > Broker C: > > > > > > > > > > >> > > offset 0, epoch 0 > > > > > > > > > > >> > > offset 1, epoch 1 > > > > > > > > > > >> > > offset 2, epoch 2 <— leader > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > >> > Before Broker C becomes a leader with leader epoch > 2, > > > the > > > > > > > consumer > > > > > > > > > > >> consumed > > > > > > > > > > >> > the following messages from broker A and broker B: > > > > > > > > > > >> > > > > > > > > > > > >> > {offset=0, leaderEpoch=0}, {offset=1, > leaderEpoch=1}, > > > > > > {offset=2, > > > > > > > > > > >> > leaderEpoch=1}. > > > > > > > > > > >> > > > > > > > > > > > >> > Consumer’s leader epoch cache at this point contains > > the > > > > > > > following > > > > > > > > > > >> entries: > > > > > > > > > > >> > > > > > > > > > > > >> > (leaderEpoch=0, startOffset=0) > > > > > > > > > > >> > > > > > > > > > > > >> > (leaderEpoch=1, startOffset=1) > > > > > > > > > > >> > > > > > > > > > > > >> > endOffset = 3 > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > >> > Then, broker B becomes the follower of broker C, > > > truncates > > > > > and > > > > > > > > > starts > > > > > > > > > > >> > fetching from offset 2. > > > > > > > > > > >> > > > > > > > > > > > >> > Consumer sends fetchRequest(offset=3, leaderEpoch=1) > > and > > > > > gets > > > > > > > > > > >> > LOG_TRUNCATION > > > > > > > > > > >> > error from broker C. > > > > > > > > > > >> > > > > > > > > > > > >> > In response, the client calls > > KafkaConsumer#findOffsets( > > > > > > > offset=3, > > > > > > > > > > >> > leaderEpoch=1). The underlying consumer sends > > > > > > > > > > >> > OffsetsForLeaderEpoch(leaderEpoch=1), broker C > > responds > > > > with > > > > > > > > > > >> > {leaderEpoch=1, endOffset=2}. So, > > > > > > > > > KafkaConsumer#findOffsets(offset=3, > > > > > > > > > > >> > leaderEpoch=1) returns offset=2. > > > > > > > > > > >> > > > > > > > > > > > >> > In response, consumer calls KafkaConsumer@seek > > > (offset=2) > > > > > > > followed > > > > > > > > > by > > > > > > > > > > >> > poll(), which results in FetchRequest(offset=2, > > > > > leaderEpoch=1) > > > > > > > to > > > > > > > > > > >> broker C. > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > >> > I will continue with this example with the goal to > > > answer > > > > > your > > > > > > > > first > > > > > > > > > > >> > question about truncation just after findOffsets() > > > > followed > > > > > by > > > > > > > > > seek(): > > > > > > > > > > >> > > > > > > > > > > > >> > Suppose, brokers B and C go down, and broker A comes > > up > > > > and > > > > > > > > becomes > > > > > > > > > a > > > > > > > > > > >> > leader with leader epoch 3, and writes a message to > > > offset > > > > > 1. > > > > > > > > > Suppose, > > > > > > > > > > >> this > > > > > > > > > > >> > happens before the consumer gets response from > broker > > C > > > to > > > > > the > > > > > > > > > > previous > > > > > > > > > > >> > fetch request: FetchRequest(offset=2, > leaderEpoch=1). > > > > > > > > > > >> > > > > > > > > > > > >> > Consumer re-sends FetchRequest(offset=2, > > leaderEpoch=1) > > > to > > > > > > > broker > > > > > > > > A, > > > > > > > > > > >> which > > > > > > > > > > >> > returns LOG_TRUNCATION error, because broker A has > > > leader > > > > > > epoch > > > > > > > 3 > > > > > > > > > > > > > > > > > > > >> leader > > > > > > > > > > >> > epoch in FetchRequest with starting offset = 1 < > > offset > > > 2 > > > > in > > > > > > > > > > >> > FetchRequest(). > > > > > > > > > > >> > > > > > > > > > > > >> > In response, the user calls > KafkaConsumer#findOffsets( > > > > > > offset=2, > > > > > > > > > > >> > leaderEpoch=1). The underlying consumer sends > > > > > > > > > > >> > OffsetsForLeaderEpoch(leaderEpoch=1), broker A > > responds > > > > with > > > > > > > > > > >> > {leaderEpoch=0, endOffset=1}; the underlying > consumer > > > > finds > > > > > > > > > > leaderEpoch > > > > > > > > > > >> = 0 > > > > > > > > > > >> > in its cache with end offset == 1, which results in > > > > > > > > > > >> > KafkaConsumer#findOffsets(offset=2, leaderEpoch=1) > > > > returning > > > > > > > > offset > > > > > > > > > > = 1. > > > > > > > > > > >> > > > > > > > > > > > >> > In response, the user calls KafkaConsumer@seek > > > (offset=1) > > > > > > > followed > > > > > > > > > by > > > > > > > > > > >> > poll(), which results in FetchRequest(offset=1, > > > > > leaderEpoch=0) > > > > > > > to > > > > > > > > > > >> broker A, > > > > > > > > > > >> > which responds with message at offset 1, leader > epoch > > 3. > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > >> > I will think some more about consumers restarting > from > > > > > > committed > > > > > > > > > > >> offsets, > > > > > > > > > > >> > and send a follow up. > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > >> > Thanks, > > > > > > > > > > >> > > > > > > > > > > > >> > Anna > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > >> > On Sat, Jul 7, 2018 at 1:36 AM Dong Lin < > > > > > lindon...@gmail.com> > > > > > > > > > wrote: > > > > > > > > > > >> > > > > > > > > > > > >> > > Hey Anna, > > > > > > > > > > >> > > > > > > > > > > > > >> > > Thanks much for the thoughtful reply. It makes > sense > > > to > > > > > > > > different > > > > > > > > > > >> between > > > > > > > > > > >> > > "seeking to a message" and "seeking to a > position". > > I > > > > have > > > > > > to > > > > > > > > > > >> questions > > > > > > > > > > >> > > here: > > > > > > > > > > >> > > > > > > > > > > > > >> > > - For "seeking to a message" use-case, with the > > > proposed > > > > > > > > approach > > > > > > > > > > user > > > > > > > > > > >> > > needs to call findOffset(offset, leaderEpoch) > > followed > > > > by > > > > > > > > > > >> seek(offset). > > > > > > > > > > >> > If > > > > > > > > > > >> > > message truncation and message append happen > > > immediately > > > > > > after > > > > > > > > > > >> > > findOffset(offset, > > > > > > > > > > >> > > leaderEpoch) but before seek(offset), it seems > that > > > user > > > > > > will > > > > > > > > seek > > > > > > > > > > to > > > > > > > > > > >> the > > > > > > > > > > >> > > wrong message without knowing the truncation has > > > > happened. > > > > > > > Would > > > > > > > > > > this > > > > > > > > > > >> be > > > > > > > > > > >> > a > > > > > > > > > > >> > > problem? > > > > > > > > > > >> > > > > > > > > > > > > >> > > - For "seeking to a position" use-case, it seems > > that > > > > > there > > > > > > > can > > > > > > > > be > > > > > > > > > > two > > > > > > > > > > >> > > positions, i.e. earliest and latest. So these two > > > cases > > > > > can > > > > > > be > > > > > > > > > > >> > > Consumer.fulfilled by seekToBeginning() and > > > > > > > > Consumer.seekToEnd(). > > > > > > > > > > >> Then it > > > > > > > > > > >> > > seems that user will only need to call position() > > and > > > > > seek() > > > > > > > for > > > > > > > > > > >> "seeking > > > > > > > > > > >> > > to a message" use-case? > > > > > > > > > > >> > > > > > > > > > > > > >> > > Thanks, > > > > > > > > > > >> > > Dong > > > > > > > > > > >> > > > > > > > > > > > > >> > > > > > > > > > > > > >> > > On Wed, Jul 4, 2018 at 12:33 PM, Anna Povzner < > > > > > > > > a...@confluent.io> > > > > > > > > > > >> wrote: > > > > > > > > > > >> > > > > > > > > > > > > >> > > > Hi Jason and Dong, > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > I’ve been thinking about your suggestions and > > > > discussion > > > > > > > > > regarding > > > > > > > > > > >> > > > position(), seek(), and new proposed API. > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > Here is my thought process why we should keep > > > > position() > > > > > > and > > > > > > > > > > seek() > > > > > > > > > > >> API > > > > > > > > > > >> > > > unchanged. > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > I think we should separate {offset, leader > epoch} > > > that > > > > > > > > uniquely > > > > > > > > > > >> > > identifies > > > > > > > > > > >> > > > a message from an offset that is a position. In > > some > > > > > > cases, > > > > > > > > > > offsets > > > > > > > > > > >> > > > returned from position() could be actual > consumed > > > > > messages > > > > > > > by > > > > > > > > > this > > > > > > > > > > >> > > consumer > > > > > > > > > > >> > > > identified by {offset, leader epoch}. In other > > > cases, > > > > > > > > position() > > > > > > > > > > >> > returns > > > > > > > > > > >> > > > offset that was not actually consumed. Suppose, > > the > > > > user > > > > > > > calls > > > > > > > > > > >> > position() > > > > > > > > > > >> > > > for the last offset. Suppose we return {offset, > > > leader > > > > > > > epoch} > > > > > > > > of > > > > > > > > > > the > > > > > > > > > > >> > > > message currently in the log. Then, the message > > gets > > > > > > > truncated > > > > > > > > > > >> before > > > > > > > > > > >> > > > consumer’s first poll(). It does not make sense > > for > > > > > poll() > > > > > > > to > > > > > > > > > fail > > > > > > > > > > >> in > > > > > > > > > > >> > > this > > > > > > > > > > >> > > > case, because the log truncation did not > actually > > > > happen > > > > > > > from > > > > > > > > > the > > > > > > > > > > >> > > consumer > > > > > > > > > > >> > > > perspective. On the other hand, as the KIP > > proposes, > > > > it > > > > > > > makes > > > > > > > > > > sense > > > > > > > > > > >> for > > > > > > > > > > >> > > the > > > > > > > > > > >> > > > committed() method to return {offset, leader > > epoch} > > > > > > because > > > > > > > > > those > > > > > > > > > > >> > offsets > > > > > > > > > > >> > > > represent actual consumed messages. > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > The same argument applies to the seek() method — > > we > > > > are > > > > > > not > > > > > > > > > > seeking > > > > > > > > > > >> to > > > > > > > > > > >> > a > > > > > > > > > > >> > > > message, we are seeking to a position. > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > I like the proposal to add > > > KafkaConsumer#findOffsets() > > > > > > API. > > > > > > > I > > > > > > > > am > > > > > > > > > > >> > assuming > > > > > > > > > > >> > > > something like: > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > Map<TopicPartition, Long> > > > > > findOffsets(Map<TopicPartition, > > > > > > > > > > >> > OffsetAndEpoch> > > > > > > > > > > >> > > > offsetsToSearch) > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > Similar to seek() and position(), I think > > > > findOffsets() > > > > > > > should > > > > > > > > > > >> return > > > > > > > > > > >> > > > offset without leader epoch, because what we > want > > is > > > > the > > > > > > > > offset > > > > > > > > > > >> that we > > > > > > > > > > >> > > > think is closest to the not divergent message > from > > > the > > > > > > given > > > > > > > > > > >> consumed > > > > > > > > > > >> > > > message. Until the consumer actually fetches the > > > > > message, > > > > > > we > > > > > > > > > > should > > > > > > > > > > >> not > > > > > > > > > > >> > > let > > > > > > > > > > >> > > > the consumer store the leader epoch for a > message > > it > > > > did > > > > > > not > > > > > > > > > > >> consume. > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > So, the workflow will be: > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > 1) The user gets LogTruncationException with > > > {offset, > > > > > > leader > > > > > > > > > epoch > > > > > > > > > > >> of > > > > > > > > > > >> > the > > > > > > > > > > >> > > > previous message} (whatever we send with new > > > > > FetchRecords > > > > > > > > > > request). > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > 2) offset = findOffsets(tp -> {offset, leader > > > epoch}) > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > 3) seek(offset) > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > For the use-case where the users store committed > > > > offsets > > > > > > > > > > externally: > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > 1) Such users would have to track the leader > epoch > > > > > > together > > > > > > > > with > > > > > > > > > > an > > > > > > > > > > >> > > offset. > > > > > > > > > > >> > > > Otherwise, there is no way to detect later what > > > leader > > > > > > epoch > > > > > > > > was > > > > > > > > > > >> > > associated > > > > > > > > > > >> > > > with the message. I think it’s reasonable to ask > > > that > > > > > from > > > > > > > > users > > > > > > > > > > if > > > > > > > > > > >> > they > > > > > > > > > > >> > > > want to detect log truncation. Otherwise, they > > will > > > > get > > > > > > the > > > > > > > > > > current > > > > > > > > > > >> > > > behavior. > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > If the users currently get an offset to be > stored > > > > using > > > > > > > > > > position(), > > > > > > > > > > >> I > > > > > > > > > > >> > see > > > > > > > > > > >> > > > two possibilities. First, they call save offset > > > > returned > > > > > > > from > > > > > > > > > > >> > position() > > > > > > > > > > >> > > > that they call before poll(). In that case, it > > would > > > > not > > > > > > be > > > > > > > > > > correct > > > > > > > > > > >> to > > > > > > > > > > >> > > > store {offset, leader epoch} if we would have > > > changed > > > > > > > > position() > > > > > > > > > > to > > > > > > > > > > >> > > return > > > > > > > > > > >> > > > {offset, leader epoch} since actual fetched > > message > > > > > could > > > > > > be > > > > > > > > > > >> different > > > > > > > > > > >> > > > (from the example I described earlier). So, it > > would > > > > be > > > > > > more > > > > > > > > > > >> correct to > > > > > > > > > > >> > > > call position() after poll(). However, the user > > > > already > > > > > > gets > > > > > > > > > > >> > > > ConsumerRecords at this point, from which the > user > > > can > > > > > > > extract > > > > > > > > > > >> {offset, > > > > > > > > > > >> > > > leader epoch} of the last message. > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > So, I like the idea of adding a helper method to > > > > > > > > > ConsumerRecords, > > > > > > > > > > as > > > > > > > > > > >> > > Jason > > > > > > > > > > >> > > > proposed, something like: > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > public OffsetAndEpoch > lastOffsetWithLeaderEpoch(), > > > > where > > > > > > > > > > >> OffsetAndEpoch > > > > > > > > > > >> > > is > > > > > > > > > > >> > > > a data struct holding {offset, leader epoch}. > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > In this case, we would advise the user to follow > > the > > > > > > > workflow: > > > > > > > > > > >> poll(), > > > > > > > > > > >> > > get > > > > > > > > > > >> > > > {offset, leader epoch} from > > > > > ConsumerRecords#lastOffsetWith > > > > > > > > > > >> > LeaderEpoch(), > > > > > > > > > > >> > > > save offset and leader epoch, process records. > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > 2) When the user needs to seek to the last > > committed > > > > > > offset, > > > > > > > > > they > > > > > > > > > > >> call > > > > > > > > > > >> > > new > > > > > > > > > > >> > > > findOffsets(saved offset, leader epoch), and > then > > > > > > > > seek(offset). > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > What do you think? > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > Thanks, > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > Anna > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > On Tue, Jul 3, 2018 at 4:06 PM Dong Lin < > > > > > > > lindon...@gmail.com> > > > > > > > > > > >> wrote: > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > > Hey Jason, > > > > > > > > > > >> > > > > > > > > > > > > > > >> > > > > Thanks much for your thoughtful explanation. > > > > > > > > > > >> > > > > > > > > > > > > > > >> > > > > Yes the solution using findOffsets(offset, > > > > > leaderEpoch) > > > > > > > also > > > > > > > > > > >> works. > > > > > > > > > > >> > The > > > > > > > > > > >> > > > > advantage of this solution it adds only one > API > > > > > instead > > > > > > of > > > > > > > > two > > > > > > > > > > >> APIs. > > > > > > > > > > >> > > The > > > > > > > > > > >> > > > > concern is that its usage seems a bit more > > clumsy > > > > for > > > > > > > > advanced > > > > > > > > > > >> users. > > > > > > > > > > >> > > > More > > > > > > > > > > >> > > > > specifically, advanced users who store offsets > > > > > > externally > > > > > > > > will > > > > > > > > > > >> always > > > > > > > > > > >> > > > need > > > > > > > > > > >> > > > > to call findOffsets() before calling > > seek(offset) > > > > > during > > > > > > > > > > consumer > > > > > > > > > > >> > > > > initialization. And those advanced users will > > need > > > > to > > > > > > > > manually > > > > > > > > > > >> keep > > > > > > > > > > >> > > track > > > > > > > > > > >> > > > > of the leaderEpoch of the last ConsumerRecord. > > > > > > > > > > >> > > > > > > > > > > > > > > >> > > > > The other solution may be more user-friendly > for > > > > > > advanced > > > > > > > > > users > > > > > > > > > > >> is to > > > > > > > > > > >> > > add > > > > > > > > > > >> > > > > two APIs, `void seek(offset, leaderEpoch)` and > > > > > `(offset, > > > > > > > > > epoch) > > > > > > > > > > = > > > > > > > > > > >> > > > > offsetEpochs(topicPartition)`. > > > > > > > > > > >> > > > > > > > > > > > > > > >> > > > > I kind of prefer the second solution because > it > > is > > > > > > easier > > > > > > > to > > > > > > > > > use > > > > > > > > > > >> for > > > > > > > > > > >> > > > > advanced users. If we need to expose > leaderEpoch > > > > > anyway > > > > > > to > > > > > > > > > > safely > > > > > > > > > > >> > > > identify > > > > > > > > > > >> > > > > a message, it may be conceptually simpler to > > > expose > > > > it > > > > > > > > > directly > > > > > > > > > > in > > > > > > > > > > >> > > > > seek(...) rather than requiring one more > > > translation > > > > > > using > > > > > > > > > > >> > > > > findOffsets(...). But I am also OK with the > > first > > > > > > solution > > > > > > > > if > > > > > > > > > > >> other > > > > > > > > > > >> > > > > developers also favor that one :) > > > > > > > > > > >> > > > > > > > > > > > > > > >> > > > > Thanks, > > > > > > > > > > >> > > > > Dong > > > > > > > > > > >> > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > >> > > > > On Mon, Jul 2, 2018 at 11:10 AM, Jason > > Gustafson < > > > > > > > > > > >> ja...@confluent.io > > > > > > > > > > >> > > > > > > > > > > > > >> > > > > wrote: > > > > > > > > > > >> > > > > > > > > > > > > > > >> > > > > > Hi Dong, > > > > > > > > > > >> > > > > > > > > > > > > > > > >> > > > > > Thanks, I've been thinking about your > > > suggestions > > > > a > > > > > > bit. > > > > > > > > It > > > > > > > > > is > > > > > > > > > > >> > > > > challenging > > > > > > > > > > >> > > > > > to make this work given the current APIs. > One > > of > > > > the > > > > > > > > > > >> difficulties > > > > > > > > > > >> > is > > > > > > > > > > >> > > > that > > > > > > > > > > >> > > > > > we don't have an API to find the leader > epoch > > > for > > > > a > > > > > > > given > > > > > > > > > > >> offset at > > > > > > > > > > >> > > the > > > > > > > > > > >> > > > > > moment. So if the user does a seek to offset > > 5, > > > > then > > > > > > > we'll > > > > > > > > > > need > > > > > > > > > > >> a > > > > > > > > > > >> > new > > > > > > > > > > >> > > > API > > > > > > > > > > >> > > > > > to find the corresponding epoch in order to > > > > fulfill > > > > > > the > > > > > > > > new > > > > > > > > > > >> > > position() > > > > > > > > > > >> > > > > API. > > > > > > > > > > >> > > > > > Potentially we could modify ListOffsets to > > > enable > > > > > > > finding > > > > > > > > > the > > > > > > > > > > >> > leader > > > > > > > > > > >> > > > > epoch, > > > > > > > > > > >> > > > > > but I am not sure it is worthwhile. Perhaps > it > > > is > > > > > > > > reasonable > > > > > > > > > > for > > > > > > > > > > >> > > > advanced > > > > > > > > > > >> > > > > > usage to expect that the epoch information, > if > > > > > needed, > > > > > > > > will > > > > > > > > > be > > > > > > > > > > >> > > > extracted > > > > > > > > > > >> > > > > > from the records directly? It might make > sense > > > to > > > > > > > expose a > > > > > > > > > > >> helper > > > > > > > > > > >> > in > > > > > > > > > > >> > > > > > `ConsumerRecords` to make this a little > easier > > > > > though. > > > > > > > > > > >> > > > > > > > > > > > > > > > >> > > > > > Alternatively, if we think it is important > to > > > have > > > > > > this > > > > > > > > > > >> information > > > > > > > > > > >> > > > > exposed > > > > > > > > > > >> > > > > > directly, we could create batch APIs to > solve > > > the > > > > > > naming > > > > > > > > > > >> problem. > > > > > > > > > > >> > For > > > > > > > > > > >> > > > > > example: > > > > > > > > > > >> > > > > > > > > > > > > > > > >> > > > > > Map<TopicPartition, OffsetAndEpoch> > > positions(); > > > > > > > > > > >> > > > > > void seek(Map<TopicPartition, > OffsetAndEpoch> > > > > > > > positions); > > > > > > > > > > >> > > > > > > > > > > > > > > > >> > > > > > However, I'm actually leaning toward leaving > > the > > > > > > seek() > > > > > > > > and > > > > > > > > > > >> > > position() > > > > > > > > > > >> > > > > APIs > > > > > > > > > > >> > > > > > unchanged. Instead, we can add a new API to > > > search > > > > > for > > > > > > > > > offset > > > > > > > > > > by > > > > > > > > > > >> > > > > timestamp > > > > > > > > > > >> > > > > > or by offset/leader epoch. Let's say we call > > it > > > > > > > > > `findOffsets`. > > > > > > > > > > >> If > > > > > > > > > > >> > the > > > > > > > > > > >> > > > > user > > > > > > > > > > >> > > > > > hits a log truncation error, they can use > this > > > API > > > > > to > > > > > > > find > > > > > > > > > the > > > > > > > > > > >> > > closest > > > > > > > > > > >> > > > > > offset and then do a seek(). At the same > time, > > > we > > > > > > > > deprecate > > > > > > > > > > the > > > > > > > > > > >> > > > > > `offsetsForTimes` APIs. We now have two use > > > cases > > > > > > which > > > > > > > > > > require > > > > > > > > > > >> > > finding > > > > > > > > > > >> > > > > > offsets, so I think we should make this API > > > > general > > > > > > and > > > > > > > > > leave > > > > > > > > > > >> the > > > > > > > > > > >> > > door > > > > > > > > > > >> > > > > open > > > > > > > > > > >> > > > > > for future extensions. > > > > > > > > > > >> > > > > > > > > > > > > > > > >> > > > > > By the way, I'm unclear about the desire to > > move > > > > > part > > > > > > of > > > > > > > > > this > > > > > > > > > > >> > > > > functionality > > > > > > > > > > >> > > > > > to AdminClient. Guozhang suggested this > > > > previously, > > > > > > but > > > > > > > I > > > > > > > > > > think > > > > > > > > > > >> it > > > > > > > > > > >> > > only > > > > > > > > > > >> > > > > > makes sense for cross-cutting capabilities > > such > > > as > > > > > > topic > > > > > > > > > > >> creation. > > > > > > > > > > >> > If > > > > > > > > > > >> > > > we > > > > > > > > > > >> > > > > > have an API which is primarily useful by > > > > consumers, > > > > > > > then I > > > > > > > > > > think > > > > > > > > > > >> > > that's > > > > > > > > > > >> > > > > > where it should be exposed. The AdminClient > > also > > > > has > > > > > > its > > > > > > > > own > > > > > > > > > > API > > > > > > > > > > >> > > > > integrity > > > > > > > > > > >> > > > > > and should not become a dumping ground for > > > > advanced > > > > > > use > > > > > > > > > cases. > > > > > > > > > > >> I'll > > > > > > > > > > >> > > > > update > > > > > > > > > > >> > > > > > the KIP with the `findOffsets` API > suggested > > > > above > > > > > > and > > > > > > > we > > > > > > > > > can > > > > > > > > > > >> see > > > > > > > > > > >> > if > > > > > > > > > > >> > > > it > > > > > > > > > > >> > > > > > does a good enough job of keeping the API > > simple > > > > for > > > > > > > > common > > > > > > > > > > >> cases. > > > > > > > > > > >> > > > > > > > > > > > > > > > >> > > > > > Thanks, > > > > > > > > > > >> > > > > > Jason > > > > > > > > > > >> > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > >> > > > > > On Sat, Jun 30, 2018 at 4:39 AM, Dong Lin < > > > > > > > > > > lindon...@gmail.com> > > > > > > > > > > >> > > wrote: > > > > > > > > > > >> > > > > > > > > > > > > > > > >> > > > > > > Hey Jason, > > > > > > > > > > >> > > > > > > > > > > > > > > > > >> > > > > > > Regarding seek(...), it seems that we want > > an > > > > API > > > > > > for > > > > > > > > user > > > > > > > > > > to > > > > > > > > > > >> > > > > initialize > > > > > > > > > > >> > > > > > > consumer with (offset, leaderEpoch) and > that > > > API > > > > > > > should > > > > > > > > > > allow > > > > > > > > > > >> > > > throwing > > > > > > > > > > >> > > > > > > PartitionTruncationException. Suppose we > > agree > > > > on > > > > > > > this, > > > > > > > > > then > > > > > > > > > > >> > > > > > > seekToNearest() is not sufficient because > it > > > > will > > > > > > > always > > > > > > > > > > >> swallow > > > > > > > > > > >> > > > > > > PartitionTruncationException. Here we have > > two > > > > > > > options. > > > > > > > > > The > > > > > > > > > > >> first > > > > > > > > > > >> > > > > option > > > > > > > > > > >> > > > > > is > > > > > > > > > > >> > > > > > > to add API offsetsForLeaderEpochs() to > > > translate > > > > > > > > > > (leaderEpoch, > > > > > > > > > > >> > > > offset) > > > > > > > > > > >> > > > > to > > > > > > > > > > >> > > > > > > offset. The second option is to have add > > > > > > seek(offset, > > > > > > > > > > >> > leaderEpoch). > > > > > > > > > > >> > > > It > > > > > > > > > > >> > > > > > > seems that second option may be more > simpler > > > > > because > > > > > > > it > > > > > > > > > > makes > > > > > > > > > > >> it > > > > > > > > > > >> > > > clear > > > > > > > > > > >> > > > > > that > > > > > > > > > > >> > > > > > > (offset, leaderEpoch) will be used to > > identify > > > > > > > > consumer's > > > > > > > > > > >> > position > > > > > > > > > > >> > > > in a > > > > > > > > > > >> > > > > > > partition. And user only needs to handle > > > > > > > > > > >> > > PartitionTruncationException > > > > > > > > > > >> > > > > > from > > > > > > > > > > >> > > > > > > the poll(). In comparison the first option > > > > seems a > > > > > > bit > > > > > > > > > > harder > > > > > > > > > > >> to > > > > > > > > > > >> > > use > > > > > > > > > > >> > > > > > > because user have to also handle the > > > > > > > > > > >> PartitionTruncationException > > > > > > > > > > >> > > if > > > > > > > > > > >> > > > > > > offsetsForLeaderEpochs() returns different > > > > offset > > > > > > from > > > > > > > > > > >> > > user-provided > > > > > > > > > > >> > > > > > > offset. What do you think? > > > > > > > > > > >> > > > > > > > > > > > > > > > > >> > > > > > > If we decide to add API seek(offset, > > > > leaderEpoch), > > > > > > > then > > > > > > > > we > > > > > > > > > > can > > > > > > > > > > >> > > decide > > > > > > > > > > >> > > > > > > whether and how to add API to translate > > > (offset, > > > > > > > > > > leaderEpoch) > > > > > > > > > > >> to > > > > > > > > > > >> > > > > offset. > > > > > > > > > > >> > > > > > It > > > > > > > > > > >> > > > > > > seems that this API will be needed by > > advanced > > > > > user > > > > > > to > > > > > > > > > don't > > > > > > > > > > >> want > > > > > > > > > > >> > > > auto > > > > > > > > > > >> > > > > > > offset reset (so that it can be notified) > > but > > > > > still > > > > > > > > wants > > > > > > > > > to > > > > > > > > > > >> > reset > > > > > > > > > > >> > > > > offset > > > > > > > > > > >> > > > > > > to closest. For those users if probably > > makes > > > > > sense > > > > > > to > > > > > > > > > only > > > > > > > > > > >> have > > > > > > > > > > >> > > the > > > > > > > > > > >> > > > > API > > > > > > > > > > >> > > > > > in > > > > > > > > > > >> > > > > > > AdminClient. offsetsForTimes() seems like > a > > > > common > > > > > > API > > > > > > > > > that > > > > > > > > > > >> will > > > > > > > > > > >> > be > > > > > > > > > > >> > > > > > needed > > > > > > > > > > >> > > > > > > by user's of consumer in general, so it > may > > be > > > > > more > > > > > > > > > > >> reasonable to > > > > > > > > > > >> > > > stay > > > > > > > > > > >> > > > > in > > > > > > > > > > >> > > > > > > the consumer API. I don't have a strong > > > opinion > > > > on > > > > > > > > whether > > > > > > > > > > >> > > > > > > offsetsForTimes() should be replaced by > API > > in > > > > > > > > > AdminClient. > > > > > > > > > > >> > > > > > > > > > > > > > > > > >> > > > > > > Though (offset, leaderEpoch) is needed to > > > > uniquely > > > > > > > > > identify > > > > > > > > > > a > > > > > > > > > > >> > > message > > > > > > > > > > >> > > > > in > > > > > > > > > > >> > > > > > > general, it is only needed for advanced > > users > > > > who > > > > > > has > > > > > > > > > turned > > > > > > > > > > >> on > > > > > > > > > > >> > > > unclean > > > > > > > > > > >> > > > > > > leader election, need to use seek(..), and > > > don't > > > > > > want > > > > > > > > auto > > > > > > > > > > >> offset > > > > > > > > > > >> > > > > reset. > > > > > > > > > > >> > > > > > > Most other users probably just want to > > enable > > > > auto > > > > > > > > offset > > > > > > > > > > >> reset > > > > > > > > > > >> > and > > > > > > > > > > >> > > > > store > > > > > > > > > > >> > > > > > > offset in Kafka. Thus we might want to > keep > > > the > > > > > > > existing > > > > > > > > > > >> > > offset-only > > > > > > > > > > >> > > > > APIs > > > > > > > > > > >> > > > > > > (e.g. seek() and position()) for most > users > > > > while > > > > > > > adding > > > > > > > > > new > > > > > > > > > > >> APIs > > > > > > > > > > >> > > for > > > > > > > > > > >> > > > > > > advanced users. And yes, it seems that we > > need > > > > new > > > > > > > name > > > > > > > > > for > > > > > > > > > > >> > > > position(). > > > > > > > > > > >> > > > > > > > > > > > > > > > > >> > > > > > > Though I think we need new APIs to carry > the > > > new > > > > > > > > > information > > > > > > > > > > >> > (e.g. > > > > > > > > > > >> > > > > > > leaderEpoch), I am not very sure how that > > > should > > > > > > look > > > > > > > > > like. > > > > > > > > > > >> One > > > > > > > > > > >> > > > > possible > > > > > > > > > > >> > > > > > > option is those APIs in KIP-232. Another > > > option > > > > is > > > > > > > > > something > > > > > > > > > > >> like > > > > > > > > > > >> > > > this: > > > > > > > > > > >> > > > > > > > > > > > > > > > > >> > > > > > > ````` > > > > > > > > > > >> > > > > > > class OffsetEpochs { > > > > > > > > > > >> > > > > > > long offset; > > > > > > > > > > >> > > > > > > int leaderEpoch; > > > > > > > > > > >> > > > > > > int partitionEpoch; // This may be > > needed > > > > > later > > > > > > as > > > > > > > > > > >> discussed > > > > > > > > > > >> > in > > > > > > > > > > >> > > > > > KIP-232 > > > > > > > > > > >> > > > > > > ... // Hopefully these are all we need > to > > > > > identify > > > > > > > > > message > > > > > > > > > > >> in > > > > > > > > > > >> > > > Kafka. > > > > > > > > > > >> > > > > > But > > > > > > > > > > >> > > > > > > if we need more then we can add new fields > > in > > > > this > > > > > > > > class. > > > > > > > > > > >> > > > > > > } > > > > > > > > > > >> > > > > > > > > > > > > > > > > >> > > > > > > OffsetEpochs offsetEpochs(TopicPartition); > > > > > > > > > > >> > > > > > > > > > > > > > > > > >> > > > > > > void seek(TopicPartition, OffsetEpochs); > > > > > > > > > > >> > > > > > > `````` > > > > > > > > > > >> > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > >> > > > > > > Thanks, > > > > > > > > > > >> > > > > > > Dong > > > > > > > > > > >> > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > >> > > > > > > On Fri, Jun 29, 2018 at 11:13 AM, Jason > > > > Gustafson > > > > > < > > > > > > > > > > >> > > > ja...@confluent.io> > > > > > > > > > > >> > > > > > > wrote: > > > > > > > > > > >> > > > > > > > > > > > > > > > > >> > > > > > > > Hey Dong, > > > > > > > > > > >> > > > > > > > > > > > > > > > > > >> > > > > > > > Thanks for the feedback. The first three > > > > points > > > > > > are > > > > > > > > > easy: > > > > > > > > > > >> > > > > > > > > > > > > > > > > > >> > > > > > > > 1. Yes, we should be consistent. > > > > > > > > > > >> > > > > > > > 2. Yes, I will add this. > > > > > > > > > > >> > > > > > > > 3. Yes, I think we should document the > > > changes > > > > > to > > > > > > > the > > > > > > > > > > >> committed > > > > > > > > > > >> > > > > offset > > > > > > > > > > >> > > > > >
