Hey Jun, Jason, Thanks much for all the review! I will open the voting thread.
Regards, Dong On Tue, Jan 23, 2018 at 3:37 PM, Jun Rao <j...@confluent.io> wrote: > Hi, Dong, > > The current KIP looks good to me. > > Thanks, > > Jun > > On Tue, Jan 23, 2018 at 12:29 PM, Dong Lin <lindon...@gmail.com> wrote: > > > Hey Jun, > > > > Do you think the current KIP looks OK? I am wondering if we can open the > > voting thread. > > > > Thanks! > > Dong > > > > On Fri, Jan 19, 2018 at 3:08 PM, Dong Lin <lindon...@gmail.com> wrote: > > > > > Hey Jun, > > > > > > I think we can probably have a static method in Util class to decode > the > > > byte[]. Both KafkaConsumer implementation and the user application will > > be > > > able to decode the byte array and log its content for debug purpose. So > > it > > > seems that we can still print the information we want. It is just not > > > explicitly exposed in the consumer interface. Would this address the > > > problem here? > > > > > > Yeah we can include OffsetEpoch in AdminClient. This can be added in > > > KIP-222? Is there something you would like me to add in this KIP? > > > > > > Thanks! > > > Dong > > > > > > On Fri, Jan 19, 2018 at 3:00 PM, Jun Rao <j...@confluent.io> wrote: > > > > > >> Hi, Dong, > > >> > > >> The issue with using just byte[] for OffsetEpoch is that it won't be > > >> printable, which makes debugging harder. > > >> > > >> Also, KIP-222 proposes a listGroupOffset() method in AdminClient. If > > that > > >> gets adopted before this KIP, we probably want to include OffsetEpoch > in > > >> the AdminClient too. > > >> > > >> Thanks, > > >> > > >> Jun > > >> > > >> > > >> On Thu, Jan 18, 2018 at 6:30 PM, Dong Lin <lindon...@gmail.com> > wrote: > > >> > > >> > Hey Jun, > > >> > > > >> > I agree. I have updated the KIP to remove the class OffetEpoch and > > >> replace > > >> > OffsetEpoch with byte[] in APIs that use it. Can you see if it looks > > >> good? > > >> > > > >> > Thanks! > > >> > Dong > > >> > > > >> > On Thu, Jan 18, 2018 at 6:07 PM, Jun Rao <j...@confluent.io> wrote: > > >> > > > >> > > Hi, Dong, > > >> > > > > >> > > Thanks for the updated KIP. It looks good to me now. The only > thing > > is > > >> > > for OffsetEpoch. > > >> > > If we expose the individual fields in the class, we probably don't > > >> need > > >> > the > > >> > > encode/decode methods. If we want to hide the details of > > OffsetEpoch, > > >> we > > >> > > probably don't want expose the individual fields. > > >> > > > > >> > > Jun > > >> > > > > >> > > On Wed, Jan 17, 2018 at 10:10 AM, Dong Lin <lindon...@gmail.com> > > >> wrote: > > >> > > > > >> > > > Thinking about point 61 more, I realize that the async zookeeper > > >> read > > >> > may > > >> > > > make it less of an issue for controller to read more zookeeper > > >> nodes. > > >> > > > Writing partition_epoch in the per-partition znode makes it > > simpler > > >> to > > >> > > > handle the broker failure between zookeeper writes for a topic > > >> > creation. > > >> > > I > > >> > > > have updated the KIP to use the suggested approach. > > >> > > > > > >> > > > > > >> > > > On Wed, Jan 17, 2018 at 9:57 AM, Dong Lin <lindon...@gmail.com> > > >> wrote: > > >> > > > > > >> > > > > Hey Jun, > > >> > > > > > > >> > > > > Thanks much for the comments. Please see my comments inline. > > >> > > > > > > >> > > > > On Tue, Jan 16, 2018 at 4:38 PM, Jun Rao <j...@confluent.io> > > >> wrote: > > >> > > > > > > >> > > > >> Hi, Dong, > > >> > > > >> > > >> > > > >> Thanks for the updated KIP. Looks good to me overall. Just a > > few > > >> > minor > > >> > > > >> comments. > > >> > > > >> > > >> > > > >> 60. OffsetAndMetadata positionAndOffsetEpoch(TopicPartition > > >> > > partition): > > >> > > > >> It > > >> > > > >> seems that there is no need to return metadata. We probably > > want > > >> to > > >> > > > return > > >> > > > >> sth like OffsetAndEpoch. > > >> > > > >> > > >> > > > > > > >> > > > > Previously I think we may want to re-use the existing class to > > >> keep > > >> > our > > >> > > > > consumer interface simpler. I have updated the KIP to add > class > > >> > > > > OffsetAndOffsetEpoch. I didn't use OffsetAndEpoch because user > > may > > >> > > > confuse > > >> > > > > this name with OffsetEpoch. Does this sound OK? > > >> > > > > > > >> > > > > > > >> > > > >> > > >> > > > >> 61. Should we store partition_epoch in > > >> > > > >> /brokers/topics/[topic]/partitions/[partitionId] in ZK? > > >> > > > >> > > >> > > > > > > >> > > > > I have considered this. I think the advantage of adding the > > >> > > > > partition->partition_epoch map in the existing > > >> > > > > znode /brokers/topics/[topic]/partitions is that controller > > only > > >> > needs > > >> > > > to > > >> > > > > read one znode per topic to gets its partition_epoch > > information. > > >> > > > Otherwise > > >> > > > > controller may need to read one extra znode per partition to > get > > >> the > > >> > > same > > >> > > > > information. > > >> > > > > > > >> > > > > When we delete partition or expand partition of a topic, > someone > > >> > needs > > >> > > to > > >> > > > > modify partition->partition_epoch map in znode > > >> > > > > /brokers/topics/[topic]/partitions. This may seem a bit more > > >> > > complicated > > >> > > > > than simply adding or deleting znode /brokers/topics/[topic]/ > > >> > > > partitions/[partitionId]. > > >> > > > > But the complexity is probably similar to the existing > operation > > >> of > > >> > > > > modifying the partition->replica_list mapping in znode > > >> > > > > /brokers/topics/[topic]. So not sure it is better to store the > > >> > > > > partition_epoch in /brokers/topics/[topic]/partit > > >> ions/[partitionId]. > > >> > > > What > > >> > > > > do you think? > > >> > > > > > > >> > > > > > > >> > > > >> > > >> > > > >> 62. For checking outdated metadata in the client, we probably > > >> want > > >> > to > > >> > > > add > > >> > > > >> when max_partition_epoch will be used. > > >> > > > >> > > >> > > > > > > >> > > > > The max_partition_epoch is used in the Proposed Changes -> > > >> Client's > > >> > > > > metadata refresh section to determine whether a metadata is > > >> outdated. > > >> > > And > > >> > > > > this formula is referenced and re-used in other sections to > > >> determine > > >> > > > > whether a metadata is outdated. Does this formula look OK? > > >> > > > > > > >> > > > > > > >> > > > >> > > >> > > > >> 63. "The leader_epoch should be the largest leader_epoch of > > >> messages > > >> > > > whose > > >> > > > >> offset < the commit offset. If no message has been consumed > > since > > >> > > > consumer > > >> > > > >> initialization, the leader_epoch from seek(...) or > > >> > OffsetFetchResponse > > >> > > > >> should be used. The partition_epoch should be read from the > > last > > >> > > > >> FetchResponse corresponding to the given partition and commit > > >> > offset. > > >> > > ": > > >> > > > >> leader_epoch and partition_epoch are associated with an > offset. > > >> So, > > >> > if > > >> > > > no > > >> > > > >> message is consumed, there is no offset and therefore there > is > > no > > >> > need > > >> > > > to > > >> > > > >> read leader_epoch and partition_epoch. Also, the leader_epoch > > >> > > associated > > >> > > > >> with the offset should just come from the messages returned > in > > >> the > > >> > > fetch > > >> > > > >> response. > > >> > > > >> > > >> > > > > > > >> > > > > I am thinking that, if user calls seek(..) and commitSync(...) > > >> > without > > >> > > > > consuming any messages, we should re-use the leader_epoch and > > >> > > > > partition_epoch provided by the seek(...) in the > > >> OffsetCommitRequest. > > >> > > And > > >> > > > > if messages have been successfully consumed, then leader_epoch > > >> will > > >> > > come > > >> > > > > from the messages returned in the fetch response. The > condition > > >> > > "messages > > >> > > > > whose offset < the commit offset" is needed to take care of > the > > >> log > > >> > > > > compacted topic which may have offset gap due to log cleaning. > > >> > > > > > > >> > > > > Did I miss something here? Or should I rephrase the paragraph > to > > >> make > > >> > > it > > >> > > > > less confusing? > > >> > > > > > > >> > > > > > > >> > > > >> 64. Could you include the public methods in the OffsetEpoch > > >> class? > > >> > > > >> > > >> > > > > > > >> > > > > I mistakenly deleted the definition of OffsetEpoch class from > > the > > >> > KIP. > > >> > > I > > >> > > > > just added it back with the public methods. Could you take > > another > > >> > > look? > > >> > > > > > > >> > > > > > > >> > > > >> > > >> > > > >> Jun > > >> > > > >> > > >> > > > >> > > >> > > > >> On Thu, Jan 11, 2018 at 5:43 PM, Dong Lin < > lindon...@gmail.com > > > > > >> > > wrote: > > >> > > > >> > > >> > > > >> > Hey Jun, > > >> > > > >> > > > >> > > > >> > Thanks much. I agree that we can not rely on committed > > offsets > > >> to > > >> > be > > >> > > > >> always > > >> > > > >> > deleted when we delete topic. So it is necessary to use a > > >> > > > per-partition > > >> > > > >> > epoch that does not change unless this partition is > deleted. > > I > > >> > also > > >> > > > >> agree > > >> > > > >> > that it is very nice to be able to uniquely identify a > > message > > >> > with > > >> > > > >> > (offset, leader_epoch, partition_epoch) in face of > potential > > >> topic > > >> > > > >> deletion > > >> > > > >> > and unclean leader election. > > >> > > > >> > > > >> > > > >> > I agree with all your comments. And I have updated the KIP > > >> based > > >> > on > > >> > > > our > > >> > > > >> > latest discussion. In addition, I added > > >> > > InvalidPartitionEpochException > > >> > > > >> > which will be thrown by consumer.poll() if the > > partition_epoch > > >> > > > >> associated > > >> > > > >> > with the partition, which can be given to consumer using > > >> > seek(...), > > >> > > is > > >> > > > >> > different from the partition_epoch in the FetchResponse. > > >> > > > >> > > > >> > > > >> > Can you take another look at the latest KIP? > > >> > > > >> > > > >> > > > >> > Thanks! > > >> > > > >> > Dong > > >> > > > >> > > > >> > > > >> > > > >> > > > >> > > > >> > > > >> > On Wed, Jan 10, 2018 at 2:24 PM, Jun Rao <j...@confluent.io > > > > >> > wrote: > > >> > > > >> > > > >> > > > >> > > Hi, Dong, > > >> > > > >> > > > > >> > > > >> > > My replies are the following. > > >> > > > >> > > > > >> > > > >> > > 60. What you described could also work. The drawback is > > that > > >> we > > >> > > will > > >> > > > >> be > > >> > > > >> > > unnecessarily changing the partition epoch when a > partition > > >> > hasn't > > >> > > > >> really > > >> > > > >> > > changed. I was imagining that the partition epoch will be > > >> stored > > >> > > in > > >> > > > >> > > /brokers/topics/[topic]/partitions/[partitionId], > instead > > >> of at > > >> > > the > > >> > > > >> > topic > > >> > > > >> > > level. So, not sure if ZK size limit is an issue. > > >> > > > >> > > > > >> > > > >> > > 61, 62 and 65. To me, the offset + offset_epoch is a > unique > > >> > > > identifier > > >> > > > >> > for > > >> > > > >> > > a message. So, if a message hasn't changed, the offset > and > > >> the > > >> > > > >> associated > > >> > > > >> > > offset_epoch ideally should remain the same (it will be > > kind > > >> of > > >> > > > weird > > >> > > > >> if > > >> > > > >> > > two consumer apps save the offset on the same message, > but > > >> the > > >> > > > >> > offset_epoch > > >> > > > >> > > are different). partition_epoch + leader_epoch give us > > that. > > >> > > > >> > global_epoch + > > >> > > > >> > > leader_epoch don't. If we use this approach, we can solve > > not > > >> > only > > >> > > > the > > >> > > > >> > > problem that you have identified, but also other problems > > >> when > > >> > > there > > >> > > > >> is > > >> > > > >> > > data loss or topic re-creation more reliably. For > example, > > in > > >> > the > > >> > > > >> future, > > >> > > > >> > > if we include the partition_epoch and leader_epoch in the > > >> fetch > > >> > > > >> request, > > >> > > > >> > > the server can do a more reliable check of whether that > > >> offset > > >> > is > > >> > > > >> valid > > >> > > > >> > or > > >> > > > >> > > not. I am not sure that we can rely upon all external > > >> offsets to > > >> > > be > > >> > > > >> > removed > > >> > > > >> > > on topic deletion. For example, a topic may be deleted by > > an > > >> > admin > > >> > > > who > > >> > > > >> > may > > >> > > > >> > > not know all the applications. > > >> > > > >> > > > > >> > > > >> > > If we agree on the above, the second question is then how > > to > > >> > > > reliably > > >> > > > >> > > propagate the partition_epoch and the leader_epoch to the > > >> > consumer > > >> > > > >> when > > >> > > > >> > > there are leader or partition changes. The leader_epoch > > comes > > >> > from > > >> > > > the > > >> > > > >> > > message, which is reliable. So, I was suggesting that > when > > we > > >> > > store > > >> > > > an > > >> > > > >> > > offset, we can just store the leader_epoch from the > message > > >> set > > >> > > > >> > containing > > >> > > > >> > > that offset. Similarly, I was thinking that if the > > >> > partition_epoch > > >> > > > is > > >> > > > >> in > > >> > > > >> > > the fetch response, we can propagate partition_epoch > > reliably > > >> > > where > > >> > > > is > > >> > > > >> > > partition_epoch change. > > >> > > > >> > > > > >> > > > >> > > 63. My point is that once a leader is producing a message > > in > > >> the > > >> > > new > > >> > > > >> > > partition_epoch, ideally, we should associate the new > > offsets > > >> > with > > >> > > > the > > >> > > > >> > new > > >> > > > >> > > partition_epoch. Otherwise, the offset_epoch won't be the > > >> > correct > > >> > > > >> unique > > >> > > > >> > > identifier (useful for solving other problems mentioned > > >> above). > > >> > I > > >> > > > was > > >> > > > >> > > originally thinking that the leader will include the > > >> > > partition_epoch > > >> > > > >> in > > >> > > > >> > the > > >> > > > >> > > metadata cache in the fetch response. It's just that > right > > >> now, > > >> > > > >> metadata > > >> > > > >> > > cache is updated on UpdateMetadataRequest, which > typically > > >> > happens > > >> > > > >> after > > >> > > > >> > > the LeaderAndIsrRequest. Another approach is for the > leader > > >> to > > >> > > cache > > >> > > > >> the > > >> > > > >> > > partition_epoch in the Partition object and return that > > >> (instead > > >> > > of > > >> > > > >> the > > >> > > > >> > one > > >> > > > >> > > in metadata cache) in the fetch response. > > >> > > > >> > > > > >> > > > >> > > 65. It seems to me that the global_epoch and the > > >> partition_epoch > > >> > > > have > > >> > > > >> > > different purposes. A partition_epoch has the benefit > that > > it > > >> > (1) > > >> > > > can > > >> > > > >> be > > >> > > > >> > > used to form a unique identifier for a message and (2) > can > > be > > >> > used > > >> > > > to > > >> > > > >> > > solve other > > >> > > > >> > > corner case problems in the future. I am not sure having > > >> just a > > >> > > > >> > > global_epoch can achieve these. global_epoch is useful to > > >> > > determine > > >> > > > >> which > > >> > > > >> > > version of the metadata is newer, especially with topic > > >> > deletion. > > >> > > > >> > > > > >> > > > >> > > Thanks, > > >> > > > >> > > > > >> > > > >> > > Jun > > >> > > > >> > > > > >> > > > >> > > On Tue, Jan 9, 2018 at 11:34 PM, Dong Lin < > > >> lindon...@gmail.com> > > >> > > > >> wrote: > > >> > > > >> > > > > >> > > > >> > > > Regarding the use of the global epoch in 65), it is > very > > >> > similar > > >> > > > to > > >> > > > >> the > > >> > > > >> > > > proposal of the metadata_epoch we discussed earlier. > The > > >> main > > >> > > > >> > difference > > >> > > > >> > > is > > >> > > > >> > > > that this epoch is incremented when we > > create/expand/delete > > >> > > topic > > >> > > > >> and > > >> > > > >> > > does > > >> > > > >> > > > not change when controller re-send metadata. > > >> > > > >> > > > > > >> > > > >> > > > I looked at our previous discussion. It seems that we > > >> prefer > > >> > > > >> > > > partition_epoch over the metadata_epoch because 1) we > > >> prefer > > >> > not > > >> > > > to > > >> > > > >> > have > > >> > > > >> > > an > > >> > > > >> > > > ever growing metadata_epoch and 2) we can reset offset > > >> better > > >> > > when > > >> > > > >> > topic > > >> > > > >> > > is > > >> > > > >> > > > re-created. The use of global topic_epoch avoids the > > >> drawback > > >> > of > > >> > > > an > > >> > > > >> > ever > > >> > > > >> > > > quickly ever growing metadata_epoch. Though the global > > >> epoch > > >> > > does > > >> > > > >> not > > >> > > > >> > > allow > > >> > > > >> > > > us to recognize the invalid offset committed before the > > >> topic > > >> > > > >> > > re-creation, > > >> > > > >> > > > we can probably just delete the offset when we delete a > > >> topic. > > >> > > > Thus > > >> > > > >> I > > >> > > > >> > am > > >> > > > >> > > > not very sure whether it is still worthwhile to have a > > >> > > > per-partition > > >> > > > >> > > > partition_epoch if the metadata already has the global > > >> epoch. > > >> > > > >> > > > > > >> > > > >> > > > > > >> > > > >> > > > On Tue, Jan 9, 2018 at 6:58 PM, Dong Lin < > > >> lindon...@gmail.com > > >> > > > > >> > > > >> wrote: > > >> > > > >> > > > > > >> > > > >> > > > > Hey Jun, > > >> > > > >> > > > > > > >> > > > >> > > > > Thanks so much. These comments very useful. Please > see > > >> below > > >> > > my > > >> > > > >> > > comments. > > >> > > > >> > > > > > > >> > > > >> > > > > On Mon, Jan 8, 2018 at 5:52 PM, Jun Rao < > > >> j...@confluent.io> > > >> > > > wrote: > > >> > > > >> > > > > > > >> > > > >> > > > >> Hi, Dong, > > >> > > > >> > > > >> > > >> > > > >> > > > >> Thanks for the updated KIP. A few more comments. > > >> > > > >> > > > >> > > >> > > > >> > > > >> 60. Perhaps having a partition epoch is more > flexible > > >> since > > >> > > in > > >> > > > >> the > > >> > > > >> > > > future, > > >> > > > >> > > > >> we may support deleting a partition as well. > > >> > > > >> > > > >> > > >> > > > >> > > > > > > >> > > > >> > > > > Yeah I have considered this. I think we can probably > > >> still > > >> > > > support > > >> > > > >> > > > > deleting a partition by using the topic_epoch -- when > > >> > > partition > > >> > > > >> of a > > >> > > > >> > > > topic > > >> > > > >> > > > > is deleted or created, epoch of all partitions of > this > > >> topic > > >> > > > will > > >> > > > >> be > > >> > > > >> > > > > incremented by 1. Therefore, if that partition is > > >> re-created > > >> > > > >> later, > > >> > > > >> > the > > >> > > > >> > > > > epoch of that partition will still be larger than its > > >> epoch > > >> > > > before > > >> > > > >> > the > > >> > > > >> > > > > deletion, which still allows the client to order the > > >> > metadata > > >> > > > for > > >> > > > >> the > > >> > > > >> > > > > purpose of this KIP. Does this sound reasonable? > > >> > > > >> > > > > > > >> > > > >> > > > > The advantage of using topic_epoch instead of > > >> > partition_epoch > > >> > > is > > >> > > > >> that > > >> > > > >> > > the > > >> > > > >> > > > > size of the /brokers/topics/[topic] znode and > > >> > request/response > > >> > > > >> size > > >> > > > >> > can > > >> > > > >> > > > be > > >> > > > >> > > > > smaller. We have a limit on the maximum size of znode > > >> > > (typically > > >> > > > >> > 1MB). > > >> > > > >> > > > Use > > >> > > > >> > > > > partition epoch can effectively reduce the number of > > >> > > partitions > > >> > > > >> that > > >> > > > >> > > can > > >> > > > >> > > > be > > >> > > > >> > > > > described by the /brokers/topics/[topic] znode. > > >> > > > >> > > > > > > >> > > > >> > > > > One use-case of partition_epoch for client to detect > > that > > >> > the > > >> > > > >> > committed > > >> > > > >> > > > > offset, either from kafka offset topic or from the > > >> external > > >> > > > store > > >> > > > >> is > > >> > > > >> > > > > invalid after partition deletion and re-creation. > > >> However, > > >> > it > > >> > > > >> seems > > >> > > > >> > > that > > >> > > > >> > > > we > > >> > > > >> > > > > can also address this use-case with other approaches. > > For > > >> > > > example, > > >> > > > >> > when > > >> > > > >> > > > > AdminClient deletes partitions, it can also delete > the > > >> > > committed > > >> > > > >> > > offsets > > >> > > > >> > > > > for those partitions from the offset topic. If user > > >> stores > > >> > > > offset > > >> > > > >> > > > > externally, it might make sense for user to similarly > > >> remove > > >> > > > >> offsets > > >> > > > >> > of > > >> > > > >> > > > > related partitions after these partitions are > deleted. > > >> So I > > >> > am > > >> > > > not > > >> > > > >> > sure > > >> > > > >> > > > > that we should use partition_epoch in this KIP. > > >> > > > >> > > > > > > >> > > > >> > > > > > > >> > > > >> > > > >> > > >> > > > >> > > > >> 61. It seems that the leader epoch returned in the > > >> > position() > > >> > > > >> call > > >> > > > >> > > > should > > >> > > > >> > > > >> the the leader epoch returned in the fetch response, > > not > > >> > the > > >> > > > one > > >> > > > >> in > > >> > > > >> > > the > > >> > > > >> > > > >> metadata cache of the client. > > >> > > > >> > > > > > > >> > > > >> > > > > > > >> > > > >> > > > > I think this is a good idea. Just to double check, > this > > >> > change > > >> > > > >> does > > >> > > > >> > not > > >> > > > >> > > > > affect the correctness or performance of this KIP. > But > > it > > >> > can > > >> > > be > > >> > > > >> > useful > > >> > > > >> > > > if > > >> > > > >> > > > > we want to use the leader_epoch to better handle the > > >> offset > > >> > > rest > > >> > > > >> in > > >> > > > >> > > case > > >> > > > >> > > > of > > >> > > > >> > > > > unclean leader election, which is listed in the > future > > >> work. > > >> > > Is > > >> > > > >> this > > >> > > > >> > > > > understanding correct? > > >> > > > >> > > > > > > >> > > > >> > > > > I have updated the KIP to specify that the > leader_epoch > > >> > > returned > > >> > > > >> by > > >> > > > >> > > > > position() should be the largest leader_epoch of > those > > >> > already > > >> > > > >> > consumed > > >> > > > >> > > > > messages whose offset < position. If no message has > > been > > >> > > > consumed > > >> > > > >> > since > > >> > > > >> > > > > consumer initialization, the leader_epoch from seek() > > or > > >> > > > >> > > > > OffsetFetchResponse should be used. The offset > included > > >> in > > >> > the > > >> > > > >> > > > > OffsetCommitRequest will also be determined in the > > >> similar > > >> > > > manner. > > >> > > > >> > > > > > > >> > > > >> > > > > > > >> > > > >> > > > >> > > >> > > > >> > > > >> 62. I am wondering if we should return the partition > > >> epoch > > >> > in > > >> > > > the > > >> > > > >> > > fetch > > >> > > > >> > > > >> response as well. In the current proposal, if a > topic > > is > > >> > > > >> recreated > > >> > > > >> > and > > >> > > > >> > > > the > > >> > > > >> > > > >> new leader is on the same broker as the old one, > there > > >> is > > >> > > > >> nothing to > > >> > > > >> > > > force > > >> > > > >> > > > >> the metadata refresh in the client. So, the client > may > > >> > still > > >> > > > >> > associate > > >> > > > >> > > > the > > >> > > > >> > > > >> offset with the old partition epoch. > > >> > > > >> > > > >> > > >> > > > >> > > > > > > >> > > > >> > > > > Could you help me understand the problem if a client > > >> > > associates > > >> > > > >> old > > >> > > > >> > > > > partition_epoch (or the topic_epoch as of the current > > >> KIP) > > >> > > with > > >> > > > >> the > > >> > > > >> > > > offset? > > >> > > > >> > > > > The main purpose of the topic_epoch is to be able to > > drop > > >> > > > >> > leader_epoch > > >> > > > >> > > > to 0 > > >> > > > >> > > > > after a partition is deleted and re-created. I guess > > you > > >> may > > >> > > be > > >> > > > >> > > thinking > > >> > > > >> > > > > about using the partition_epoch to detect that the > > >> committed > > >> > > > >> offset > > >> > > > >> > is > > >> > > > >> > > > > invalid? In that case, I am wondering if the > > alternative > > >> > > > approach > > >> > > > >> > > > described > > >> > > > >> > > > > in 60) would be reasonable. > > >> > > > >> > > > > > > >> > > > >> > > > > > > >> > > > >> > > > >> > > >> > > > >> > > > >> 63. There is some subtle coordination between the > > >> > > > >> > LeaderAndIsrRequest > > >> > > > >> > > > and > > >> > > > >> > > > >> UpdateMetadataRequest. Currently, when a leader > > changes, > > >> > the > > >> > > > >> > > controller > > >> > > > >> > > > >> first sends the LeaderAndIsrRequest to the assigned > > >> > replicas > > >> > > > and > > >> > > > >> the > > >> > > > >> > > > >> UpdateMetadataRequest to every broker. So, there > could > > >> be a > > >> > > > small > > >> > > > >> > > window > > >> > > > >> > > > >> when the leader already receives the new partition > > >> epoch in > > >> > > the > > >> > > > >> > > > >> LeaderAndIsrRequest, but the metadata cache in the > > >> broker > > >> > > > hasn't > > >> > > > >> > been > > >> > > > >> > > > >> updated with the latest partition epoch. Not sure > > what's > > >> > the > > >> > > > best > > >> > > > >> > way > > >> > > > >> > > to > > >> > > > >> > > > >> address this issue. Perhaps we can update the > metadata > > >> > cache > > >> > > on > > >> > > > >> the > > >> > > > >> > > > broker > > >> > > > >> > > > >> with both LeaderAndIsrRequest and > > UpdateMetadataRequest. > > >> > The > > >> > > > >> > challenge > > >> > > > >> > > > is > > >> > > > >> > > > >> that the two have slightly different data. For > > example, > > >> > only > > >> > > > the > > >> > > > >> > > latter > > >> > > > >> > > > >> has > > >> > > > >> > > > >> all endpoints. > > >> > > > >> > > > >> > > >> > > > >> > > > > > > >> > > > >> > > > > I am not sure whether this is a problem. Could you > > >> explain a > > >> > > bit > > >> > > > >> more > > >> > > > >> > > > what > > >> > > > >> > > > > specific problem this small window can cause? > > >> > > > >> > > > > > > >> > > > >> > > > > Since client can fetch metadata from any broker in > the > > >> > > cluster, > > >> > > > >> and > > >> > > > >> > > given > > >> > > > >> > > > > that different brokers receive request (e.g. > > >> > > LeaderAndIsrRequest > > >> > > > >> and > > >> > > > >> > > > > UpdateMetadataRequest) in arbitrary order, the > metadata > > >> > > received > > >> > > > >> by > > >> > > > >> > > > client > > >> > > > >> > > > > can be in arbitrary order (either newer or older) > > >> compared > > >> > to > > >> > > > the > > >> > > > >> > > > broker's > > >> > > > >> > > > > leadership state even if a given broker receives > > >> > > > >> LeaderAndIsrRequest > > >> > > > >> > > and > > >> > > > >> > > > > UpdateMetadataRequest simultaneously. So I am not > sure > > >> it is > > >> > > > >> useful > > >> > > > >> > to > > >> > > > >> > > > > update broker's cache with LeaderAndIsrRequest. > > >> > > > >> > > > > > > >> > > > >> > > > > > > >> > > > >> > > > >> 64. The enforcement of leader epoch in Offset > commit: > > We > > >> > > allow > > >> > > > a > > >> > > > >> > > > consumer > > >> > > > >> > > > >> to set an arbitrary offset. So it's possible for > > >> offsets or > > >> > > > >> leader > > >> > > > >> > > epoch > > >> > > > >> > > > >> to > > >> > > > >> > > > >> go backwards. I am not sure if we could always > enforce > > >> that > > >> > > the > > >> > > > >> > leader > > >> > > > >> > > > >> epoch only goes up on the broker. > > >> > > > >> > > > >> > > >> > > > >> > > > > > > >> > > > >> > > > > Sure. I have removed this check from the KIP. > > >> > > > >> > > > > > > >> > > > >> > > > > BTW, we can probably still ensure that the > leader_epoch > > >> > always > > >> > > > >> > increase > > >> > > > >> > > > if > > >> > > > >> > > > > the leader_epoch used with offset commit is the > > >> > > max(leader_epoch > > >> > > > >> of > > >> > > > >> > the > > >> > > > >> > > > > message with offset = the committed offset - 1, the > > >> largest > > >> > > > known > > >> > > > >> > > > > leader_epoch from the metadata). But I don't have a > > good > > >> > > > use-case > > >> > > > >> for > > >> > > > >> > > > this > > >> > > > >> > > > > alternative definition. So I choose the keep the KIP > > >> simple > > >> > by > > >> > > > >> > > requiring > > >> > > > >> > > > > leader_epoch to always increase. > > >> > > > >> > > > > > > >> > > > >> > > > > > > >> > > > >> > > > >> 65. Good point on handling missing partition epoch > due > > >> to > > >> > > topic > > >> > > > >> > > > deletion. > > >> > > > >> > > > >> Another potential way to address this is to > > additionally > > >> > > > >> propagate > > >> > > > >> > the > > >> > > > >> > > > >> global partition epoch to brokers and the clients. > > This > > >> > way, > > >> > > > >> when a > > >> > > > >> > > > >> partition epoch is missing, we can use the global > > >> partition > > >> > > > >> epoch to > > >> > > > >> > > > >> reason > > >> > > > >> > > > >> about which metadata is more recent. > > >> > > > >> > > > >> > > >> > > > >> > > > > > > >> > > > >> > > > > This is a great idea. The global epoch can be used to > > >> order > > >> > > the > > >> > > > >> > > metadata > > >> > > > >> > > > > and help us recognize the more recent metadata if a > > topic > > >> > (or > > >> > > > >> > > partition) > > >> > > > >> > > > is > > >> > > > >> > > > > deleted and re-created. > > >> > > > >> > > > > > > >> > > > >> > > > > Actually, it seems we only need to propagate the > global > > >> > epoch > > >> > > to > > >> > > > >> > > brokers > > >> > > > >> > > > > and clients without propagating this epoch on a > > >> per-topic or > > >> > > > >> > > > per-partition > > >> > > > >> > > > > basic. Doing so would simply interface changes made > > this > > >> > KIP. > > >> > > > Does > > >> > > > >> > this > > >> > > > >> > > > > approach sound reasonable? > > >> > > > >> > > > > > > >> > > > >> > > > > > > >> > > > >> > > > >> 66. A client may also get an offset by time using > the > > >> > > > >> > offsetForTimes() > > >> > > > >> > > > >> api. > > >> > > > >> > > > >> So, we probably want to include > offsetInternalMetadata > > >> in > > >> > > > >> > > > >> OffsetAndTimestamp > > >> > > > >> > > > >> as well. > > >> > > > >> > > > >> > > >> > > > >> > > > > > > >> > > > >> > > > > You are right. This probably also requires us to > change > > >> the > > >> > > > >> > > > > ListOffsetRequest as well. I will update the KIP > after > > we > > >> > > agree > > >> > > > on > > >> > > > >> > the > > >> > > > >> > > > > solution for 65). > > >> > > > >> > > > > > > >> > > > >> > > > > > > >> > > > >> > > > >> > > >> > > > >> > > > >> 67. InteralMetadata can be a bit confusing with the > > >> > metadata > > >> > > > >> field > > >> > > > >> > > > already > > >> > > > >> > > > >> there. Perhaps we can just call it OffsetEpoch. It > > >> might be > > >> > > > >> useful > > >> > > > >> > to > > >> > > > >> > > > make > > >> > > > >> > > > >> OffsetEpoch printable at least for debugging > purpose. > > >> Once > > >> > > you > > >> > > > do > > >> > > > >> > > that, > > >> > > > >> > > > we > > >> > > > >> > > > >> are already exposing the internal fields. So, not > sure > > >> if > > >> > > it's > > >> > > > >> worth > > >> > > > >> > > > >> hiding > > >> > > > >> > > > >> them. If we do want to hide them, perhaps we can > have > > >> sth > > >> > > like > > >> > > > >> the > > >> > > > >> > > > >> following. The binary encoding is probably more > > >> efficient > > >> > > than > > >> > > > >> JSON > > >> > > > >> > > for > > >> > > > >> > > > >> external storage. > > >> > > > >> > > > >> > > >> > > > >> > > > >> OffsetEpoch { > > >> > > > >> > > > >> static OffsetEpoch decode(byte[]); > > >> > > > >> > > > >> > > >> > > > >> > > > >> public byte[] encode(); > > >> > > > >> > > > >> > > >> > > > >> > > > >> public String toString(); > > >> > > > >> > > > >> } > > >> > > > >> > > > >> > > >> > > > >> > > > > > > >> > > > >> > > > > Thanks much. I like this solution. I have updated the > > KIP > > >> > > > >> > accordingly. > > >> > > > >> > > > > > > >> > > > >> > > > > > > >> > > > >> > > > > > > >> > > > >> > > > >> > > >> > > > >> > > > >> Jun > > >> > > > >> > > > >> > > >> > > > >> > > > >> On Mon, Jan 8, 2018 at 4:22 PM, Dong Lin < > > >> > > lindon...@gmail.com> > > >> > > > >> > wrote: > > >> > > > >> > > > >> > > >> > > > >> > > > >> > Hey Jason, > > >> > > > >> > > > >> > > > >> > > > >> > > > >> > Certainly. This sounds good. I have updated the > KIP > > to > > >> > > > clarity > > >> > > > >> > that > > >> > > > >> > > > the > > >> > > > >> > > > >> > global epoch will be incremented by 1 each time a > > >> topic > > >> > is > > >> > > > >> > deleted. > > >> > > > >> > > > >> > > > >> > > > >> > > > >> > Thanks, > > >> > > > >> > > > >> > Dong > > >> > > > >> > > > >> > > > >> > > > >> > > > >> > On Mon, Jan 8, 2018 at 4:09 PM, Jason Gustafson < > > >> > > > >> > ja...@confluent.io > > >> > > > >> > > > > > >> > > > >> > > > >> > wrote: > > >> > > > >> > > > >> > > > >> > > > >> > > > >> > > Hi Dong, > > >> > > > >> > > > >> > > > > >> > > > >> > > > >> > > > > >> > > > >> > > > >> > > I think your approach will allow user to > > distinguish > > >> > > > between > > >> > > > >> the > > >> > > > >> > > > >> metadata > > >> > > > >> > > > >> > > > before and after the topic deletion. I also > > agree > > >> > that > > >> > > > this > > >> > > > >> > can > > >> > > > >> > > be > > >> > > > >> > > > >> > > > potentially be useful to user. I am just not > > very > > >> > sure > > >> > > > >> whether > > >> > > > >> > > we > > >> > > > >> > > > >> > already > > >> > > > >> > > > >> > > > have a good use-case to make the additional > > >> > complexity > > >> > > > >> > > worthwhile. > > >> > > > >> > > > >> It > > >> > > > >> > > > >> > > seems > > >> > > > >> > > > >> > > > that this feature is kind of independent of > the > > >> main > > >> > > > >> problem > > >> > > > >> > of > > >> > > > >> > > > this > > >> > > > >> > > > >> > KIP. > > >> > > > >> > > > >> > > > Could we add this as a future work? > > >> > > > >> > > > >> > > > > >> > > > >> > > > >> > > > > >> > > > >> > > > >> > > Do you think it's fair if we bump the topic > epoch > > on > > >> > > > deletion > > >> > > > >> > and > > >> > > > >> > > > >> leave > > >> > > > >> > > > >> > > propagation of the epoch for deleted topics for > > >> future > > >> > > > work? > > >> > > > >> I > > >> > > > >> > > don't > > >> > > > >> > > > >> > think > > >> > > > >> > > > >> > > this adds much complexity and it makes the > > behavior > > >> > > > >> consistent: > > >> > > > >> > > > every > > >> > > > >> > > > >> > topic > > >> > > > >> > > > >> > > mutation results in an epoch bump. > > >> > > > >> > > > >> > > > > >> > > > >> > > > >> > > Thanks, > > >> > > > >> > > > >> > > Jason > > >> > > > >> > > > >> > > > > >> > > > >> > > > >> > > On Mon, Jan 8, 2018 at 3:14 PM, Dong Lin < > > >> > > > >> lindon...@gmail.com> > > >> > > > >> > > > wrote: > > >> > > > >> > > > >> > > > > >> > > > >> > > > >> > > > Hey Ismael, > > >> > > > >> > > > >> > > > > > >> > > > >> > > > >> > > > I guess we actually need user to see this > field > > so > > >> > that > > >> > > > >> user > > >> > > > >> > can > > >> > > > >> > > > >> store > > >> > > > >> > > > >> > > this > > >> > > > >> > > > >> > > > value in the external store together with the > > >> offset. > > >> > > We > > >> > > > >> just > > >> > > > >> > > > prefer > > >> > > > >> > > > >> > the > > >> > > > >> > > > >> > > > value to be opaque to discourage most users > from > > >> > > > >> interpreting > > >> > > > >> > > this > > >> > > > >> > > > >> > value. > > >> > > > >> > > > >> > > > One more advantage of using such an opaque > field > > >> is > > >> > to > > >> > > be > > >> > > > >> able > > >> > > > >> > > to > > >> > > > >> > > > >> > evolve > > >> > > > >> > > > >> > > > the information (or schema) of this value > > without > > >> > > > changing > > >> > > > >> > > > consumer > > >> > > > >> > > > >> API > > >> > > > >> > > > >> > > in > > >> > > > >> > > > >> > > > the future. > > >> > > > >> > > > >> > > > > > >> > > > >> > > > >> > > > I also thinking it is probably OK for user to > be > > >> able > > >> > > to > > >> > > > >> > > interpret > > >> > > > >> > > > >> this > > >> > > > >> > > > >> > > > value, particularly for those advanced users. > > >> > > > >> > > > >> > > > > > >> > > > >> > > > >> > > > Thanks, > > >> > > > >> > > > >> > > > Dong > > >> > > > >> > > > >> > > > > > >> > > > >> > > > >> > > > On Mon, Jan 8, 2018 at 2:34 PM, Ismael Juma < > > >> > > > >> > ism...@juma.me.uk> > > >> > > > >> > > > >> wrote: > > >> > > > >> > > > >> > > > > > >> > > > >> > > > >> > > > > On Fri, Jan 5, 2018 at 7:15 PM, Jason > > Gustafson > > >> < > > >> > > > >> > > > >> ja...@confluent.io> > > >> > > > >> > > > >> > > > > wrote: > > >> > > > >> > > > >> > > > > > > > >> > > > >> > > > >> > > > > > class OffsetAndMetadata { > > >> > > > >> > > > >> > > > > > long offset; > > >> > > > >> > > > >> > > > > > byte[] offsetMetadata; > > >> > > > >> > > > >> > > > > > String metadata; > > >> > > > >> > > > >> > > > > > } > > >> > > > >> > > > >> > > > > > > >> > > > >> > > > >> > > > > > > >> > > > >> > > > >> > > > > > Admittedly, the naming is a bit annoying, > > but > > >> we > > >> > > can > > >> > > > >> > > probably > > >> > > > >> > > > >> come > > >> > > > >> > > > >> > up > > >> > > > >> > > > >> > > > > with > > >> > > > >> > > > >> > > > > > something better. Internally the byte > array > > >> would > > >> > > > have > > >> > > > >> a > > >> > > > >> > > > >> version. > > >> > > > >> > > > >> > If > > >> > > > >> > > > >> > > in > > >> > > > >> > > > >> > > > > the > > >> > > > >> > > > >> > > > > > future we have anything else we need to > add, > > >> we > > >> > can > > >> > > > >> update > > >> > > > >> > > the > > >> > > > >> > > > >> > > version > > >> > > > >> > > > >> > > > > and > > >> > > > >> > > > >> > > > > > we wouldn't need any new APIs. > > >> > > > >> > > > >> > > > > > > > >> > > > >> > > > >> > > > > > > >> > > > >> > > > >> > > > > We can also add fields to a class in a > > >> compatible > > >> > > way. > > >> > > > >> So, > > >> > > > >> > it > > >> > > > >> > > > >> seems > > >> > > > >> > > > >> > to > > >> > > > >> > > > >> > > me > > >> > > > >> > > > >> > > > > that the main advantage of the byte array is > > >> that > > >> > > it's > > >> > > > >> > opaque > > >> > > > >> > > to > > >> > > > >> > > > >> the > > >> > > > >> > > > >> > > > user. > > >> > > > >> > > > >> > > > > Is that correct? If so, we could also add > any > > >> > opaque > > >> > > > >> > metadata > > >> > > > >> > > > in a > > >> > > > >> > > > >> > > > subclass > > >> > > > >> > > > >> > > > > so that users don't even see it (unless they > > >> cast > > >> > it, > > >> > > > but > > >> > > > >> > then > > >> > > > >> > > > >> > they're > > >> > > > >> > > > >> > > on > > >> > > > >> > > > >> > > > > their own). > > >> > > > >> > > > >> > > > > > > >> > > > >> > > > >> > > > > Ismael > > >> > > > >> > > > >> > > > > > > >> > > > >> > > > >> > > > > The corresponding seek() and position() APIs > > >> might > > >> > > look > > >> > > > >> > > > something > > >> > > > >> > > > >> > like > > >> > > > >> > > > >> > > > > this: > > >> > > > >> > > > >> > > > > > > > >> > > > >> > > > >> > > > > > void seek(TopicPartition partition, long > > >> offset, > > >> > > > byte[] > > >> > > > >> > > > >> > > > offsetMetadata); > > >> > > > >> > > > >> > > > > > byte[] positionMetadata(TopicPartition > > >> > partition); > > >> > > > >> > > > >> > > > > > > > >> > > > >> > > > >> > > > > > What do you think? > > >> > > > >> > > > >> > > > > > > > >> > > > >> > > > >> > > > > > Thanks, > > >> > > > >> > > > >> > > > > > Jason > > >> > > > >> > > > >> > > > > > > > >> > > > >> > > > >> > > > > > On Thu, Jan 4, 2018 at 7:04 PM, Dong Lin < > > >> > > > >> > > lindon...@gmail.com > > >> > > > >> > > > > > > >> > > > >> > > > >> > > wrote: > > >> > > > >> > > > >> > > > > > > > >> > > > >> > > > >> > > > > > > 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 > > >> > > > >> > > > >> > > > > > > >> > > > > > > > > > >> > > > >> > > > >> > > > > > > >> > > > > > > > > >> > > > >> > > > >> > > > > > > >> > > > > > > > >> > > > >> > > > >> > > > > > > >> > > > > > > >> > > > >> > > > >> > > > > > > >> > > > > > >> > > > >> > > > >> > > > > > > >> > > > > >> > > > >> > > > >> > > > > > > >> > > > >> > > > >> > > > >> > > > > > > >> > > >> > > > >> > > > >> > > > > > > > > > >> > > > >> > > > >> > > > > > > > > > >> > > > >> > > > >> > > > > > > > > >> > > > >> > > > >> > > > > > > > >> > > > >> > > > >> > > > > > > >> > > > >> > > > >> > > > > > >> > > > >> > > > >> > > > > >> > > > >> > > > >> > > > >> > > > >> > > > >> > > >> > > > >> > > > > > > >> > > > >> > > > > > > >> > > > >> > > > > > >> > > > >> > > > > >> > > > >> > > > >> > > > >> > > >> > > > > > > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > > > > > > > >