Hi, Dong,
Sorry for the late response. Since KIP-320 is covering some of the similar
problems described in this KIP, perhaps we can wait until KIP-320 settles
and see what's still left uncovered in this KIP.
Thanks,
Jun
On Mon, Jun 4, 2018 at 7:03 PM, Dong Lin <lindon...@gmail.com> wrote:
> Hey Jun,
>
> It seems that we have made considerable progress on the discussion of
> KIP-253 since February. Do you think we should continue the discussion
> there, or can we continue the voting for this KIP? I am happy to submit the
> PR and move forward the progress for this KIP.
>
> Thanks!
> Dong
>
>
> On Wed, Feb 7, 2018 at 11:42 PM, Dong Lin <lindon...@gmail.com> wrote:
>
> > Hey Jun,
> >
> > Sure, I will come up with a KIP this week. I think there is a way to
> allow
> > partition expansion to arbitrary number without introducing new concepts
> > such as read-only partition or repartition epoch.
> >
> > Thanks,
> > Dong
> >
> > On Wed, Feb 7, 2018 at 5:28 PM, Jun Rao <j...@confluent.io> wrote:
> >
> >> Hi, Dong,
> >>
> >> Thanks for the reply. The general idea that you had for adding
> partitions
> >> is similar to what we had in mind. It would be useful to make this more
> >> general, allowing adding an arbitrary number of partitions (instead of
> >> just
> >> doubling) and potentially removing partitions as well. The following is
> >> the
> >> high level idea from the discussion with Colin, Jason and Ismael.
> >>
> >> * To change the number of partitions from X to Y in a topic, the
> >> controller
> >> marks all existing X partitions as read-only and creates Y new
> partitions.
> >> The new partitions are writable and are tagged with a higher repartition
> >> epoch (RE).
> >>
> >> * The controller propagates the new metadata to every broker. Once the
> >> leader of a partition is marked as read-only, it rejects the produce
> >> requests on this partition. The producer will then refresh the metadata
> >> and
> >> start publishing to the new writable partitions.
> >>
> >> * The consumers will then be consuming messages in RE order. The
> consumer
> >> coordinator will only assign partitions in the same RE to consumers.
> Only
> >> after all messages in an RE are consumed, will partitions in a higher RE
> >> be
> >> assigned to consumers.
> >>
> >> As Colin mentioned, if we do the above, we could potentially (1) use a
> >> globally unique partition id, or (2) use a globally unique topic id to
> >> distinguish recreated partitions due to topic deletion.
> >>
> >> So, perhaps we can sketch out the re-partitioning KIP a bit more and see
> >> if
> >> there is any overlap with KIP-232. Would you be interested in doing
> that?
> >> If not, we can do that next week.
> >>
> >> Jun
> >>
> >>
> >> On Tue, Feb 6, 2018 at 11:30 AM, Dong Lin <lindon...@gmail.com> wrote:
> >>
> >> > Hey Jun,
> >> >
> >> > Interestingly I am also planning to sketch a KIP to allow partition
> >> > expansion for keyed topics after this KIP. Since you are already doing
> >> > that, I guess I will just share my high level idea here in case it is
> >> > helpful.
> >> >
> >> > The motivation for the KIP is that we currently lose order guarantee
> for
> >> > messages with the same key if we expand partitions of keyed topic.
> >> >
> >> > The solution can probably be built upon the following ideas:
> >> >
> >> > - Partition number of the keyed topic should always be doubled (or
> >> > multiplied by power of 2). Given that we select a partition based on
> >> > hash(key) % partitionNum, this should help us ensure that, a message
> >> > assigned to an existing partition will not be mapped to another
> existing
> >> > partition after partition expansion.
> >> >
> >> > - Producer includes in the ProduceRequest some information that helps
> >> > ensure that messages produced ti a partition will monotonically
> >> increase in
> >> > the partitionNum of the topic. In other words, if broker receives a
> >> > ProduceRequest and notices that the producer does not know the
> partition
> >> > number has increased, broker should reject this request. That
> >> "information"
> >> > maybe leaderEpoch, max partitionEpoch of the partitions of the topic,
> or
> >> > simply partitionNum of the topic. The benefit of this property is that
> >> we
> >> > can keep the new logic for in-order message consumption entirely in
> how
> >> > consumer leader determines the partition -> consumer mapping.
> >> >
> >> > - When consumer leader determines partition -> consumer mapping,
> leader
> >> > first reads the start position for each partition using
> >> OffsetFetchRequest.
> >> > If start position are all non-zero, then assignment can be done in its
> >> > current manner. The assumption is that, a message in the new partition
> >> > should only be consumed after all messages with the same key produced
> >> > before it has been consumed. Since some messages in the new partition
> >> has
> >> > been consumed, we should not worry about consuming messages
> >> out-of-order.
> >> > This benefit of this approach is that we can avoid unnecessary
> overhead
> >> in
> >> > the common case.
> >> >
> >> > - If the consumer leader finds that the start position for some
> >> partition
> >> > is 0. Say the current partition number is 18 and the partition index
> is
> >> 12,
> >> > then consumer leader should ensure that messages produced to partition
> >> 12 -
> >> > 18/2 = 3 before the first message of partition 12 is consumed, before
> it
> >> > assigned partition 12 to any consumer in the consumer group. Since we
> >> have
> >> > a "information" that is monotonically increasing per partition,
> consumer
> >> > can read the value of this information from the first message in
> >> partition
> >> > 12, get the offset corresponding to this value in partition 3, assign
> >> > partition except for partition 12 (and probably other new partitions)
> to
> >> > the existing consumers, waiting for the committed offset to go beyond
> >> this
> >> > offset for partition 3, and trigger rebalance again so that partition
> 3
> >> can
> >> > be reassigned to some consumer.
> >> >
> >> >
> >> > Thanks,
> >> > Dong
> >> >
> >> >
> >> > On Tue, Feb 6, 2018 at 10:10 AM, Jun Rao <j...@confluent.io> wrote:
> >> >
> >> > > Hi, Dong,
> >> > >
> >> > > Thanks for the KIP. It looks good overall. We are working on a
> >> separate
> >> > KIP
> >> > > for adding partitions while preserving the ordering guarantees. That
> >> may
> >> > > require another flavor of partition epoch. It's not very clear
> whether
> >> > that
> >> > > partition epoch can be merged with the partition epoch in this KIP.
> >> So,
> >> > > perhaps you can wait on this a bit until we post the other KIP in
> the
> >> > next
> >> > > few days.
> >> > >
> >> > > Jun
> >> > >
> >> > >
> >> > >
> >> > > On Mon, Feb 5, 2018 at 2:43 PM, Becket Qin <becket....@gmail.com>
> >> wrote:
> >> > >
> >> > > > +1 on the KIP.
> >> > > >
> >> > > > I think the KIP is mainly about adding the capability of tracking
> >> the
> >> > > > system state change lineage. It does not seem necessary to bundle
> >> this
> >> > > KIP
> >> > > > with replacing the topic partition with partition epoch in
> >> > produce/fetch.
> >> > > > Replacing topic-partition string with partition epoch is
> >> essentially a
> >> > > > performance improvement on top of this KIP. That can probably be
> >> done
> >> > > > separately.
> >> > > >
> >> > > > Thanks,
> >> > > >
> >> > > > Jiangjie (Becket) Qin
> >> > > >
> >> > > > On Mon, Jan 29, 2018 at 11:52 AM, Dong Lin <lindon...@gmail.com>
> >> > wrote:
> >> > > >
> >> > > > > Hey Colin,
> >> > > > >
> >> > > > > On Mon, Jan 29, 2018 at 11:23 AM, Colin McCabe <
> >> cmcc...@apache.org>
> >> > > > wrote:
> >> > > > >
> >> > > > > > > On Mon, Jan 29, 2018 at 10:35 AM, Dong Lin <
> >> lindon...@gmail.com>
> >> > > > > wrote:
> >> > > > > > >
> >> > > > > > > > Hey Colin,
> >> > > > > > > >
> >> > > > > > > > I understand that the KIP will adds overhead by
> introducing
> >> > > > > > per-partition
> >> > > > > > > > partitionEpoch. I am open to alternative solutions that
> does
> >> > not
> >> > > > > incur
> >> > > > > > > > additional overhead. But I don't see a better way now.
> >> > > > > > > >
> >> > > > > > > > IMO the overhead in the FetchResponse may not be that
> much.
> >> We
> >> > > > > probably
> >> > > > > > > > should discuss the percentage increase rather than the
> >> absolute
> >> > > > > number
> >> > > > > > > > increase. Currently after KIP-227, per-partition header
> has
> >> 23
> >> > > > bytes.
> >> > > > > > This
> >> > > > > > > > KIP adds another 4 bytes. Assume the records size is 10KB,
> >> the
> >> > > > > > percentage
> >> > > > > > > > increase is 4 / (23 + 10000) = 0.03%. It seems negligible,
> >> > right?
> >> > > > > >
> >> > > > > > Hi Dong,
> >> > > > > >
> >> > > > > > Thanks for the response. I agree that the FetchRequest /
> >> > > FetchResponse
> >> > > > > > overhead should be OK, now that we have incremental fetch
> >> requests
> >> > > and
> >> > > > > > responses. However, there are a lot of cases where the
> >> percentage
> >> > > > > increase
> >> > > > > > is much greater. For example, if a client is doing full
> >> > > > > MetadataRequests /
> >> > > > > > Responses, we have some math kind of like this per partition:
> >> > > > > >
> >> > > > > > > UpdateMetadataRequestPartitionState => topic partition
> >> > > > > controller_epoch
> >> > > > > > leader leader_epoch partition_epoch isr zk_version replicas
> >> > > > > > offline_replicas
> >> > > > > > > 14 bytes: topic => string (assuming about 10 byte topic
> >> names)
> >> > > > > > > 4 bytes: partition => int32
> >> > > > > > > 4 bytes: conroller_epoch => int32
> >> > > > > > > 4 bytes: leader => int32
> >> > > > > > > 4 bytes: leader_epoch => int32
> >> > > > > > > +4 EXTRA bytes: partition_epoch => int32 <-- NEW
> >> > > > > > > 2+4+4+4 bytes: isr => [int32] (assuming 3 in the ISR)
> >> > > > > > > 4 bytes: zk_version => int32
> >> > > > > > > 2+4+4+4 bytes: replicas => [int32] (assuming 3 replicas)
> >> > > > > > > 2 offline_replicas => [int32] (assuming no offline
> replicas)
> >> > > > > >
> >> > > > > > Assuming I added that up correctly, the per-partition overhead
> >> goes
> >> > > > from
> >> > > > > > 64 bytes per partition to 68, a 6.2% increase.
> >> > > > > >
> >> > > > > > We could do similar math for a lot of the other RPCs. And you
> >> will
> >> > > > have
> >> > > > > a
> >> > > > > > similar memory and garbage collection impact on the brokers
> >> since
> >> > you
> >> > > > > have
> >> > > > > > to store all this extra state as well.
> >> > > > > >
> >> > > > >
> >> > > > > That is correct. IMO the Metadata is only updated periodically
> >> and is
> >> > > > > probably not a big deal if we increase it by 6%. The
> FetchResponse
> >> > and
> >> > > > > ProduceRequest are probably the only requests that are bounded
> by
> >> the
> >> > > > > bandwidth throughput.
> >> > > > >
> >> > > > >
> >> > > > > >
> >> > > > > > > >
> >> > > > > > > > I agree that we can probably save more space by using
> >> partition
> >> > > ID
> >> > > > so
> >> > > > > > that
> >> > > > > > > > we no longer needs the string topic name. The similar idea
> >> has
> >> > > also
> >> > > > > > been
> >> > > > > > > > put in the Rejected Alternative section in KIP-227. While
> >> this
> >> > > idea
> >> > > > > is
> >> > > > > > > > promising, it seems orthogonal to the goal of this KIP.
> >> Given
> >> > > that
> >> > > > > > there is
> >> > > > > > > > already many work to do in this KIP, maybe we can do the
> >> > > partition
> >> > > > ID
> >> > > > > > in a
> >> > > > > > > > separate KIP?
> >> > > > > >
> >> > > > > > I guess my thinking is that the goal here is to replace an
> >> > identifier
> >> > > > > > which can be re-used (the tuple of topic name, partition ID)
> >> with
> >> > an
> >> > > > > > identifier that cannot be re-used (the tuple of topic name,
> >> > partition
> >> > > > ID,
> >> > > > > > partition epoch) in order to gain better semantics. As long
> as
> >> we
> >> > > are
> >> > > > > > replacing the identifier, why not replace it with an
> identifier
> >> > that
> >> > > > has
> >> > > > > > important performance advantages? The KIP freeze for the next
> >> > > release
> >> > > > > has
> >> > > > > > already passed, so there is time to do this.
> >> > > > > >
> >> > > > >
> >> > > > > In general it can be easier for discussion and implementation if
> >> we
> >> > can
> >> > > > > split a larger task into smaller and independent tasks. For
> >> example,
> >> > > > > KIP-112 and KIP-113 both deals with the JBOD support. KIP-31,
> >> KIP-32
> >> > > and
> >> > > > > KIP-33 are about timestamp support. The option on this can be
> >> subject
> >> > > > > though.
> >> > > > >
> >> > > > > IMO the change to switch from (topic, partition ID) to
> >> partitionEpch
> >> > in
> >> > > > all
> >> > > > > request/response requires us to going through all request one by
> >> one.
> >> > > It
> >> > > > > may not be hard but it can be time consuming and tedious. At
> high
> >> > level
> >> > > > the
> >> > > > > goal and the change for that will be orthogonal to the changes
> >> > required
> >> > > > in
> >> > > > > this KIP. That is the main reason I think we can split them into
> >> two
> >> > > > KIPs.
> >> > > > >
> >> > > > >
> >> > > > > > On Mon, Jan 29, 2018, at 10:54, Dong Lin wrote:
> >> > > > > > > I think it is possible to move to entirely use
> partitionEpoch
> >> > > instead
> >> > > > > of
> >> > > > > > > (topic, partition) to identify a partition. Client can
> obtain
> >> the
> >> > > > > > > partitionEpoch -> (topic, partition) mapping from
> >> > MetadataResponse.
> >> > > > We
> >> > > > > > > probably need to figure out a way to assign partitionEpoch
> to
> >> > > > existing
> >> > > > > > > partitions in the cluster. But this should be doable.
> >> > > > > > >
> >> > > > > > > This is a good idea. I think it will save us some space in
> the
> >> > > > > > > request/response. The actual space saving in percentage
> >> probably
> >> > > > > depends
> >> > > > > > on
> >> > > > > > > the amount of data and the number of partitions of the same
> >> > topic.
> >> > > I
> >> > > > > just
> >> > > > > > > think we can do it in a separate KIP.
> >> > > > > >
> >> > > > > > Hmm. How much extra work would be required? It seems like we
> >> are
> >> > > > > already
> >> > > > > > changing almost every RPC that involves topics and partitions,
> >> > > already
> >> > > > > > adding new per-partition state to ZooKeeper, already changing
> >> how
> >> > > > clients
> >> > > > > > interact with partitions. Is there some other big piece of
> work
> >> > we'd
> >> > > > > have
> >> > > > > > to do to move to partition IDs that we wouldn't need for
> >> partition
> >> > > > > epochs?
> >> > > > > > I guess we'd have to find a way to support regular
> >> expression-based
> >> > > > topic
> >> > > > > > subscriptions. If we split this into multiple KIPs, wouldn't
> we
> >> > end
> >> > > up
> >> > > > > > changing all that RPCs and ZK state a second time? Also, I'm
> >> > curious
> >> > > > if
> >> > > > > > anyone has done any proof of concept GC, memory, and network
> >> usage
> >> > > > > > measurements on switching topic names for topic IDs.
> >> > > > > >
> >> > > > >
> >> > > > >
> >> > > > > We will need to go over all requests/responses to check how to
> >> > replace
> >> > > > > (topic, partition ID) with partition epoch. It requires
> >> non-trivial
> >> > > work
> >> > > > > and could take time. As you mentioned, we may want to see how
> much
> >> > > saving
> >> > > > > we can get by switching from topic names to partition epoch.
> That
> >> > > itself
> >> > > > > requires time and experiment. It seems that the new idea does
> not
> >> > > > rollback
> >> > > > > any change proposed in this KIP. So I am not sure we can get
> much
> >> by
> >> > > > > putting them into the same KIP.
> >> > > > >
> >> > > > > Anyway, if more people are interested in seeing the new idea in
> >> the
> >> > > same
> >> > > > > KIP, I can try that.
> >> > > > >
> >> > > > >
> >> > > > >
> >> > > > > >
> >> > > > > > best,
> >> > > > > > Colin
> >> > > > > >
> >> > > > > > >
> >> > > > > > >
> >> > > > > > >
> >> > > > > > > >
> >> > > > > > > >
> >> > > > > > > >
> >> > > > > > > >
> >> > > > > > > >
> >> > > > > > > >
> >> > > > > > > >
> >> > > > > > > >
> >> > > > > > > >
> >> > > > > > > >
> >> > > > > > > > On Mon, Jan 29, 2018 at 10:18 AM, Colin McCabe <
> >> > > cmcc...@apache.org
> >> > > > >
> >> > > > > > wrote:
> >> > > > > > > >
> >> > > > > > > >> On Fri, Jan 26, 2018, at 12:17, Dong Lin wrote:
> >> > > > > > > >> > Hey Colin,
> >> > > > > > > >> >
> >> > > > > > > >> >
> >> > > > > > > >> > On Fri, Jan 26, 2018 at 10:16 AM, Colin McCabe <
> >> > > > > cmcc...@apache.org>
> >> > > > > > > >> wrote:
> >> > > > > > > >> >
> >> > > > > > > >> > > On Thu, Jan 25, 2018, at 16:47, Dong Lin wrote:
> >> > > > > > > >> > > > Hey Colin,
> >> > > > > > > >> > > >
> >> > > > > > > >> > > > Thanks for the comment.
> >> > > > > > > >> > > >
> >> > > > > > > >> > > > On Thu, Jan 25, 2018 at 4:15 PM, Colin McCabe <
> >> > > > > > cmcc...@apache.org>
> >> > > > > > > >> > > wrote:
> >> > > > > > > >> > > >
> >> > > > > > > >> > > > > On Wed, Jan 24, 2018, at 21:07, Dong Lin wrote:
> >> > > > > > > >> > > > > > Hey Colin,
> >> > > > > > > >> > > > > >
> >> > > > > > > >> > > > > > Thanks for reviewing the KIP.
> >> > > > > > > >> > > > > >
> >> > > > > > > >> > > > > > If I understand you right, you maybe suggesting
> >> that
> >> > > we
> >> > > > > can
> >> > > > > > use
> >> > > > > > > >> a
> >> > > > > > > >> > > global
> >> > > > > > > >> > > > > > metadataEpoch that is incremented every time
> >> > > controller
> >> > > > > > updates
> >> > > > > > > >> > > metadata.
> >> > > > > > > >> > > > > > The problem with this solution is that, if a
> >> topic
> >> > is
> >> > > > > > deleted
> >> > > > > > > >> and
> >> > > > > > > >> > > created
> >> > > > > > > >> > > > > > again, user will not know whether that the
> offset
> >> > > which
> >> > > > is
> >> > > > > > > >> stored
> >> > > > > > > >> > > before
> >> > > > > > > >> > > > > > the topic deletion is no longer valid. This
> >> > motivates
> >> > > > the
> >> > > > > > idea
> >> > > > > > > >> to
> >> > > > > > > >> > > include
> >> > > > > > > >> > > > > > per-partition partitionEpoch. Does this sound
> >> > > > reasonable?
> >> > > > > > > >> > > > >
> >> > > > > > > >> > > > > Hi Dong,
> >> > > > > > > >> > > > >
> >> > > > > > > >> > > > > Perhaps we can store the last valid offset of
> each
> >> > > deleted
> >> > > > > > topic
> >> > > > > > > >> in
> >> > > > > > > >> > > > > ZooKeeper. Then, when a topic with one of those
> >> names
> >> > > > gets
> >> > > > > > > >> > > re-created, we
> >> > > > > > > >> > > > > can start the topic at the previous end offset
> >> rather
> >> > > than
> >> > > > > at
> >> > > > > > 0.
> >> > > > > > > >> This
> >> > > > > > > >> > > > > preserves immutability. It is no more burdensome
> >> than
> >> > > > > having
> >> > > > > > to
> >> > > > > > > >> > > preserve a
> >> > > > > > > >> > > > > "last epoch" for the deleted partition somewhere,
> >> > right?
> >> > > > > > > >> > > > >
> >> > > > > > > >> > > >
> >> > > > > > > >> > > > My concern with this solution is that the number of
> >> > > > zookeeper
> >> > > > > > nodes
> >> > > > > > > >> get
> >> > > > > > > >> > > > more and more over time if some users keep deleting
> >> and
> >> > > > > creating
> >> > > > > > > >> topics.
> >> > > > > > > >> > > Do
> >> > > > > > > >> > > > you think this can be a problem?
> >> > > > > > > >> > >
> >> > > > > > > >> > > Hi Dong,
> >> > > > > > > >> > >
> >> > > > > > > >> > > We could expire the "partition tombstones" after an
> >> hour
> >> > or
> >> > > > so.
> >> > > > > > In
> >> > > > > > > >> > > practice this would solve the issue for clients that
> >> like
> >> > to
> >> > > > > > destroy
> >> > > > > > > >> and
> >> > > > > > > >> > > re-create topics all the time. In any case, doesn't
> >> the
> >> > > > current
> >> > > > > > > >> proposal
> >> > > > > > > >> > > add per-partition znodes as well that we have to
> track
> >> > even
> >> > > > > after
> >> > > > > > the
> >> > > > > > > >> > > partition is deleted? Or did I misunderstand that?
> >> > > > > > > >> > >
> >> > > > > > > >> >
> >> > > > > > > >> > Actually the current KIP does not add per-partition
> >> znodes.
> >> > > > Could
> >> > > > > > you
> >> > > > > > > >> > double check? I can fix the KIP wiki if there is
> anything
> >> > > > > > misleading.
> >> > > > > > > >>
> >> > > > > > > >> Hi Dong,
> >> > > > > > > >>
> >> > > > > > > >> I double-checked the KIP, and I can see that you are in
> >> fact
> >> > > > using a
> >> > > > > > > >> global counter for initializing partition epochs. So,
> you
> >> are
> >> > > > > > correct, it
> >> > > > > > > >> doesn't add per-partition znodes for partitions that no
> >> longer
> >> > > > > exist.
> >> > > > > > > >>
> >> > > > > > > >> >
> >> > > > > > > >> > If we expire the "partition tomstones" after an hour,
> and
> >> > the
> >> > > > > topic
> >> > > > > > is
> >> > > > > > > >> > re-created after more than an hour since the topic
> >> deletion,
> >> > > > then
> >> > > > > > we are
> >> > > > > > > >> > back to the situation where user can not tell whether
> the
> >> > > topic
> >> > > > > has
> >> > > > > > been
> >> > > > > > > >> > re-created or not, right?
> >> > > > > > > >>
> >> > > > > > > >> Yes, with an expiration period, it would not ensure
> >> > > immutability--
> >> > > > > you
> >> > > > > > > >> could effectively reuse partition names and they would
> look
> >> > the
> >> > > > > same.
> >> > > > > > > >>
> >> > > > > > > >> >
> >> > > > > > > >> >
> >> > > > > > > >> > >
> >> > > > > > > >> > > It's not really clear to me what should happen when a
> >> > topic
> >> > > is
> >> > > > > > > >> destroyed
> >> > > > > > > >> > > and re-created with new data. Should consumers
> >> continue
> >> > to
> >> > > be
> >> > > > > > able to
> >> > > > > > > >> > > consume? We don't know where they stopped consuming
> >> from
> >> > > the
> >> > > > > > previous
> >> > > > > > > >> > > incarnation of the topic, so messages may have been
> >> lost.
> >> > > > > > Certainly
> >> > > > > > > >> > > consuming data from offset X of the new incarnation
> of
> >> the
> >> > > > topic
> >> > > > > > may
> >> > > > > > > >> give
> >> > > > > > > >> > > something totally different from what you would have
> >> > gotten
> >> > > > from
> >> > > > > > > >> offset X
> >> > > > > > > >> > > of the previous incarnation of the topic.
> >> > > > > > > >> > >
> >> > > > > > > >> >
> >> > > > > > > >> > With the current KIP, if a consumer consumes a topic
> >> based
> >> > on
> >> > > > the
> >> > > > > > last
> >> > > > > > > >> > remembered (offset, partitionEpoch, leaderEpoch), and
> if
> >> the
> >> > > > topic
> >> > > > > > is
> >> > > > > > > >> > re-created, consume will throw
> >> > InvalidPartitionEpochException
> >> > > > > > because
> >> > > > > > > >> the
> >> > > > > > > >> > previous partitionEpoch will be different from the
> >> current
> >> > > > > > > >> partitionEpoch.
> >> > > > > > > >> > This is described in the Proposed Changes ->
> Consumption
> >> > after
> >> > > > > topic
> >> > > > > > > >> > deletion in the KIP. I can improve the KIP if there is
> >> > > anything
> >> > > > > not
> >> > > > > > > >> clear.
> >> > > > > > > >>
> >> > > > > > > >> Thanks for the clarification. It sounds like what you
> >> really
> >> > > want
> >> > > > > is
> >> > > > > > > >> immutability-- i.e., to never "really" reuse partition
> >> > > > identifiers.
> >> > > > > > And
> >> > > > > > > >> you do this by making the partition name no longer the
> >> "real"
> >> > > > > > identifier.
> >> > > > > > > >>
> >> > > > > > > >> My big concern about this KIP is that it seems like an
> >> > > > > > anti-scalability
> >> > > > > > > >> feature. Now we are adding 4 extra bytes for every
> >> partition
> >> > in
> >> > > > the
> >> > > > > > > >> FetchResponse and Request, for example. That could be 40
> >> kb
> >> > per
> >> > > > > > request,
> >> > > > > > > >> if the user has 10,000 partitions. And of course, the
> KIP
> >> > also
> >> > > > > makes
> >> > > > > > > >> massive changes to UpdateMetadataRequest,
> MetadataResponse,
> >> > > > > > > >> OffsetCommitRequest, OffsetFetchResponse,
> >> LeaderAndIsrRequest,
> >> > > > > > > >> ListOffsetResponse, etc. which will also increase their
> >> size
> >> > on
> >> > > > the
> >> > > > > > wire
> >> > > > > > > >> and in memory.
> >> > > > > > > >>
> >> > > > > > > >> One thing that we talked a lot about in the past is
> >> replacing
> >> > > > > > partition
> >> > > > > > > >> names with IDs. IDs have a lot of really nice features.
> >> They
> >> > > > take
> >> > > > > > up much
> >> > > > > > > >> less space in memory than strings (especially 2-byte Java
> >> > > > strings).
> >> > > > > > They
> >> > > > > > > >> can often be allocated on the stack rather than the heap
> >> > > > (important
> >> > > > > > when
> >> > > > > > > >> you are dealing with hundreds of thousands of them).
> They
> >> can
> >> > > be
> >> > > > > > > >> efficiently deserialized and serialized. If we use
> 64-bit
> >> > ones,
> >> > > > we
> >> > > > > > will
> >> > > > > > > >> never run out of IDs, which means that they can always be
> >> > unique
> >> > > > per
> >> > > > > > > >> partition.
> >> > > > > > > >>
> >> > > > > > > >> Given that the partition name is no longer the "real"
> >> > identifier
> >> > > > for
> >> > > > > > > >> partitions in the current KIP-232 proposal, why not just
> >> move
> >> > to
> >> > > > > using
> >> > > > > > > >> partition IDs entirely instead of strings? You have to
> >> change
> >> > > all
> >> > > > > the
> >> > > > > > > >> messages anyway. There isn't much point any more to
> >> carrying
> >> > > > around
> >> > > > > > the
> >> > > > > > > >> partition name in every RPC, since you really need (name,
> >> > epoch)
> >> > > > to
> >> > > > > > > >> identify the partition.
> >> > > > > > > >> Probably the metadata response and a few other messages
> >> would
> >> > > have
> >> > > > > to
> >> > > > > > > >> still carry the partition name, to allow clients to go
> from
> >> > name
> >> > > > to
> >> > > > > > id.
> >> > > > > > > >> But we could mostly forget about the strings. And then
> >> this
> >> > > would
> >> > > > > be
> >> > > > > > a
> >> > > > > > > >> scalability improvement rather than a scalability
> problem.
> >> > > > > > > >>
> >> > > > > > > >> >
> >> > > > > > > >> >
> >> > > > > > > >> > > By choosing to reuse the same (topic, partition,
> >> offset)
> >> > > > > 3-tuple,
> >> > > > > > we
> >> > > > > > > >> have
> >> > > > > > > >> >
> >> > > > > > > >> > chosen to give up immutability. That was a really bad
> >> > > decision.
> >> > > > > > And
> >> > > > > > > >> now
> >> > > > > > > >> > > we have to worry about time dependencies, stale
> cached
> >> > data,
> >> > > > and
> >> > > > > > all
> >> > > > > > > >> the
> >> > > > > > > >> > > rest. We can't completely fix this inside Kafka no
> >> matter
> >> > > > what
> >> > > > > > we do,
> >> > > > > > > >> > > because not all that cached data is inside Kafka
> >> itself.
> >> > > Some
> >> > > > > of
> >> > > > > > it
> >> > > > > > > >> may be
> >> > > > > > > >> > > in systems that Kafka has sent data to, such as other
> >> > > daemons,
> >> > > > > SQL
> >> > > > > > > >> > > databases, streams, and so forth.
> >> > > > > > > >> > >
> >> > > > > > > >> >
> >> > > > > > > >> > The current KIP will uniquely identify a message using
> >> > (topic,
> >> > > > > > > >> partition,
> >> > > > > > > >> > offset, partitionEpoch) 4-tuple. This addresses the
> >> message
> >> > > > > > immutability
> >> > > > > > > >> > issue that you mentioned. Is there any corner case
> where
> >> the
> >> > > > > message
> >> > > > > > > >> > immutability is still not preserved with the current
> KIP?
> >> > > > > > > >> >
> >> > > > > > > >> >
> >> > > > > > > >> > >
> >> > > > > > > >> > > I guess the idea here is that mirror maker should
> work
> >> as
> >> > > > > expected
> >> > > > > > > >> when
> >> > > > > > > >> > > users destroy a topic and re-create it with the same
> >> name.
> >> > > > > That's
> >> > > > > > > >> kind of
> >> > > > > > > >> > > tough, though, since in that scenario, mirror maker
> >> > probably
> >> > > > > > should
> >> > > > > > > >> destroy
> >> > > > > > > >> > > and re-create the topic on the other end, too, right?
> >> > > > > Otherwise,
> >> > > > > > > >> what you
> >> > > > > > > >> > > end up with on the other end could be half of one
> >> > > incarnation
> >> > > > of
> >> > > > > > the
> >> > > > > > > >> topic,
> >> > > > > > > >> > > and half of another.
> >> > > > > > > >> > >
> >> > > > > > > >> > > What mirror maker really needs is to be able to
> follow
> >> a
> >> > > > stream
> >> > > > > of
> >> > > > > > > >> events
> >> > > > > > > >> > > about the kafka cluster itself. We could have some
> >> master
> >> > > > topic
> >> > > > > > > >> which is
> >> > > > > > > >> > > always present and which contains data about all
> topic
> >> > > > > deletions,
> >> > > > > > > >> > > creations, etc. Then MM can simply follow this topic
> >> and
> >> > do
> >> > > > > what
> >> > > > > > is
> >> > > > > > > >> needed.
> >> > > > > > > >> > >
> >> > > > > > > >> > > >
> >> > > > > > > >> > > >
> >> > > > > > > >> > > > >
> >> > > > > > > >> > > > > >
> >> > > > > > > >> > > > > > Then the next question maybe, should we use a
> >> global
> >> > > > > > > >> metadataEpoch +
> >> > > > > > > >> > > > > > per-partition partitionEpoch, instead of using
> >> > > > > per-partition
> >> > > > > > > >> > > leaderEpoch
> >> > > > > > > >> > > > > +
> >> > > > > > > >> > > > > > per-partition leaderEpoch. The former solution
> >> using
> >> > > > > > > >> metadataEpoch
> >> > > > > > > >> > > would
> >> > > > > > > >> > > > > > not work due to the following scenario
> (provided
> >> by
> >> > > > Jun):
> >> > > > > > > >> > > > > >
> >> > > > > > > >> > > > > > "Consider the following scenario. In metadata
> v1,
> >> > the
> >> > > > > leader
> >> > > > > > > >> for a
> >> > > > > > > >> > > > > > partition is at broker 1. In metadata v2,
> leader
> >> is
> >> > at
> >> > > > > > broker
> >> > > > > > > >> 2. In
> >> > > > > > > >> > > > > > metadata v3, leader is at broker 1 again. The
> >> last
> >> > > > > committed
> >> > > > > > > >> offset
> >> > > > > > > >> > > in
> >> > > > > > > >> > > > > v1,
> >> > > > > > > >> > > > > > v2 and v3 are 10, 20 and 30, respectively. A
> >> > consumer
> >> > > is
> >> > > > > > > >> started and
> >> > > > > > > >> > > > > reads
> >> > > > > > > >> > > > > > metadata v1 and reads messages from offset 0 to
> >> 25
> >> > > from
> >> > > > > > broker
> >> > > > > > > >> 1. My
> >> > > > > > > >> > > > > > understanding is that in the current proposal,
> >> the
> >> > > > > metadata
> >> > > > > > > >> version
> >> > > > > > > >> > > > > > associated with offset 25 is v1. The consumer
> is
> >> > then
> >> > > > > > restarted
> >> > > > > > > >> and
> >> > > > > > > >> > > > > fetches
> >> > > > > > > >> > > > > > metadata v2. The consumer tries to read from
> >> broker
> >> > 2,
> >> > > > > > which is
> >> > > > > > > >> the
> >> > > > > > > >> > > old
> >> > > > > > > >> > > > > > leader with the last offset at 20. In this
> case,
> >> the
> >> > > > > > consumer
> >> > > > > > > >> will
> >> > > > > > > >> > > still
> >> > > > > > > >> > > > > > get OffsetOutOfRangeException incorrectly."
> >> > > > > > > >> > > > > >
> >> > > > > > > >> > > > > > Regarding your comment "For the second purpose,
> >> this
> >> > > is
> >> > > > > > "soft
> >> > > > > > > >> state"
> >> > > > > > > >> > > > > > anyway. If the client thinks X is the leader
> >> but Y
> >> > is
> >> > > > > > really
> >> > > > > > > >> the
> >> > > > > > > >> > > leader,
> >> > > > > > > >> > > > > > the client will talk to X, and X will point out
> >> its
> >> > > > > mistake
> >> > > > > > by
> >> > > > > > > >> > > sending
> >> > > > > > > >> > > > > back
> >> > > > > > > >> > > > > > a NOT_LEADER_FOR_PARTITION.", it is probably no
> >> > true.
> >> > > > The
> >> > > > > > > >> problem
> >> > > > > > > >> > > here is
> >> > > > > > > >> > > > > > that the old leader X may still think it is the
> >> > leader
> >> > > > of
> >> > > > > > the
> >> > > > > > > >> > > partition
> >> > > > > > > >> > > > > and
> >> > > > > > > >> > > > > > thus it will not send back
> >> NOT_LEADER_FOR_PARTITION.
> >> > > The
> >> > > > > > reason
> >> > > > > > > >> is
> >> > > > > > > >> > > > > provided
> >> > > > > > > >> > > > > > in KAFKA-6262. Can you check if that makes
> sense?
> >> > > > > > > >> > > > >
> >> > > > > > > >> > > > > This is solvable with a timeout, right? If the
> >> leader
> >> > > > can't
> >> > > > > > > >> > > communicate
> >> > > > > > > >> > > > > with the controller for a certain period of time,
> >> it
> >> > > > should
> >> > > > > > stop
> >> > > > > > > >> > > acting as
> >> > > > > > > >> > > > > the leader. We have to solve this problem,
> >> anyway, in
> >> > > > order
> >> > > > > > to
> >> > > > > > > >> fix
> >> > > > > > > >> > > all the
> >> > > > > > > >> > > > > corner cases.
> >> > > > > > > >> > > > >
> >> > > > > > > >> > > >
> >> > > > > > > >> > > > Not sure if I fully understand your proposal. The
> >> > proposal
> >> > > > > > seems to
> >> > > > > > > >> > > require
> >> > > > > > > >> > > > non-trivial changes to our existing leadership
> >> election
> >> > > > > > mechanism.
> >> > > > > > > >> Could
> >> > > > > > > >> > > > you provide more detail regarding how it works? For
> >> > > example,
> >> > > > > how
> >> > > > > > > >> should
> >> > > > > > > >> > > > user choose this timeout, how leader determines
> >> whether
> >> > it
> >> > > > can
> >> > > > > > still
> >> > > > > > > >> > > > communicate with controller, and how this triggers
> >> > > > controller
> >> > > > > to
> >> > > > > > > >> elect
> >> > > > > > > >> > > new
> >> > > > > > > >> > > > leader?
> >> > > > > > > >> > >
> >> > > > > > > >> > > Before I come up with any proposal, let me make sure
> I
> >> > > > > understand
> >> > > > > > the
> >> > > > > > > >> > > problem correctly. My big question was, what
> prevents
> >> > > > > split-brain
> >> > > > > > > >> here?
> >> > > > > > > >> > >
> >> > > > > > > >> > > Let's say I have a partition which is on nodes A, B,
> >> and
> >> > C,
> >> > > > with
> >> > > > > > > >> min-ISR
> >> > > > > > > >> > > 2. The controller is D. At some point, there is a
> >> > network
> >> > > > > > partition
> >> > > > > > > >> > > between A and B and the rest of the cluster. The
> >> > Controller
> >> > > > > > > >> re-assigns the
> >> > > > > > > >> > > partition to nodes C, D, and E. But A and B keep
> >> chugging
> >> > > > away,
> >> > > > > > even
> >> > > > > > > >> > > though they can no longer communicate with the
> >> controller.
> >> > > > > > > >> > >
> >> > > > > > > >> > > At some point, a client with stale metadata writes to
> >> the
> >> > > > > > partition.
> >> > > > > > > >> It
> >> > > > > > > >> > > still thinks the partition is on node A, B, and C, so
> >> > that's
> >> > > > > > where it
> >> > > > > > > >> sends
> >> > > > > > > >> > > the data. It's unable to talk to C, but A and B
> reply
> >> > back
> >> > > > that
> >> > > > > > all
> >> > > > > > > >> is
> >> > > > > > > >> > > well.
> >> > > > > > > >> > >
> >> > > > > > > >> > > Is this not a case where we could lose data due to
> >> split
> >> > > > brain?
> >> > > > > > Or is
> >> > > > > > > >> > > there a mechanism for preventing this that I missed?
> >> If
> >> > it
> >> > > > is,
> >> > > > > it
> >> > > > > > > >> seems
> >> > > > > > > >> > > like a pretty serious failure case that we should be
> >> > > handling
> >> > > > > > with our
> >> > > > > > > >> > > metadata rework. And I think epoch numbers and
> >> timeouts
> >> > > might
> >> > > > > be
> >> > > > > > > >> part of
> >> > > > > > > >> > > the solution.
> >> > > > > > > >> > >
> >> > > > > > > >> >
> >> > > > > > > >> > Right, split brain can happen if RF=4 and minIsr=2.
> >> > However, I
> >> > > > am
> >> > > > > > not
> >> > > > > > > >> sure
> >> > > > > > > >> > it is a pretty serious issue which we need to address
> >> today.
> >> > > > This
> >> > > > > > can be
> >> > > > > > > >> > prevented by configuring the Kafka topic so that
> minIsr >
> >> > > RF/2.
> >> > > > > > > >> Actually,
> >> > > > > > > >> > if user sets minIsr=2, is there anything reason that
> user
> >> > > wants
> >> > > > to
> >> > > > > > set
> >> > > > > > > >> RF=4
> >> > > > > > > >> > instead of 4?
> >> > > > > > > >> >
> >> > > > > > > >> > Introducing timeout in leader election mechanism is
> >> > > > non-trivial. I
> >> > > > > > > >> think we
> >> > > > > > > >> > probably want to do that only if there is good use-case
> >> that
> >> > > can
> >> > > > > not
> >> > > > > > > >> > otherwise be addressed with the current mechanism.
> >> > > > > > > >>
> >> > > > > > > >> I still would like to think about these corner cases
> more.
> >> > But
> >> > > > > > perhaps
> >> > > > > > > >> it's not directly related to this KIP.
> >> > > > > > > >>
> >> > > > > > > >> regards,
> >> > > > > > > >> Colin
> >> > > > > > > >>
> >> > > > > > > >>
> >> > > > > > > >> >
> >> > > > > > > >> >
> >> > > > > > > >> > > best,
> >> > > > > > > >> > > Colin
> >> > > > > > > >> > >
> >> > > > > > > >> > >
> >> > > > > > > >> > > >
> >> > > > > > > >> > > >
> >> > > > > > > >> > > > > best,
> >> > > > > > > >> > > > > Colin
> >> > > > > > > >> > > > >
> >> > > > > > > >> > > > > >
> >> > > > > > > >> > > > > > Regards,
> >> > > > > > > >> > > > > > Dong
> >> > > > > > > >> > > > > >
> >> > > > > > > >> > > > > >
> >> > > > > > > >> > > > > > On Wed, Jan 24, 2018 at 10:39 AM, Colin McCabe
> <
> >> > > > > > > >> cmcc...@apache.org>
> >> > > > > > > >> > > > > wrote:
> >> > > > > > > >> > > > > >
> >> > > > > > > >> > > > > > > Hi Dong,
> >> > > > > > > >> > > > > > >
> >> > > > > > > >> > > > > > > Thanks for proposing this KIP. I think a
> >> metadata
> >> > > > epoch
> >> > > > > > is a
> >> > > > > > > >> > > really
> >> > > > > > > >> > > > > good
> >> > > > > > > >> > > > > > > idea.
> >> > > > > > > >> > > > > > >
> >> > > > > > > >> > > > > > > I read through the DISCUSS thread, but I
> still
> >> > don't
> >> > > > > have
> >> > > > > > a
> >> > > > > > > >> clear
> >> > > > > > > >> > > > > picture
> >> > > > > > > >> > > > > > > of why the proposal uses a metadata epoch per
> >> > > > partition
> >> > > > > > rather
> >> > > > > > > >> > > than a
> >> > > > > > > >> > > > > > > global metadata epoch. A metadata epoch per
> >> > > partition
> >> > > > > is
> >> > > > > > > >> kind of
> >> > > > > > > >> > > > > > > unpleasant-- it's at least 4 extra bytes per
> >> > > partition
> >> > > > > > that we
> >> > > > > > > >> > > have to
> >> > > > > > > >> > > > > send
> >> > > > > > > >> > > > > > > over the wire in every full metadata request,
> >> > which
> >> > > > > could
> >> > > > > > > >> become
> >> > > > > > > >> > > extra
> >> > > > > > > >> > > > > > > kilobytes on the wire when the number of
> >> > partitions
> >> > > > > > becomes
> >> > > > > > > >> large.
> >> > > > > > > >> > > > > Plus,
> >> > > > > > > >> > > > > > > we have to update all the auxillary classes
> to
> >> > > include
> >> > > > > an
> >> > > > > > > >> epoch.
> >> > > > > > > >> > > > > > >
> >> > > > > > > >> > > > > > > We need to have a global metadata epoch
> anyway
> >> to
> >> > > > handle
> >> > > > > > > >> partition
> >> > > > > > > >> > > > > > > addition and deletion. For example, if I
> give
> >> you
> >> > > > > > > >> > > > > > > MetadataResponse{part1,epoch 1, part2, epoch
> 1}
> >> > and
> >> > > > > > {part1,
> >> > > > > > > >> > > epoch1},
> >> > > > > > > >> > > > > which
> >> > > > > > > >> > > > > > > MetadataResponse is newer? You have no way
> of
> >> > > > knowing.
> >> > > > > > It
> >> > > > > > > >> could
> >> > > > > > > >> > > be
> >> > > > > > > >> > > > > that
> >> > > > > > > >> > > > > > > part2 has just been created, and the response
> >> > with 2
> >> > > > > > > >> partitions is
> >> > > > > > > >> > > > > newer.
> >> > > > > > > >> > > > > > > Or it coudl be that part2 has just been
> >> deleted,
> >> > and
> >> > > > > > > >> therefore the
> >> > > > > > > >> > > > > response
> >> > > > > > > >> > > > > > > with 1 partition is newer. You must have a
> >> global
> >> > > > epoch
> >> > > > > > to
> >> > > > > > > >> > > > > disambiguate
> >> > > > > > > >> > > > > > > these two cases.
> >> > > > > > > >> > > > > > >
> >> > > > > > > >> > > > > > > Previously, I worked on the Ceph distributed
> >> > > > filesystem.
> >> > > > > > > >> Ceph had
> >> > > > > > > >> > > the
> >> > > > > > > >> > > > > > > concept of a map of the whole cluster,
> >> maintained
> >> > > by a
> >> > > > > few
> >> > > > > > > >> servers
> >> > > > > > > >> > > > > doing
> >> > > > > > > >> > > > > > > paxos. This map was versioned by a single
> >> 64-bit
> >> > > > epoch
> >> > > > > > number
> >> > > > > > > >> > > which
> >> > > > > > > >> > > > > > > increased on every change. It was propagated
> >> to
> >> > > > clients
> >> > > > > > > >> through
> >> > > > > > > >> > > > > gossip. I
> >> > > > > > > >> > > > > > > wonder if something similar could work here?
> >> > > > > > > >> > > > > > >
> >> > > > > > > >> > > > > > > It seems like the the Kafka MetadataResponse
> >> > serves
> >> > > > two
> >> > > > > > > >> somewhat
> >> > > > > > > >> > > > > unrelated
> >> > > > > > > >> > > > > > > purposes. Firstly, it lets clients know what
> >> > > > partitions
> >> > > > > > > >> exist in
> >> > > > > > > >> > > the
> >> > > > > > > >> > > > > > > system and where they live. Secondly, it
> lets
> >> > > clients
> >> > > > > > know
> >> > > > > > > >> which
> >> > > > > > > >> > > nodes
> >> > > > > > > >> > > > > > > within the partition are in-sync (in the ISR)
> >> and
> >> > > > which
> >> > > > > > node
> >> > > > > > > >> is the
> >> > > > > > > >> > > > > leader.
> >> > > > > > > >> > > > > > >
> >> > > > > > > >> > > > > > > The first purpose is what you really need a
> >> > metadata
> >> > > > > epoch
> >> > > > > > > >> for, I
> >> > > > > > > >> > > > > think.
> >> > > > > > > >> > > > > > > You want to know whether a partition exists
> or
> >> > not,
> >> > > or
> >> > > > > you
> >> > > > > > > >> want to
> >> > > > > > > >> > > know
> >> > > > > > > >> > > > > > > which nodes you should talk to in order to
> >> write
> >> > to
> >> > > a
> >> > > > > > given
> >> > > > > > > >> > > > > partition. A
> >> > > > > > > >> > > > > > > single metadata epoch for the whole response
> >> > should
> >> > > be
> >> > > > > > > >> adequate
> >> > > > > > > >> > > here.
> >> > > > > > > >> > > > > We
> >> > > > > > > >> > > > > > > should not change the partition assignment
> >> without
> >> > > > going
> >> > > > > > > >> through
> >> > > > > > > >> > > > > zookeeper
> >> > > > > > > >> > > > > > > (or a similar system), and this inherently
> >> > > serializes
> >> > > > > > updates
> >> > > > > > > >> into
> >> > > > > > > >> > > a
> >> > > > > > > >> > > > > > > numbered stream. Brokers should also stop
> >> > > responding
> >> > > > to
> >> > > > > > > >> requests
> >> > > > > > > >> > > when
> >> > > > > > > >> > > > > they
> >> > > > > > > >> > > > > > > are unable to contact ZK for a certain time
> >> > period.
> >> > > > > This
> >> > > > > > > >> prevents
> >> > > > > > > >> > > the
> >> > > > > > > >> > > > > case
> >> > > > > > > >> > > > > > > where a given partition has been moved off
> some
> >> > set
> >> > > of
> >> > > > > > nodes,
> >> > > > > > > >> but a
> >> > > > > > > >> > > > > client
> >> > > > > > > >> > > > > > > still ends up talking to those nodes and
> >> writing
> >> > > data
> >> > > > > > there.
> >> > > > > > > >> > > > > > >
> >> > > > > > > >> > > > > > > For the second purpose, this is "soft state"
> >> > anyway.
> >> > > > If
> >> > > > > > the
> >> > > > > > > >> client
> >> > > > > > > >> > > > > thinks
> >> > > > > > > >> > > > > > > X is the leader but Y is really the leader,
> the
> >> > > client
> >> > > > > > will
> >> > > > > > > >> talk
> >> > > > > > > >> > > to X,
> >> > > > > > > >> > > > > and
> >> > > > > > > >> > > > > > > X will point out its mistake by sending back
> a
> >> > > > > > > >> > > > > NOT_LEADER_FOR_PARTITION.
> >> > > > > > > >> > > > > > > Then the client can update its metadata again
> >> and
> >> > > find
> >> > > > > > the new
> >> > > > > > > >> > > leader,
> >> > > > > > > >> > > > > if
> >> > > > > > > >> > > > > > > there is one. There is no need for an epoch
> to
> >> > > handle
> >> > > > > > this.
> >> > > > > > > >> > > > > Similarly, I
> >> > > > > > > >> > > > > > > can't think of a reason why changing the
> >> in-sync
> >> > > > replica
> >> > > > > > set
> >> > > > > > > >> needs
> >> > > > > > > >> > > to
> >> > > > > > > >> > > > > bump
> >> > > > > > > >> > > > > > > the epoch.
> >> > > > > > > >> > > > > > >
> >> > > > > > > >> > > > > > > best,
> >> > > > > > > >> > > > > > > Colin
> >> > > > > > > >> > > > > > >
> >> > > > > > > >> > > > > > >
> >> > > > > > > >> > > > > > > On Wed, Jan 24, 2018, at 09:45, Dong Lin
> wrote:
> >> > > > > > > >> > > > > > > > Thanks much for reviewing the KIP!
> >> > > > > > > >> > > > > > > >
> >> > > > > > > >> > > > > > > > Dong
> >> > > > > > > >> > > > > > > >
> >> > > > > > > >> > > > > > > > On Wed, Jan 24, 2018 at 7:10 AM, Guozhang
> >> Wang <
> >> > > > > > > >> > > wangg...@gmail.com>
> >> > > > > > > >> > > > > > > wrote:
> >> > > > > > > >> > > > > > > >
> >> > > > > > > >> > > > > > > > > Yeah that makes sense, again I'm just
> >> making
> >> > > sure
> >> > > > we
> >> > > > > > > >> understand
> >> > > > > > > >> > > > > all the
> >> > > > > > > >> > > > > > > > > scenarios and what to expect.
> >> > > > > > > >> > > > > > > > >
> >> > > > > > > >> > > > > > > > > I agree that if, more generally speaking,
> >> say
> >> > > > users
> >> > > > > > have
> >> > > > > > > >> only
> >> > > > > > > >> > > > > consumed
> >> > > > > > > >> > > > > > > to
> >> > > > > > > >> > > > > > > > > offset 8, and then call seek(16) to
> "jump"
> >> to
> >> > a
> >> > > > > > further
> >> > > > > > > >> > > position,
> >> > > > > > > >> > > > > then
> >> > > > > > > >> > > > > > > she
> >> > > > > > > >> > > > > > > > > needs to be aware that OORE maybe thrown
> >> and
> >> > she
> >> > > > > > needs to
> >> > > > > > > >> > > handle
> >> > > > > > > >> > > > > it or
> >> > > > > > > >> > > > > > > rely
> >> > > > > > > >> > > > > > > > > on reset policy which should not surprise
> >> her.
> >> > > > > > > >> > > > > > > > >
> >> > > > > > > >> > > > > > > > >
> >> > > > > > > >> > > > > > > > > I'm +1 on the KIP.
> >> > > > > > > >> > > > > > > > >
> >> > > > > > > >> > > > > > > > > Guozhang
> >> > > > > > > >> > > > > > > > >
> >> > > > > > > >> > > > > > > > >
> >> > > > > > > >> > > > > > > > > On Wed, Jan 24, 2018 at 12:31 AM, Dong
> Lin
> >> <
> >> > > > > > > >> > > lindon...@gmail.com>
> >> > > > > > > >> > > > > > > wrote:
> >> > > > > > > >> > > > > > > > >
> >> > > > > > > >> > > > > > > > > > Yes, in general we can not prevent
> >> > > > > > > >> OffsetOutOfRangeException
> >> > > > > > > >> > > if
> >> > > > > > > >> > > > > user
> >> > > > > > > >> > > > > > > > > seeks
> >> > > > > > > >> > > > > > > > > > to a wrong offset. The main goal is to
> >> > prevent
> >> > > > > > > >> > > > > > > OffsetOutOfRangeException
> >> > > > > > > >> > > > > > > > > if
> >> > > > > > > >> > > > > > > > > > user has done things in the right way,
> >> e.g.
> >> > > user
> >> > > > > > should
> >> > > > > > > >> know
> >> > > > > > > >> > > that
> >> > > > > > > >> > > > > > > there
> >> > > > > > > >> > > > > > > > > is
> >> > > > > > > >> > > > > > > > > > message with this offset.
> >> > > > > > > >> > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > For example, if user calls seek(..)
> right
> >> > > after
> >> > > > > > > >> > > construction, the
> >> > > > > > > >> > > > > > > only
> >> > > > > > > >> > > > > > > > > > reason I can think of is that user
> stores
> >> > > offset
> >> > > > > > > >> externally.
> >> > > > > > > >> > > In
> >> > > > > > > >> > > > > this
> >> > > > > > > >> > > > > > > > > case,
> >> > > > > > > >> > > > > > > > > > user currently needs to use the offset
> >> which
> >> > > is
> >> > > > > > obtained
> >> > > > > > > >> > > using
> >> > > > > > > >> > > > > > > > > position(..)
> >> > > > > > > >> > > > > > > > > > from the last run. With this KIP, user
> >> needs
> >> > > to
> >> > > > > get
> >> > > > > > the
> >> > > > > > > >> > > offset
> >> > > > > > > >> > > > > and
> >> > > > > > > >> > > > > > > the
> >> > > > > > > >> > > > > > > > > > offsetEpoch using
> >> > positionAndOffsetEpoch(...)
> >> > > > and
> >> > > > > > stores
> >> > > > > > > >> > > these
> >> > > > > > > >> > > > > > > > > information
> >> > > > > > > >> > > > > > > > > > externally. The next time user starts
> >> > > consumer,
> >> > > > > > he/she
> >> > > > > > > >> needs
> >> > > > > > > >> > > to
> >> > > > > > > >> > > > > call
> >> > > > > > > >> > > > > > > > > > seek(..., offset, offsetEpoch) right
> >> after
> >> > > > > > construction.
> >> > > > > > > >> > > Then KIP
> >> > > > > > > >> > > > > > > should
> >> > > > > > > >> > > > > > > > > be
> >> > > > > > > >> > > > > > > > > > able to ensure that we don't throw
> >> > > > > > > >> OffsetOutOfRangeException
> >> > > > > > > >> > > if
> >> > > > > > > >> > > > > > > there is
> >> > > > > > > >> > > > > > > > > no
> >> > > > > > > >> > > > > > > > > > unclean leader election.
> >> > > > > > > >> > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > Does this sound OK?
> >> > > > > > > >> > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > Regards,
> >> > > > > > > >> > > > > > > > > > Dong
> >> > > > > > > >> > > > > > > > > >
> >> > > > > > > >> > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > On Tue, Jan 23, 2018 at 11:44 PM,
> >> Guozhang
> >> > > Wang
> >> > > > <
> >> > > > > > > >> > > > > wangg...@gmail.com>
> >> > > > > > > >> > > > > > > > > > wrote:
> >> > > > > > > >> > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > "If consumer wants to consume message
> >> with
> >> > > > > offset
> >> > > > > > 16,
> >> > > > > > > >> then
> >> > > > > > > >> > > > > consumer
> >> > > > > > > >> > > > > > > > > must
> >> > > > > > > >> > > > > > > > > > > have
> >> > > > > > > >> > > > > > > > > > > already fetched message with offset
> 15"
> >> > > > > > > >> > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > --> this may not be always true
> right?
> >> > What
> >> > > if
> >> > > > > > > >> consumer
> >> > > > > > > >> > > just
> >> > > > > > > >> > > > > call
> >> > > > > > > >> > > > > > > > > > seek(16)
> >> > > > > > > >> > > > > > > > > > > after construction and then poll
> >> without
> >> > > > > committed
> >> > > > > > > >> offset
> >> > > > > > > >> > > ever
> >> > > > > > > >> > > > > > > stored
> >> > > > > > > >> > > > > > > > > > > before? Admittedly it is rare but we
> do
> >> > not
> >> > > > > > > >> programmably
> >> > > > > > > >> > > > > disallow
> >> > > > > > > >> > > > > > > it.
> >> > > > > > > >> > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > Guozhang
> >> > > > > > > >> > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > On Tue, Jan 23, 2018 at 10:42 PM,
> Dong
> >> > Lin <
> >> > > > > > > >> > > > > lindon...@gmail.com>
> >> > > > > > > >> > > > > > > > > wrote:
> >> > > > > > > >> > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > > Hey Guozhang,
> >> > > > > > > >> > > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > > Thanks much for reviewing the KIP!
> >> > > > > > > >> > > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > > In the scenario you described,
> let's
> >> > > assume
> >> > > > > that
> >> > > > > > > >> broker
> >> > > > > > > >> > > A has
> >> > > > > > > >> > > > > > > > > messages
> >> > > > > > > >> > > > > > > > > > > with
> >> > > > > > > >> > > > > > > > > > > > offset up to 10, and broker B has
> >> > messages
> >> > > > > with
> >> > > > > > > >> offset
> >> > > > > > > >> > > up to
> >> > > > > > > >> > > > > 20.
> >> > > > > > > >> > > > > > > If
> >> > > > > > > >> > > > > > > > > > > > consumer wants to consume message
> >> with
> >> > > > offset
> >> > > > > > 9, it
> >> > > > > > > >> will
> >> > > > > > > >> > > not
> >> > > > > > > >> > > > > > > receive
> >> > > > > > > >> > > > > > > > > > > > OffsetOutOfRangeException
> >> > > > > > > >> > > > > > > > > > > > from broker A.
> >> > > > > > > >> > > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > > If consumer wants to consume
> message
> >> > with
> >> > > > > offset
> >> > > > > > > >> 16, then
> >> > > > > > > >> > > > > > > consumer
> >> > > > > > > >> > > > > > > > > must
> >> > > > > > > >> > > > > > > > > > > > have already fetched message with
> >> offset
> >> > > 15,
> >> > > > > > which
> >> > > > > > > >> can
> >> > > > > > > >> > > only
> >> > > > > > > >> > > > > come
> >> > > > > > > >> > > > > > > from
> >> > > > > > > >> > > > > > > > > > > > broker B. Because consumer will
> fetch
> >> > from
> >> > > > > > broker B
> >> > > > > > > >> only
> >> > > > > > > >> > > if
> >> > > > > > > >> > > > > > > > > leaderEpoch
> >> > > > > > > >> > > > > > > > > > > >=
> >> > > > > > > >> > > > > > > > > > > > 2, then the current consumer
> >> leaderEpoch
> >> > > can
> >> > > > > > not be
> >> > > > > > > >> 1
> >> > > > > > > >> > > since
> >> > > > > > > >> > > > > this
> >> > > > > > > >> > > > > > > KIP
> >> > > > > > > >> > > > > > > > > > > > prevents leaderEpoch rewind. Thus
> we
> >> > will
> >> > > > not
> >> > > > > > have
> >> > > > > > > >> > > > > > > > > > > > OffsetOutOfRangeException
> >> > > > > > > >> > > > > > > > > > > > in this case.
> >> > > > > > > >> > > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > > Does this address your question, or
> >> > maybe
> >> > > > > there
> >> > > > > > is
> >> > > > > > > >> more
> >> > > > > > > >> > > > > advanced
> >> > > > > > > >> > > > > > > > > > scenario
> >> > > > > > > >> > > > > > > > > > > > that the KIP does not handle?
> >> > > > > > > >> > > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > > Thanks,
> >> > > > > > > >> > > > > > > > > > > > Dong
> >> > > > > > > >> > > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > > On Tue, Jan 23, 2018 at 9:43 PM,
> >> > Guozhang
> >> > > > > Wang <
> >> > > > > > > >> > > > > > > wangg...@gmail.com>
> >> > > > > > > >> > > > > > > > > > > wrote:
> >> > > > > > > >> > > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > > > Thanks Dong, I made a pass over
> the
> >> > wiki
> >> > > > and
> >> > > > > > it
> >> > > > > > > >> lgtm.
> >> > > > > > > >> > > > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > > > Just a quick question: can we
> >> > completely
> >> > > > > > > >> eliminate the
> >> > > > > > > >> > > > > > > > > > > > > OffsetOutOfRangeException with
> this
> >> > > > > approach?
> >> > > > > > Say
> >> > > > > > > >> if
> >> > > > > > > >> > > there
> >> > > > > > > >> > > > > is
> >> > > > > > > >> > > > > > > > > > > consecutive
> >> > > > > > > >> > > > > > > > > > > > > leader changes such that the
> cached
> >> > > > > metadata's
> >> > > > > > > >> > > partition
> >> > > > > > > >> > > > > epoch
> >> > > > > > > >> > > > > > > is
> >> > > > > > > >> > > > > > > > > 1,
> >> > > > > > > >> > > > > > > > > > > and
> >> > > > > > > >> > > > > > > > > > > > > the metadata fetch response
> returns
> >> > > with
> >> > > > > > > >> partition
> >> > > > > > > >> > > epoch 2
> >> > > > > > > >> > > > > > > > > pointing
> >> > > > > > > >> > > > > > > > > > to
> >> > > > > > > >> > > > > > > > > > > > > leader broker A, while the actual
> >> > > > up-to-date
> >> > > > > > > >> metadata
> >> > > > > > > >> > > has
> >> > > > > > > >> > > > > > > partition
> >> > > > > > > >> > > > > > > > > > > > epoch 3
> >> > > > > > > >> > > > > > > > > > > > > whose leader is now broker B, the
> >> > > metadata
> >> > > > > > > >> refresh will
> >> > > > > > > >> > > > > still
> >> > > > > > > >> > > > > > > > > succeed
> >> > > > > > > >> > > > > > > > > > > and
> >> > > > > > > >> > > > > > > > > > > > > the follow-up fetch request may
> >> still
> >> > > see
> >> > > > > > OORE?
> >> > > > > > > >> > > > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > > > Guozhang
> >> > > > > > > >> > > > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > > > On Tue, Jan 23, 2018 at 3:47 PM,
> >> Dong
> >> > > Lin
> >> > > > <
> >> > > > > > > >> > > > > lindon...@gmail.com
> >> > > > > > > >> > > > > > > >
> >> > > > > > > >> > > > > > > > > > wrote:
> >> > > > > > > >> > > > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > > > > Hi all,
> >> > > > > > > >> > > > > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > > > > I would like to start the
> voting
> >> > > process
> >> > > > > for
> >> > > > > > > >> KIP-232:
> >> > > > > > > >> > > > > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > > > > https://cwiki.apache.org/
> >> > > > > > > >> > > confluence/display/KAFKA/KIP-
> >> > > > > > > >> > > > > > > > > > > > > > 232%3A+Detect+outdated+metadat
> >> > > > > > > >> a+using+leaderEpoch+
> >> > > > > > > >> > > > > > > > > > and+partitionEpoch
> >> > > > > > > >> > > > > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > > > > The KIP will help fix a
> >> concurrency
> >> > > > issue
> >> > > > > in
> >> > > > > > > >> Kafka
> >> > > > > > > >> > > which
> >> > > > > > > >> > > > > > > > > currently
> >> > > > > > > >> > > > > > > > > > > can
> >> > > > > > > >> > > > > > > > > > > > > > cause message loss or message
> >> > > > duplication
> >> > > > > in
> >> > > > > > > >> > > consumer.
> >> > > > > > > >> > > > > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > > > > Regards,
> >> > > > > > > >> > > > > > > > > > > > > > Dong
> >> > > > > > > >> > > > > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > > > --
> >> > > > > > > >> > > > > > > > > > > > > -- Guozhang
> >> > > > > > > >> > > > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > > > --
> >> > > > > > > >> > > > > > > > > > > -- Guozhang
> >> > > > > > > >> > > > > > > > > > >
> >> > > > > > > >> > > > > > > > > >
> >> > > > > > > >> > > > > > > > >
> >> > > > > > > >> > > > > > > > >
> >> > > > > > > >> > > > > > > > >
> >> > > > > > > >> > > > > > > > > --
> >> > > > > > > >> > > > > > > > > -- Guozhang
> >> > > > > > > >> > > > > > > > >
> >> > > > > > > >> > > > > > >
> >> > > > > > > >> > > > >
> >> > > > > > > >> > >
> >> > > > > > > >>
> >> > > > > > > >
> >> > > > > > > >
> >> > > > > >
> >> > > > >
> >> > > >
> >> > >
> >> >
> >>
> >
> >
>
