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
>> > > > >> > > > > > > >> > > > > > > >
>> > > > >> > > > > > > >> > > > > > >
>> > > > >> > > > > > > >> > > > > >
>> > > > >> > > > > > > >> > > > >
>> > > > >> > > > > > > >> > > >
>> > > > >> > > > > > > >> > >
>> > > > >> > > > > > > >> >
>> > > > >> > > > > > > >>
>> > > > >> > > > > > > >
>> > > > >> > > > > > > >
>> > > > >> > > > > > >
>> > > > >> > > > > >
>> > > > >> > > > >
>> > > > >> > > >
>> > > > >> > >
>> > > > >> >
>> > > > >>
>> > > > >
>> > > > >
>> > > >
>> > >
>> >
>>
>
>
