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]/partitions/[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 > > > >> > > > >> > > > > > > >> > > > > > > > > > > >> > > > >> > > > > > > >> > > > > > > > > > >> > > > >> > > > > > > >> > > > > > > > > >> > > > >> > > > > > > >> > > > > > > > >> > > > >> > > > > > > >> > > > > > > >> > > > >> > > > > > > >> > > > > > >> > > > >> > > > > > > >> > > > > >> > > > >> > > > > > > >> > > > >> > > > >> > > > > > > > > > > >> > > > >> > > > > > > > > > > >> > > > >> > > > > > > > > > >> > > > >> > > > > > > > > >> > > > >> > > > > > > > >> > > > >> > > > > > > >> > > > >> > > > > > >> > > > >> > > > > >> > > > >> > > > >> > > > > > > > >> > > > > > > > >> > > > > > > >> > > > > > >> > > > > >> > > > > > > > > > > > > > >