Hi, Jan, Dong, John, Guozhang, Perhaps it will be useful to have a KIP meeting to discuss this together as a group. Would Apr. 9 (Monday) at 9:00am PDT work? If so, I will send out an invite to the mailing list.
Thanks, Jun On Wed, Apr 4, 2018 at 1:25 AM, Jan Filipiak <jan.filip...@trivago.com> wrote: > Want to quickly step in here again because it is going places again. > > The last part of the discussion is just a pain to read and completely > diverged from what I suggested without making the reasons clear to me. > > I don't know why this happens.... here are my comments anyway. > > @Guozhang: That Streams is working on automatic creating > copartition-usuable topics: great for streams, has literally nothing todo > with the KIP as we want to grow the > input topic. Everyone can reshuffle rel. easily but that is not what we > need todo, we need to grow the topic in question. After streams > automatically reshuffled the input topic > still has the same size and it didn't help a bit. I fail to see why this > is relevant. What am i missing here? > > @Dong > I am still on the position that the current proposal brings us into the > wrong direction. Especially introducing PartitionKeyRebalanceListener > From this point we can never move away to proper state full handling > without completely deprecating this creature from hell again. > Linear hashing is not the optimising step we have todo here. An interface > that when a topic is a topic its always the same even after it had > grown or shrunk is important. So from my POV I have major concerns that > this KIP is benefitial in its current state. > > What is it that makes everyone so addicted to the idea of linear hashing? > not attractive at all for me. > And with statefull consumers still a complete mess. Why not stick with the > Kappa architecture??? > > > > > > On 03.04.2018 17:38, Dong Lin wrote: > >> Hey John, >> >> Thanks much for your comments!! >> >> I have yet to go through the emails of John/Jun/Guozhang in detail. But >> let >> me present my idea for how to minimize the delay for state loading for >> stream use-case. >> >> For ease of understanding, let's assume that the initial partition number >> of input topics and change log topic are both 10. And initial number of >> stream processor is also 10. If we only increase initial partition number >> of input topics to 15 without changing number of stream processor, the >> current KIP already guarantees in-order delivery and no state needs to be >> moved between consumers for stream use-case. Next, let's say we want to >> increase the number of processor to expand the processing capacity for >> stream use-case. This requires us to move state between processors which >> will take time. Our goal is to minimize the impact (i.e. delay) for >> processing while we increase the number of processors. >> >> Note that stream processor generally includes both consumer and producer. >> In addition to consume from the input topic, consumer may also need to >> consume from change log topic on startup for recovery. And producer may >> produce state to the change log topic. >> >> > The solution will include the following steps: >> >> 1) Increase partition number of the input topic from 10 to 15. Since the >> messages with the same key will still go to the same consume before and >> after the partition expansion, this step can be done without having to >> move >> state between processors. >> >> 2) Increase partition number of the change log topic from 10 to 15. Note >> that this step can also be done without impacting existing workflow. After >> we increase partition number of the change log topic, key space may split >> and some key will be produced to the newly-added partition. But the same >> key will still go to the same processor (i.e. consumer) before and after >> the partition. Thus this step can also be done without having to move >> state >> between processors. >> >> 3) Now, let's add 5 new consumers whose groupId is different from the >> existing processor's groupId. Thus these new consumers will not impact >> existing workflow. Each of these new consumers should consume two >> partitions from the earliest offset, where these two partitions are the >> same partitions that will be consumed if the consumers have the same >> groupId as the existing processor's groupId. For example, the first of the >> five consumers will consume partition 0 and partition 10. The purpose of >> these consumers is to rebuild the state (e.g. RocksDB) for the processors >> in advance. Also note that, by design of the current KIP, each consume >> will >> consume the existing partition of the change log topic up to the offset >> before the partition expansion. Then they will only need to consume the >> state of the new partition of the change log topic. >> >> 4) After consumers have caught up in step 3), we should stop these >> consumers and add 5 new processors to the stream processing job. These 5 >> new processors should run in the same location as the previous 5 consumers >> to re-use the state (e.g. RocksDB). And these processors' consumers should >> consume partitions of the change log topic from the committed offset the >> previous 5 consumers so that no state is missed. >> >> One important trick to note here is that, the mapping from partition to >> consumer should also use linear hashing. And we need to remember the >> initial number of processors in the job, 10 in this example, and use this >> number in the linear hashing algorithm. This is pretty much the same as >> how >> we use linear hashing to map key to partition. In this case, we get an >> identity map from partition -> processor, for both input topic and the >> change log topic. For example, processor 12 will consume partition 12 of >> the input topic and produce state to the partition 12 of the change log >> topic. >> >> There are a few important properties of this solution to note: >> >> - We can increase the number of partitions for input topic and the change >> log topic in any order asynchronously. >> - The expansion of the processors in a given job in step 4) only requires >> the step 3) for the same job. It does not require coordination across >> different jobs for step 3) and 4). Thus different jobs can independently >> expand there capacity without waiting for each other. >> - The logic for 1) and 2) is already supported in the current KIP. The >> logic for 3) and 4) appears to be independent of the core Kafka logic and >> can be implemented separately outside core Kafka. Thus the current KIP is >> probably sufficient after we agree on the efficiency and the correctness >> of >> the solution. We can have a separate KIP for Kafka Stream to support 3) >> and >> 4). >> >> >> Cheers, >> Dong >> >> >> On Mon, Apr 2, 2018 at 3:25 PM, Guozhang Wang <wangg...@gmail.com> wrote: >> >> Hey guys, just sharing my two cents here (I promise it will be shorter >>> than >>> John's article :). >>> >>> 0. Just to quickly recap, the main discussion point now is how to support >>> "key partitioning preservation" (John's #4 in topic characteristics >>> above) >>> beyond the "single-key ordering preservation" that KIP-253 was originally >>> proposed to maintain (John's #6 above). >>> >>> 1. From the streams project, we are actively working on improving the >>> elastic scalability of the library. One of the key features is to >>> decouple >>> the input topics from the parallelism model of Streams: i.e. not >>> enforcing >>> the topic to be partitioned by the key, not enforcing joining topics to >>> be >>> co-partitioned, not relying the number of parallel tasks on the input >>> topic >>> partitions. This can be achieved by re-shuffling on the input topics to >>> make sure key-partitioning / co-partitioning on the internal topics. Note >>> the re-shuffling task is purely stateless and hence does not require "key >>> partitioning preservation". Operational-wise it is similar to the >>> "creating >>> a new topic with new number of partitions, pipe the data to the new topic >>> and cut over consumers from old topics" idea, just that users can >>> optionally let Streams to handle such rather than doing it manually >>> themselves. There are a few more details on that regard but I will skip >>> since they are not directly related to this discussion. >>> >>> 2. Assuming that 1) above is done, then the only topics involved in the >>> scaling events are all input topics. For these topics the only producers >>> / >>> consumers of these topics are controlled by Streams clients themselves, >>> and >>> hence achieving "key partitioning preservation" is simpler than >>> non-Streams >>> scenarios: consumers know the partitioning scheme that producers are >>> using, >>> so that for their stateful operations it is doable to split the local >>> state >>> stores accordingly or execute backfilling on its own. Of course, if we >>> decide to do server-side backfilling, it can still help Streams to >>> directly >>> rely on that functionality. >>> >>> 3. As John mentioned, another way inside Streams is to do >>> over-partitioning >>> on all internal topics; then with 1) Streams would not rely on KIP-253 at >>> all. But personally I'd like to avoid it if possible to reduce Kafka side >>> footprint: say we overpartition each input topic up to 1k, with a >>> reasonable sized stateful topology it can still contribute to tens of >>> thousands of topics to the topic partition capacity of a single cluster. >>> >>> 4. Summing up 1/2/3, I think we should focus more on non-Streams users >>> writing their stateful computations with local states, and think whether >>> / >>> how we could enable "key partitioning preservation" for them easily, than >>> to think heavily for Streams library. People may have different opinion >>> on >>> how common of a usage pattern it is (I think Jun might be suggesting that >>> for DIY apps people may more likely use remote states so that it is not a >>> problem for them). My opinion is that for non-Streams users such usage >>> pattern could still be large (think: if you are piping data from Kafka to >>> an external data storage which has single-writer requirements for each >>> single shard, even though it is not a stateful computational application >>> it >>> may still require "key partitioning preservation"), so I prefer to have >>> backfilling in our KIP than only exposing the API for expansion and >>> requires consumers to have pre-knowledge of the producer's partitioning >>> scheme. >>> >>> >>> >>> Guozhang >>> >>> >>> >>> On Thu, Mar 29, 2018 at 2:33 PM, John Roesler <j...@confluent.io> wrote: >>> >>> Hey Dong, >>>> >>>> Congrats on becoming a committer!!! >>>> >>>> Since I just sent a novel-length email, I'll try and keep this one brief >>>> >>> ;) >>> >>>> Regarding producer coordination, I'll grant that in that case, producers >>>> may coordinate among themselves to produce into the same topic or to >>>> produce co-partitioned topics. Nothing in KStreams or the Kafka >>>> ecosystem >>>> in general requires such coordination for correctness or in fact for any >>>> optional features, though, so I would not say that we require producer >>>> coordination of partition logic. If producers currently coordinate, it's >>>> completely optional and their own choice. >>>> >>>> Regarding the portability of partition algorithms, my observation is >>>> that >>>> systems requiring independent implementations of the same algorithm with >>>> 100% correctness are a large source of risk and also a burden on those >>>> >>> who >>> >>>> have to maintain them. If people could flawlessly implement algorithms >>>> in >>>> actual software, the world would be a wonderful place indeed! For a >>>> >>> system >>> >>>> as important and widespread as Kafka, I would recommend restricting >>>> limiting such requirements as aggressively as possible. >>>> >>>> I'd agree that we can always revisit decisions like allowing arbitrary >>>> partition functions, but of course, we shouldn't do that in a vacuum. >>>> >>> That >>> >>>> feels like the kind of thing we'd need to proactively seek guidance from >>>> the users list about. I do think that the general approach of saying >>>> that >>>> "if you use a custom partitioner, you cannot do partition expansion" is >>>> very reasonable (but I don't think we need to go that far with the >>>> >>> current >>> >>>> proposal). It's similar to my statement in my email to Jun that in >>>> principle KStreams doesn't *need* backfill, we only need it if we want >>>> to >>>> employ partition expansion. >>>> >>>> I reckon that the main motivation for backfill is to support KStreams >>>> use >>>> cases and also any other use cases involving stateful consumers. >>>> >>>> Thanks for your response, and congrats again! >>>> -John >>>> >>>> >>>> On Wed, Mar 28, 2018 at 1:34 AM, Dong Lin <lindon...@gmail.com> wrote: >>>> >>>> Hey John, >>>>> >>>>> Great! Thanks for all the comment. It seems that we agree that the >>>>> >>>> current >>>> >>>>> KIP is in good shape for core Kafka. IMO, what we have been discussing >>>>> >>>> in >>> >>>> the recent email exchanges is mostly about the second step, i.e. how to >>>>> address problem for the stream use-case (or stateful processing in >>>>> general). >>>>> >>>>> I will comment inline. >>>>> >>>>> >>>>> >>>>> >>>>> On Tue, Mar 27, 2018 at 4:38 PM, John Roesler <j...@confluent.io> >>>>> >>>> wrote: >>> >>>> Thanks for the response, Dong. >>>>>> >>>>>> Here are my answers to your questions: >>>>>> >>>>>> - "Asking producers and consumers, or even two different producers, >>>>>> >>>>> to >>> >>>> share code like the partition function is a pretty huge ask. What >>>>>>> >>>>>> if >>> >>>> they >>>>> >>>>>> are using different languages?". It seems that today we already >>>>>>> >>>>>> require >>>> >>>>> different producer's to use the same hash function -- otherwise >>>>>>> >>>>>> messages >>>>> >>>>>> with the same key will go to different partitions of the same topic >>>>>>> >>>>>> which >>>>> >>>>>> may cause problem for downstream consumption. So not sure if it >>>>>>> >>>>>> adds >>> >>>> any >>>>> >>>>>> more constraint by assuming consumers know the hash function of >>>>>>> >>>>>> producer. >>>>> >>>>>> Could you explain more why user would want to use a cusmtom >>>>>>> >>>>>> partition >>> >>>> function? Maybe we can check if this is something that can be >>>>>>> >>>>>> supported >>>> >>>>> in >>>>>> >>>>>>> the default Kafka hash function. Also, can you explain more why it >>>>>>> >>>>>> is >>> >>>> difficuilt to implement the same hash function in different >>>>>>> >>>>>> languages? >>>> >>>>> >>>>>> Sorry, I meant two different producers as in producers to two >>>>>> >>>>> different >>> >>>> topics. This was in response to the suggestion that we already >>>>>> >>>>> require >>> >>>> coordination among producers to different topics in order to achieve >>>>>> co-partitioning. I was saying that we do not (and should not). >>>>>> >>>>> >>>>> It is probably common for producers of different team to produce >>>>> >>>> message >>> >>>> to >>>> >>>>> the same topic. In order to ensure that messages with the same key go >>>>> >>>> to >>> >>>> same partition, we need producers of different team to share the same >>>>> partition algorithm, which by definition requires coordination among >>>>> producers of different teams in an organization. Even for producers of >>>>> different topics, it may be common to require producers to use the same >>>>> partition algorithm in order to join two topics for stream processing. >>>>> >>>> Does >>>> >>>>> this make it reasonable to say we already require coordination across >>>>> producers? >>>>> >>>>> >>>>> By design, consumers are currently ignorant of the partitioning >>>>>> >>>>> scheme. >>> >>>> It >>>>> >>>>>> suffices to trust that the producer has partitioned the topic by key, >>>>>> >>>>> if >>>> >>>>> they claim to have done so. If you don't trust that, or even if you >>>>>> >>>>> just >>>> >>>>> need some other partitioning scheme, then you must re-partition it >>>>>> yourself. Nothing we're discussing can or should change that. The >>>>>> >>>>> value >>> >>>> of >>>>> >>>>>> backfill is that it preserves the ability for consumers to avoid >>>>>> re-partitioning before consuming, in the case where they don't need >>>>>> >>>>> to >>> >>>> today. >>>>>> >>>>> >>>>> Regarding shared "hash functions", note that it's a bit inaccurate to >>>>>> >>>>> talk >>>>> >>>>>> about the "hash function" of the producer. Properly speaking, the >>>>>> >>>>> producer >>>>> >>>>>> has only a "partition function". We do not know that it is a hash. >>>>>> >>>>> The >>> >>>> producer can use any method at their disposal to assign a partition >>>>>> >>>>> to >>> >>>> a >>>> >>>>> record. The partition function obviously may we written in any >>>>>> >>>>> programming >>>>> >>>>>> language, so in general it's not something that can be shared around >>>>>> without a formal spec or the ability to execute arbitrary executables >>>>>> >>>>> in >>>> >>>>> arbitrary runtime environments. >>>>>> >>>>>> Yeah it is probably better to say partition algorithm. I guess it >>>>> >>>> should >>> >>>> not be difficult to implement same partition algorithms in different >>>>> languages, right? Yes we would need a formal specification of the >>>>> >>>> default >>> >>>> partition algorithm in the producer. I think that can be documented as >>>>> >>>> part >>>> >>>>> of the producer interface. >>>>> >>>>> >>>>> Why would a producer want a custom partition function? I don't >>>>>> >>>>> know... >>> >>>> why >>>>> >>>>>> did we design the interface so that our users can provide one? In >>>>>> >>>>> general, >>>>> >>>>>> such systems provide custom partitioners because some data sets may >>>>>> >>>>> be >>> >>>> unbalanced under the default or because they can provide some >>>>>> >>>>> interesting >>>> >>>>> functionality built on top of the partitioning scheme, etc. Having >>>>>> >>>>> provided >>>>> >>>>>> this ability, I don't know why we would remove it. >>>>>> >>>>>> Yeah it is reasonable to assume that there was reason to support >>>>> custom >>>>> partition function in producer. On the other hand it may also be >>>>> >>>> reasonable >>>> >>>>> to revisit this interface and discuss whether we actually need to >>>>> >>>> support >>> >>>> custom partition function. If we don't have a good reason, we can >>>>> >>>> choose >>> >>>> not to support custom partition function in this KIP in a backward >>>>> compatible manner, i.e. user can still use custom partition function >>>>> >>>> but >>> >>>> they would not get the benefit of in-order delivery when there is >>>>> >>>> partition >>>> >>>>> expansion. What do you think? >>>>> >>>>> >>>>> - Besides the assumption that consumer needs to share the hash >>>>>> >>>>> function >>> >>>> of >>>>> >>>>>> producer, is there other organization overhead of the proposal in >>>>>>> >>>>>> the >>> >>>> current KIP? >>>>>>> >>>>>>> It wasn't clear to me that KIP-253 currently required the producer >>>>>> >>>>> and >>> >>>> consumer to share the partition function, or in fact that it had a >>>>>> >>>>> hard >>> >>>> requirement to abandon the general partition function and use a >>>>>> >>>>> linear >>> >>>> hash >>>>> >>>>>> function instead. >>>>>> >>>>> >>>>> In my reading, there is a requirement to track the metadata about >>>>>> >>>>> what >>> >>>> partitions split into what other partitions during an expansion >>>>>> >>>>> operation. >>>>> >>>>>> If the partition function is linear, this is easy. If not, you can >>>>>> >>>>> always >>>> >>>>> just record that all old partitions split into all new partitions. >>>>>> >>>>> This >>> >>>> has >>>>> >>>>>> the effect of forcing all consumers to wait until the old epoch is >>>>>> completely consumed before starting on the new epoch. But this may >>>>>> >>>>> be a >>> >>>> reasonable tradeoff, and it doesn't otherwise alter your design. >>>>>> >>>>>> You only mention the consumer needing to know that the partition >>>>>> >>>>> function >>>> >>>>> is linear, not what the actual function is, so I don't think your >>>>>> >>>>> design >>>> >>>>> actually calls for sharing the function. Plus, really all the >>>>>> >>>>> consumer >>> >>>> needs is the metadata about what old-epoch partitions to wait for >>>>>> >>>>> before >>>> >>>>> consuming a new-epoch partition. This information is directly >>>>>> >>>>> captured >>> >>>> in >>>> >>>>> metadata, so I don't think it actually even cares whether the >>>>>> >>>>> partition >>> >>>> function is linear or not. >>>>>> >>>>>> You are right that the current KIP does not mention it. My comment >>>>> >>>> related >>>> >>>>> to the partition function coordination was related to support the >>>>> stream-use case which we have been discussing so far. >>>>> >>>>> >>>>> So, no, I really think KIP-253 is in good shape. I was really more >>>>>> >>>>> talking >>>>> >>>>>> about the part of this thread that's outside of KIP-253's scope, >>>>>> >>>>> namely, >>>> >>>>> creating the possibility of backfilling partitions after expansion. >>>>>> >>>>>> Great! Can you also confirm that the main motivation for backfilling >>>>> partitions after expansion is to support the stream use-case? >>>>> >>>>> >>>>> - Currently producer can forget about the message that has been >>>>>> >>>>>>> acknowledged by the broker. Thus the producer probably does not >>>>>>> >>>>>> know >>> >>>> most >>>>> >>>>>> of the exiting messages in topic, including those messages produced >>>>>>> >>>>>> by >>>> >>>>> other producers. We can have the owner of the producer to >>>>>>> >>>>>> split+backfill. >>>>> >>>>>> In my opion it will be a new program that wraps around the existing >>>>>>> producer and consumer classes. >>>>>>> >>>>>>> This sounds fine by me! >>>>>> >>>>>> Really, I was just emphasizing that the part of the organization that >>>>>> produces a topic shouldn't have to export their partition function to >>>>>> >>>>> the >>>> >>>>> part(s) of the organization (or other organizations) that consume the >>>>>> topic. Whether the backfill operation goes into the Producer >>>>>> >>>>> interface >>> >>>> is >>>> >>>>> secondary, I think. >>>>>> >>>>>> - Regarding point 5. The argument is in favor of the split+backfill >>>>>> >>>>> but >>> >>>> for >>>>> >>>>>> changelog topic. And it intends to address the problem for stream >>>>>>> >>>>>> use-case >>>>>> >>>>>>> in general. In this KIP we will provide interface (i.e. >>>>>>> PartitionKeyRebalanceListener in the KIP) to be used by sream >>>>>>> >>>>>> use-case >>>> >>>>> and >>>>>> >>>>>>> the goal is that user can flush/re-consume the state as part of the >>>>>>> interface implementation regardless of whether there is change log >>>>>>> >>>>>> topic. >>>>> >>>>>> Maybe you are suggesting that the main reason to do split+backfill >>>>>>> >>>>>> of >>> >>>> input >>>>>> >>>>>>> topic is to support log compacted topics? You mentioned in Point 1 >>>>>>> >>>>>> that >>>> >>>>> log >>>>>> >>>>>>> compacted topics is out of the scope of this KIP. Maybe I could >>>>>>> >>>>>> understand >>>>>> >>>>>>> your position better. Regarding Jan's proposal to split partitions >>>>>>> >>>>>> with >>>> >>>>> backfill, do you think this should replace the proposal in the >>>>>>> >>>>>> existing >>>> >>>>> KIP, or do you think this is something that we should do in >>>>>>> >>>>>> addition >>> >>>> to >>>> >>>>> the >>>>>> >>>>>>> existing KIP? >>>>>>> >>>>>>> I think that interface is a good/necessary component of KIP-253. >>>>>> >>>>>> I personally (FWIW) feel that KIP-253 is appropriately scoped, but I >>>>>> >>>>> do >>> >>>> think its utility will be limited unless there is a later KIP >>>>>> >>>>> offering >>> >>>> backfill. But, maybe unlike Jan, I think it makes sense to try and >>>>>> >>>>> tackle >>>> >>>>> the ordering problem independently of backfill, so I'm in support of >>>>>> >>>>> the >>>> >>>>> current KIP. >>>>>> >>>>>> - Regarding point 6. I guess we can agree that it is better not to >>>>>> >>>>> have >>> >>>> the >>>>> >>>>>> performance overhread of copying the input data. Before we discuss >>>>>>> >>>>>> more >>>> >>>>> on >>>>>> >>>>>>> whether the performance overhead is acceptable or not, I am trying >>>>>>> >>>>>> to >>> >>>> figure out what is the benefit of introducing this overhread. You >>>>>>> >>>>>> mentioned >>>>>> >>>>>>> that the benefit is the loose organizational coupling. By >>>>>>> >>>>>> "organizational >>>>> >>>>>> coupling", are you referring to the requirement that consumer needs >>>>>>> >>>>>> to >>>> >>>>> know >>>>>> >>>>>>> the hash function of producer? If so, maybe we can discuss the >>>>>>> >>>>>> use-case >>>> >>>>> of >>>>>> >>>>>>> custom partiton function and see whether we can find a way to >>>>>>> >>>>>> support >>> >>>> such >>>>>> >>>>>>> use-case without having to copy the input data. >>>>>>> >>>>>>> I'm not too sure about what an "input" is in this sense, since we are >>>>>> >>>>> just >>>>> >>>>>> talking about topics. Actually the point I was making there is that >>>>>> >>>>> AKAICT >>>>> >>>>>> the performance overhead of a backfill is less than any other option, >>>>>> assuming you split partitions rarely. >>>>>> >>>>>> By "input" I was referring to source Kafka topic of a stream >>>>> processing >>>>> job. >>>>> >>>>> >>>>> Separately, yes, "organizational coupling" increases if producers and >>>>>> consumers have to share code, such as the partition function. This >>>>>> >>>>> would >>>> >>>>> not be the case if producers could only pick from a menu of a few >>>>>> well-known partition functions, but I think this is a poor tradeoff. >>>>>> >>>>>> Maybe we can revisit the custom partition function and see whether we >>>>> actually need it? Otherwise, I am concerned that every user will pay >>>>> >>>> the >>> >>>> overhead of data movement to support something that was not really >>>>> >>>> needed >>> >>>> for most users. >>>>> >>>>> >>>>> To me, this is two strong arguments in favor of backfill being less >>>>>> expensive than no backfill, but again, I think that particular debate >>>>>> >>>>> comes >>>>> >>>>>> after KIP-253, so I don't want to create the impression of opposition >>>>>> >>>>> to >>>> >>>>> your proposal. >>>>>> >>>>>> >>>>>> Finally, to respond to a new email I just noticed: >>>>>> >>>>>> BTW, here is my understanding of the scope of this KIP. We want to >>>>>>> >>>>>> allow >>>>> >>>>>> consumers to always consume messages with the same key from the >>>>>>> >>>>>> same >>> >>>> producer in the order they are produced. And we need to provide a >>>>>>> >>>>>> way >>> >>>> for >>>>> >>>>>> stream use-case to be able to flush/load state when messages with >>>>>>> >>>>>> the >>> >>>> same >>>>>> >>>>>>> key are migrated between consumers. In addition to ensuring that >>>>>>> >>>>>> this >>> >>>> goal >>>>>> >>>>>>> is correctly supported, we should do our best to keep the >>>>>>> >>>>>> performance >>> >>>> and >>>>> >>>>>> organization overhead of this KIP as low as possible. >>>>>>> >>>>>>> I think we're on the same page there! In fact, I would generalize a >>>>>> >>>>> little >>>>> >>>>>> more and say that the mechanism you've designed provides *all >>>>>> >>>>> consumers* >>>> >>>>> the ability "to flush/load state when messages with the same key are >>>>>> migrated between consumers", not just Streams. >>>>>> >>>>>> Thanks for all the comment! >>>>> >>>>> >>>>> Thanks for the discussion, >>>>>> -John >>>>>> >>>>>> >>>>>> >>>>>> On Tue, Mar 27, 2018 at 3:14 PM, Dong Lin <lindon...@gmail.com> >>>>>> >>>>> wrote: >>> >>>> Hey John, >>>>>>> >>>>>>> Thanks much for the detailed comments. Here are my thoughts: >>>>>>> >>>>>>> - The need to delete messages from log compacted topics is mainly >>>>>>> >>>>>> for >>> >>>> performance (e.g. storage space) optimization than for correctness >>>>>>> >>>>>> for >>>> >>>>> this >>>>>> >>>>>>> KIP. I agree that we probably don't need to focus on this in our >>>>>>> >>>>>> discussion >>>>>> >>>>>>> since it is mostly for performance optimization. >>>>>>> >>>>>>> - "Asking producers and consumers, or even two different producers, >>>>>>> >>>>>> to >>>> >>>>> share code like the partition function is a pretty huge ask. What >>>>>>> >>>>>> if >>> >>>> they >>>>> >>>>>> are using different languages?". It seems that today we already >>>>>>> >>>>>> require >>>> >>>>> different producer's to use the same hash function -- otherwise >>>>>>> >>>>>> messages >>>>> >>>>>> with the same key will go to different partitions of the same topic >>>>>>> >>>>>> which >>>>> >>>>>> may cause problem for downstream consumption. So not sure if it >>>>>>> >>>>>> adds >>> >>>> any >>>>> >>>>>> more constraint by assuming consumers know the hash function of >>>>>>> >>>>>> producer. >>>>> >>>>>> Could you explain more why user would want to use a cusmtom >>>>>>> >>>>>> partition >>> >>>> function? Maybe we can check if this is something that can be >>>>>>> >>>>>> supported >>>> >>>>> in >>>>>> >>>>>>> the default Kafka hash function. Also, can you explain more why it >>>>>>> >>>>>> is >>> >>>> difficuilt to implement the same hash function in different >>>>>>> >>>>>> languages? >>>> >>>>> - Besides the assumption that consumer needs to share the hash >>>>>>> >>>>>> function >>>> >>>>> of >>>>>> >>>>>>> producer, is there other organization overhead of the proposal in >>>>>>> >>>>>> the >>> >>>> current KIP? >>>>>>> >>>>>>> - Currently producer can forget about the message that has been >>>>>>> acknowledged by the broker. Thus the producer probably does not >>>>>>> >>>>>> know >>> >>>> most >>>>> >>>>>> of the exiting messages in topic, including those messages produced >>>>>>> >>>>>> by >>>> >>>>> other producers. We can have the owner of the producer to >>>>>>> >>>>>> split+backfill. >>>>> >>>>>> In my opion it will be a new program that wraps around the existing >>>>>>> producer and consumer classes. >>>>>>> >>>>>>> - Regarding point 5. The argument is in favor of the split+backfill >>>>>>> >>>>>> but >>>> >>>>> for >>>>>> >>>>>>> changelog topic. And it intends to address the problem for stream >>>>>>> >>>>>> use-case >>>>>> >>>>>>> in general. In this KIP we will provide interface (i.e. >>>>>>> PartitionKeyRebalanceListener in the KIP) to be used by sream >>>>>>> >>>>>> use-case >>>> >>>>> and >>>>>> >>>>>>> the goal is that user can flush/re-consume the state as part of the >>>>>>> interface implementation regardless of whether there is change log >>>>>>> >>>>>> topic. >>>>> >>>>>> Maybe you are suggesting that the main reason to do split+backfill >>>>>>> >>>>>> of >>> >>>> input >>>>>> >>>>>>> topic is to support log compacted topics? You mentioned in Point 1 >>>>>>> >>>>>> that >>>> >>>>> log >>>>>> >>>>>>> compacted topics is out of the scope of this KIP. Maybe I could >>>>>>> >>>>>> understand >>>>>> >>>>>>> your position better. Regarding Jan's proposal to split partitions >>>>>>> >>>>>> with >>>> >>>>> backfill, do you think this should replace the proposal in the >>>>>>> >>>>>> existing >>>> >>>>> KIP, or do you think this is something that we should do in >>>>>>> >>>>>> addition >>> >>>> to >>>> >>>>> the >>>>>> >>>>>>> existing KIP? >>>>>>> >>>>>>> - Regarding point 6. I guess we can agree that it is better not to >>>>>>> >>>>>> have >>>> >>>>> the >>>>>> >>>>>>> performance overhread of copying the input data. Before we discuss >>>>>>> >>>>>> more >>>> >>>>> on >>>>>> >>>>>>> whether the performance overhead is acceptable or not, I am trying >>>>>>> >>>>>> to >>> >>>> figure out what is the benefit of introducing this overhread. You >>>>>>> >>>>>> mentioned >>>>>> >>>>>>> that the benefit is the loose organizational coupling. By >>>>>>> >>>>>> "organizational >>>>> >>>>>> coupling", are you referring to the requirement that consumer needs >>>>>>> >>>>>> to >>>> >>>>> know >>>>>> >>>>>>> the hash function of producer? If so, maybe we can discuss the >>>>>>> >>>>>> use-case >>>> >>>>> of >>>>>> >>>>>>> custom partiton function and see whether we can find a way to >>>>>>> >>>>>> support >>> >>>> such >>>>>> >>>>>>> use-case without having to copy the input data. >>>>>>> >>>>>>> Thanks, >>>>>>> Dong >>>>>>> >>>>>>> >>>>>>> On Tue, Mar 27, 2018 at 11:34 AM, John Roesler <j...@confluent.io> >>>>>>> >>>>>> wrote: >>>>>> >>>>>>> Hey Dong and Jun, >>>>>>>> >>>>>>>> Thanks for the thoughtful responses. If you don't mind, I'll mix >>>>>>>> >>>>>>> my >>> >>>> replies >>>>>>> >>>>>>>> together to try for a coherent response. I'm not too familiar >>>>>>>> >>>>>>> with >>> >>>> mailing-list etiquette, though. >>>>>>>> >>>>>>>> I'm going to keep numbering my points because it makes it easy >>>>>>>> >>>>>>> for >>> >>>> you >>>>> >>>>>> all >>>>>>> >>>>>>>> to respond. >>>>>>>> >>>>>>>> Point 1: >>>>>>>> As I read it, KIP-253 is *just* about properly fencing the >>>>>>>> >>>>>>> producers >>>> >>>>> and >>>>>> >>>>>>> consumers so that you preserve the correct ordering of records >>>>>>>> >>>>>>> during >>>> >>>>> partition expansion. This is clearly necessary regardless of >>>>>>>> >>>>>>> anything >>>> >>>>> else >>>>>>> >>>>>>>> we discuss. I think this whole discussion about backfill, >>>>>>>> >>>>>>> consumers, >>>> >>>>> streams, etc., is beyond the scope of KIP-253. But it would be >>>>>>>> >>>>>>> cumbersome >>>>>> >>>>>>> to start a new thread at this point. >>>>>>>> >>>>>>>> I had missed KIP-253's Proposed Change #9 among all the >>>>>>>> >>>>>>> details... >>> >>>> I >>>> >>>>> think >>>>>>> >>>>>>>> this is a nice addition to the proposal. One thought is that it's >>>>>>>> >>>>>>> actually >>>>>>> >>>>>>>> irrelevant whether the hash function is linear. This is simply an >>>>>>>> >>>>>>> algorithm >>>>>>> >>>>>>>> for moving a key from one partition to another, so the type of >>>>>>>> >>>>>>> hash >>> >>>> function need not be a precondition. In fact, it also doesn't >>>>>>>> >>>>>>> matter >>>> >>>>> whether the topic is compacted or not, the algorithm works >>>>>>>> >>>>>>> regardless. >>>>> >>>>>> I think this is a good algorithm to keep in mind, as it might >>>>>>>> >>>>>>> solve a >>>> >>>>> variety of problems, but it does have a downside: that the >>>>>>>> >>>>>>> producer >>> >>>> won't >>>>>> >>>>>>> know whether or not K1 was actually in P1, it just knows that K1 >>>>>>>> >>>>>>> was >>>> >>>>> in >>>>> >>>>>> P1's keyspace before the new epoch. Therefore, it will have to >>>>>>>> pessimistically send (K1,null) to P1 just in case. But the next >>>>>>>> >>>>>>> time >>>> >>>>> K1 >>>>> >>>>>> comes along, the producer *also* won't remember that it already >>>>>>>> >>>>>>> retracted >>>>>> >>>>>>> K1 from P1, so it will have to send (K1,null) *again*. By >>>>>>>> >>>>>>> extension, >>>> >>>>> every >>>>>>> >>>>>>>> time the producer sends to P2, it will also have to send a >>>>>>>> >>>>>>> tombstone >>>> >>>>> to >>>>> >>>>>> P1, >>>>>>> >>>>>>>> which is a pretty big burden. To make the situation worse, if >>>>>>>> >>>>>>> there >>> >>>> is >>>>> >>>>>> a >>>>>> >>>>>>> second split, say P2 becomes P2 and P3, then any key Kx belonging >>>>>>>> >>>>>>> to >>>> >>>>> P3 >>>>> >>>>>> will also have to be retracted from P2 *and* P1, since the >>>>>>>> >>>>>>> producer >>> >>>> can't >>>>>> >>>>>>> know whether Kx had been last written to P2 or P1. Over a long >>>>>>>> >>>>>>> period >>>> >>>>> of >>>>>> >>>>>>> time, this clearly becomes a issue, as the producer must send an >>>>>>>> >>>>>>> arbitrary >>>>>>> >>>>>>>> number of retractions along with every update. >>>>>>>> >>>>>>>> In contrast, the proposed backfill operation has an end, and >>>>>>>> >>>>>>> after >>> >>>> it >>>> >>>>> ends, >>>>>>> >>>>>>>> everyone can afford to forget that there ever was a different >>>>>>>> >>>>>>> partition >>>>> >>>>>> layout. >>>>>>>> >>>>>>>> Really, though, figuring out how to split compacted topics is >>>>>>>> >>>>>>> beyond >>>> >>>>> the >>>>>> >>>>>>> scope of KIP-253, so I'm not sure #9 really even needs to be in >>>>>>>> >>>>>>> this >>>> >>>>> KIP... >>>>>>> >>>>>>>> We do need in-order delivery during partition expansion. It would >>>>>>>> >>>>>>> be >>>> >>>>> fine >>>>>> >>>>>>> by me to say that you *cannot* expand partitions of a >>>>>>>> >>>>>>> log-compacted >>> >>>> topic >>>>>> >>>>>>> and call it a day. I think it would be better to tackle that in >>>>>>>> >>>>>>> another >>>>> >>>>>> KIP. >>>>>>>> >>>>>>>> >>>>>>>> Point 2: >>>>>>>> Regarding whether the consumer re-shuffles its inputs, this is >>>>>>>> >>>>>>> always >>>> >>>>> on >>>>>> >>>>>>> the table; any consumer who wants to re-shuffle its input is free >>>>>>>> >>>>>>> to >>>> >>>>> do >>>>> >>>>>> so. >>>>>>> >>>>>>>> But this is currently not required. It's just that the current >>>>>>>> >>>>>>> high-level >>>>>> >>>>>>> story with Kafka encourages the use of partitions as a unit of >>>>>>>> >>>>>>> concurrency. >>>>>>> >>>>>>>> As long as consumers are single-threaded, they can happily >>>>>>>> >>>>>>> consume >>> >>>> a >>>> >>>>> single >>>>>>> >>>>>>>> partition without concurrency control of any kind. This is a key >>>>>>>> >>>>>>> aspect >>>>> >>>>>> to >>>>>>> >>>>>>>> this system that lets folks design high-throughput systems on top >>>>>>>> >>>>>>> of >>>> >>>>> it >>>>> >>>>>> surprisingly easily. If all consumers were instead >>>>>>>> >>>>>>> encouraged/required >>>>> >>>>>> to >>>>>> >>>>>>> implement a repartition of their own, then the consumer becomes >>>>>>>> significantly more complex, requiring either the consumer to >>>>>>>> >>>>>>> first >>> >>>> produce >>>>>>> >>>>>>>> to its own intermediate repartition topic or to ensure that >>>>>>>> >>>>>>> consumer >>>> >>>>> threads have a reliable, high-bandwith channel of communication >>>>>>>> >>>>>>> with >>>> >>>>> every >>>>>>> >>>>>>>> other consumer thread. >>>>>>>> >>>>>>>> Either of those tradeoffs may be reasonable for a particular user >>>>>>>> >>>>>>> of >>>> >>>>> Kafka, >>>>>>> >>>>>>>> but I don't know if we're in a position to say that they are >>>>>>>> >>>>>>> reasonable >>>>> >>>>>> for >>>>>>> >>>>>>>> *every* user of Kafka. >>>>>>>> >>>>>>>> >>>>>>>> Point 3: >>>>>>>> Regarding Jun's point about this use case, "(3) stateful and >>>>>>>> >>>>>>> maintaining >>>>>> >>>>>>> the >>>>>>>> states in a local store", I agree that they may use a framework >>>>>>>> >>>>>>> *like* >>>>> >>>>>> Kafka Streams, but that is not the same as using Kafka Streams. >>>>>>>> >>>>>>> This >>>> >>>>> is >>>>> >>>>>> why >>>>>>> >>>>>>>> I think it's better to solve it in Core: because it is then >>>>>>>> >>>>>>> solved >>> >>>> for >>>>> >>>>>> KStreams and also for everything else that facilitates local >>>>>>>> >>>>>>> state >>> >>>> maintenance. To me, Streams is a member of the category of >>>>>>>> >>>>>>> "stream >>> >>>> processing frameworks", which is itself a subcategory of "things >>>>>>>> >>>>>>> requiring >>>>>>> >>>>>>>> local state maintenence". I'm not sure if it makes sense to >>>>>>> >>>>>>>
