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