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
>>>>>>>>>>>>>>>>> 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
>>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>
>>>>>>>>
>>>>>>>
>>>>>>
>>>>>
>>>>
>>>>
>>>
>>
>
signature.asc
Description: OpenPGP digital signature
