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