Hey Jun, Jason, Thanks much for all the feedback. I have updated the KIP based on the latest discussion. Can you help check whether it looks good?
Thanks, Dong On Thu, Jan 4, 2018 at 5:36 PM, Dong Lin <lindon...@gmail.com> wrote: > Hey Jun, > > Hmm... thinking about this more, I am not sure that the proposed API is > sufficient. For users that store offset externally, we probably need extra > API to return the leader_epoch and partition_epoch for all partitions that > consumers are consuming. I suppose these users currently use position() to > get the offset. Thus we probably need a new method positionWithEpoch(..) to > return <offset, partition_epoch, leader_epoch>. Does this sound reasonable? > > Thanks, > Dong > > > On Thu, Jan 4, 2018 at 5:26 PM, Jun Rao <j...@confluent.io> wrote: > >> Hi, Dong, >> >> Yes, that's what I am thinking. OffsetEpoch will be composed of >> (partition_epoch, >> leader_epoch). >> >> Thanks, >> >> Jun >> >> >> On Thu, Jan 4, 2018 at 4:22 PM, Dong Lin <lindon...@gmail.com> wrote: >> >> > Hey Jun, >> > >> > Thanks much. I like the the new API that you proposed. I am not sure >> what >> > you exactly mean by offset_epoch. I suppose that we can use the pair of >> > (partition_epoch, leader_epoch) as the offset_epoch, right? >> > >> > Thanks, >> > Dong >> > >> > On Thu, Jan 4, 2018 at 4:02 PM, Jun Rao <j...@confluent.io> wrote: >> > >> > > Hi, Dong, >> > > >> > > Got it. The api that you proposed works. The question is whether >> that's >> > the >> > > api that we want to have in the long term. My concern is that while >> the >> > api >> > > change is simple, the new api seems harder to explain and use. For >> > example, >> > > a consumer storing offsets externally now needs to call >> > > waitForMetadataUpdate() after calling seek(). >> > > >> > > An alternative approach is to make the following compatible api >> changes >> > in >> > > Consumer. >> > > * Add an additional OffsetEpoch field in OffsetAndMetadata. (no need >> to >> > > change the CommitSync() api) >> > > * Add a new api seek(TopicPartition partition, long offset, >> OffsetEpoch >> > > offsetEpoch). We can potentially deprecate the old api >> > seek(TopicPartition >> > > partition, long offset) in the future. >> > > >> > > The alternative approach has similar amount of api changes as yours >> but >> > has >> > > the following benefits. >> > > 1. The api works in a similar way as how offset management works now >> and >> > is >> > > probably what we want in the long term. >> > > 2. It can reset offsets better when there is data loss due to unclean >> > > leader election or correlated replica failure. >> > > 3. It can reset offsets better when topic is recreated. >> > > >> > > Thanks, >> > > >> > > Jun >> > > >> > > >> > > On Thu, Jan 4, 2018 at 2:05 PM, Dong Lin <lindon...@gmail.com> wrote: >> > > >> > > > Hey Jun, >> > > > >> > > > Yeah I agree that ideally we don't want an ever growing global >> metadata >> > > > version. I just think it may be more desirable to keep the consumer >> API >> > > > simple. >> > > > >> > > > In my current proposal, metadata version returned in the fetch >> response >> > > > will be stored with the offset together. More specifically, the >> > > > metadata_epoch in the new offset topic schema will be the largest >> > > > metadata_epoch from all the MetadataResponse and FetchResponse ever >> > > > received by this consumer. >> > > > >> > > > We probably don't have to change the consumer API for >> > > > commitSync(Map<TopicPartition, OffsetAndMetadata>). If user calls >> > > > commitSync(...) to commit offset 10 for a given partition, for most >> > > > use-cases, this consumer instance should have consumed message with >> > > offset >> > > > 9 from this partition, in which case the consumer can remember and >> use >> > > the >> > > > metadata_epoch from the corresponding FetchResponse when committing >> > > offset. >> > > > If user calls commitSync(..) to commit offset 10 for a given >> partition >> > > > without having consumed the message with offset 9 using this >> consumer >> > > > instance, this is probably an advanced use-case. In this case the >> > > advanced >> > > > user can retrieve the metadata_epoch using the newly added >> > > metadataEpoch() >> > > > API after it fetches the message with offset 9 (probably from >> another >> > > > consumer instance) and encode this metadata_epoch in the >> > > > string OffsetAndMetadata.metadata. Do you think this solution would >> > work? >> > > > >> > > > By "not sure that I fully understand your latest suggestion", are >> you >> > > > referring to solution related to unclean leader election using >> > > leader_epoch >> > > > in my previous email? >> > > > >> > > > Thanks, >> > > > Dong >> > > > >> > > > On Thu, Jan 4, 2018 at 1:33 PM, Jun Rao <j...@confluent.io> wrote: >> > > > >> > > > > Hi, Dong, >> > > > > >> > > > > Not sure that I fully understand your latest suggestion. >> Returning an >> > > > ever >> > > > > growing global metadata version itself is no ideal, but is fine. >> My >> > > > > question is whether the metadata version returned in the fetch >> > response >> > > > > needs to be stored with the offset together if offsets are stored >> > > > > externally. If so, we also have to change the consumer API for >> > > > commitSync() >> > > > > and need to worry about compatibility. If we don't store the >> metadata >> > > > > version together with the offset, on a consumer restart, it's not >> > clear >> > > > how >> > > > > we can ensure the metadata in the consumer is high enough since >> there >> > > is >> > > > no >> > > > > metadata version to compare with. >> > > > > >> > > > > Thanks, >> > > > > >> > > > > Jun >> > > > > >> > > > > >> > > > > On Wed, Jan 3, 2018 at 6:43 PM, Dong Lin <lindon...@gmail.com> >> > wrote: >> > > > > >> > > > > > Hey Jun, >> > > > > > >> > > > > > Thanks much for the explanation. >> > > > > > >> > > > > > I understand the advantage of partition_epoch over >> metadata_epoch. >> > My >> > > > > > current concern is that the use of leader_epoch and the >> > > partition_epoch >> > > > > > requires us considerable change to consumer's public API to take >> > care >> > > > of >> > > > > > the case where user stores offset externally. For example, >> > > *consumer*. >> > > > > > *commitSync*(..) would have to take a map whose value is >> <offset, >> > > > > metadata, >> > > > > > leader epoch, partition epoch>. *consumer*.*seek*(...) would >> also >> > > need >> > > > > > leader_epoch and partition_epoch as parameter. Technically we >> can >> > > > > probably >> > > > > > still make it work in a backward compatible manner after careful >> > > design >> > > > > and >> > > > > > discussion. But these changes can make the consumer's interface >> > > > > > unnecessarily complex for more users who do not store offset >> > > > externally. >> > > > > > >> > > > > > After thinking more about it, we can address all problems >> discussed >> > > by >> > > > > only >> > > > > > using the metadata_epoch without introducing leader_epoch or the >> > > > > > partition_epoch. The current KIP describes the changes to the >> > > consumer >> > > > > API >> > > > > > and how the new API can be used if user stores offset >> externally. >> > In >> > > > > order >> > > > > > to address the scenario you described earlier, we can include >> > > > > > metadata_epoch in the FetchResponse and the LeaderAndIsrRequest. >> > > > Consumer >> > > > > > remembers the largest metadata_epoch from all the FetchResponse >> it >> > > has >> > > > > > received. The metadata_epoch committed with the offset, either >> > within >> > > > or >> > > > > > outside Kafka, should be the largest metadata_epoch across all >> > > > > > FetchResponse and MetadataResponse ever received by this >> consumer. >> > > > > > >> > > > > > The drawback of using only the metadata_epoch is that we can not >> > > always >> > > > > do >> > > > > > the smart offset reset in case of unclean leader election which >> you >> > > > > > mentioned earlier. But in most case, unclean leader election >> > probably >> > > > > > happens when consumer is not rebalancing/restarting. In these >> > cases, >> > > > > either >> > > > > > consumer is not directly affected by unclean leader election >> since >> > it >> > > > is >> > > > > > not consuming from the end of the log, or consumer can derive >> the >> > > > > > leader_epoch from the most recent message received before it >> sees >> > > > > > OffsetOutOfRangeException. So I am not sure it is worth adding >> the >> > > > > > leader_epoch to consumer API to address the remaining corner >> case. >> > > What >> > > > > do >> > > > > > you think? >> > > > > > >> > > > > > Thanks, >> > > > > > Dong >> > > > > > >> > > > > > >> > > > > > >> > > > > > On Tue, Jan 2, 2018 at 6:28 PM, Jun Rao <j...@confluent.io> >> wrote: >> > > > > > >> > > > > > > Hi, Dong, >> > > > > > > >> > > > > > > Thanks for the reply. >> > > > > > > >> > > > > > > To solve the topic recreation issue, we could use either a >> global >> > > > > > metadata >> > > > > > > version or a partition level epoch. But either one will be a >> new >> > > > > concept, >> > > > > > > right? To me, the latter seems more natural. It also makes it >> > > easier >> > > > to >> > > > > > > detect if a consumer's offset is still valid after a topic is >> > > > > recreated. >> > > > > > As >> > > > > > > you pointed out, we don't need to store the partition epoch in >> > the >> > > > > > message. >> > > > > > > The following is what I am thinking. When a partition is >> created, >> > > we >> > > > > can >> > > > > > > assign a partition epoch from an ever-increasing global >> counter >> > and >> > > > > store >> > > > > > > it in /brokers/topics/[topic]/partitions/[partitionId] in ZK. >> > The >> > > > > > > partition >> > > > > > > epoch is propagated to every broker. The consumer will be >> > tracking >> > > a >> > > > > > tuple >> > > > > > > of <offset, leader epoch, partition epoch> for offsets. If a >> > topic >> > > is >> > > > > > > recreated, it's possible that a consumer's offset and leader >> > epoch >> > > > > still >> > > > > > > match that in the broker, but partition epoch won't be. In >> this >> > > case, >> > > > > we >> > > > > > > can potentially still treat the consumer's offset as out of >> range >> > > and >> > > > > > reset >> > > > > > > the offset based on the offset reset policy in the consumer. >> This >> > > > seems >> > > > > > > harder to do with a global metadata version. >> > > > > > > >> > > > > > > Jun >> > > > > > > >> > > > > > > >> > > > > > > >> > > > > > > On Mon, Dec 25, 2017 at 6:56 AM, Dong Lin < >> lindon...@gmail.com> >> > > > wrote: >> > > > > > > >> > > > > > > > Hey Jun, >> > > > > > > > >> > > > > > > > This is a very good example. After thinking through this in >> > > > detail, I >> > > > > > > agree >> > > > > > > > that we need to commit offset with leader epoch in order to >> > > address >> > > > > > this >> > > > > > > > example. >> > > > > > > > >> > > > > > > > I think the remaining question is how to address the >> scenario >> > > that >> > > > > the >> > > > > > > > topic is deleted and re-created. One possible solution is to >> > > commit >> > > > > > > offset >> > > > > > > > with both the leader epoch and the metadata version. The >> logic >> > > and >> > > > > the >> > > > > > > > implementation of this solution does not require a new >> concept >> > > > (e.g. >> > > > > > > > partition epoch) and it does not require any change to the >> > > message >> > > > > > format >> > > > > > > > or leader epoch. It also allows us to order the metadata in >> a >> > > > > > > > straightforward manner which may be useful in the future. >> So it >> > > may >> > > > > be >> > > > > > a >> > > > > > > > better solution than generating a random partition epoch >> every >> > > time >> > > > > we >> > > > > > > > create a partition. Does this sound reasonable? >> > > > > > > > >> > > > > > > > Previously one concern with using the metadata version is >> that >> > > > > consumer >> > > > > > > > will be forced to refresh metadata even if metadata version >> is >> > > > > > increased >> > > > > > > > due to topics that the consumer is not interested in. Now I >> > > > realized >> > > > > > that >> > > > > > > > this is probably not a problem. Currently client will >> refresh >> > > > > metadata >> > > > > > > > either due to InvalidMetadataException in the response from >> > > broker >> > > > or >> > > > > > due >> > > > > > > > to metadata expiry. The addition of the metadata version >> should >> > > > > > increase >> > > > > > > > the overhead of metadata refresh caused by >> > > > InvalidMetadataException. >> > > > > If >> > > > > > > > client refresh metadata due to expiry and it receives a >> > metadata >> > > > > whose >> > > > > > > > version is lower than the current metadata version, we can >> > reject >> > > > the >> > > > > > > > metadata but still reset the metadata age, which essentially >> > keep >> > > > the >> > > > > > > > existing behavior in the client. >> > > > > > > > >> > > > > > > > Thanks much, >> > > > > > > > Dong >> > > > > > > > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > >> > >> > >
