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