kafka-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Dong Lin <lindon...@gmail.com>
Subject Re: [DISCUSS] KIP-232: Detect outdated metadata by adding ControllerMetadataEpoch field
Date Fri, 05 Jan 2018 01:36:01 GMT
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 <jun@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 <lindong28@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 <jun@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 <lindong28@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 <jun@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 <lindong28@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 <jun@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 <lindong28@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
> > > > > > > >
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
>

Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message