Very good catch, Matthias. I updated the KIP to state that the new DSLOperation will return a new, mutated KStream.
Thank you, Shay On Wed, Jul 26, 2023 at 6:13 PM Matthias J. Sax <mj...@apache.org> wrote: > One last question. What should happen for the following case: > > KStream myStream = build.stream(...).map(...); > myStream.markAsPartiitoned().groupByKey().aggregate(...); > myStream.join(...) > > The question is about the "fan-out" pattern. `myStream`, which is marked > for partitioning, is fed into two downstream operations. Thus, it's > clear that the aggregation won't trigger a rebalance. However, the > fan-out happens before `markAsRepartiitoned` and thus I would assume > that the join would trigger a repartitioning? > > This question is important, because if we follow what I said above, > `markAsRepartiitoned` returns a new KStream object, but does mutate the > upstream KStream object, what is semantically two different things. It > also has an impact on how we need to implement the feature. The KIP > should explicitly explain this case. > > > -Matthias > > On 7/26/23 4:58 PM, Shay Lin wrote: > > Hi John, > > > > Thanks for your reply. I updated the KIP to reflect the changes we > > discussed in the thread today. > > #1 is duly noted, I learned from the examples Sophie sent earlier! =) > > > > In the new version, I also talked about why IQ and joins will not work > with > > the interface and talked about the mitigation. The proposal > > now specifically states we are solving the unneeded partition problem > when > > IQ or join does not coexist in the kafka streams. In the concerns > section, > > the proposal talks about having a reverse mapping would make this new > > interface compatible with IQ and join again but is subject to demand. > > > > Let me know what you think. Thanks! > > Shay > > > > > > > > On Wed, Jul 26, 2023 at 2:35 PM John Roesler <vvcep...@apache.org> > wrote: > > > >> Hello Shay, > >> > >> Thanks for the KIP! > >> > >> I just took a look in preparation to vote, and there are two small-ish > >> things that I'd like to fix first. Apologies if this stuff has already > come > >> up in the discussion thread; I only skimmed it. > >> > >> 1. The KIP only mentions the name of the method instead of providing a > >> code snippet showing exactly what the method signature will be in the > >> interface. Normally, KIPs do the latter because it removes all ambiguity > >> from the proposal. It also gives you an opportunity to write down the > >> Javadoc you would add to the method instead of just mentioning the > points > >> that you plan to document. > >> > >> 2. The KIP lists some concerns, but not what you will do to mitigate > them. > >> For example, the concern about IQ not behaving correctly. Will you > disable > >> the use of the implicit partitioner downstream of one of these > >> cancellations? Or provide a new interface to supply the "reverse > mapping" > >> you mentioned? Or include documentation in the Javadoc for how to deal > with > >> the situation? I think there are a range of options for each of those > >> concerns, and we should state up front what we plan to do. > >> > >> Thanks again! > >> -John > >> > >> On 2023/07/24 20:33:05 Sophie Blee-Goldman wrote: > >>> Thanks Shay! You and Matthias have convinced me, I'm happy with the > >> current > >>> proposal. I think once you make the minor > >>> updates to the KIP document this will be ready for voting again. > >>> > >>> Cheers, > >>> Sophie > >>> > >>> On Mon, Jul 24, 2023 at 8:26 AM Shay Lin <lqxs...@gmail.com> wrote: > >>> > >>>> Hi Sophie and Matthias, thanks for your comments and replies. > >>>> > >>>> 1. Scope of change: KStreams only or KStreams/KTable > >>>> I took some time to digest your points, looking through how KStreams > >>>> triggers repartitions today. I noticed that `repartitionRequired`is a > >> flag > >>>> in KStreamImpl etc and not in KTableImpl etc. When I look further, in > >> the > >>>> case of KTable, instead of passing in a boolean flag, a repartition > >> node ` > >>>> TableRepartitionMapNode` is directly created. I went back and > >> referenced > >>>> the two issue tickets KAFKA-10844 and KAFKA-4835, both requests were > >>>> focused on KStreams, i.e. not to change the partition why the input > >> streams > >>>> are already correctly keyed. Is it possible that in the case of > KTable, > >>>> users always intend to repartition (change key) when they call on > >>>> aggregate? -- (this was written before I saw Matthias's comment) > >>>> > >>>> Overall, based on the tickets, I see the benefit of doing a contained > >>>> change focusing on KStreams, i.e. repartitionRequired, which would > >> solve > >>>> the pain points nicely. If we ran into similar complaints/optimization > >>>> requests for KTable down the line, we can address them on top of > >> this(let > >>>> me know if we have these requests already, I might just be negligent). > >>>> > >>>> 2. API: markAsPartitioned() vs config > >>>> If we go with the KStreams only scope, markAsPartition() is more > >>>> adequate, i.e. maps nicely to repartitionRequired. There is a list of > >>>> NamedOperations that may or may not trigger repartition based on its > >>>> context(KStreams or KTable) which would make the implementation more > >>>> confusing. > >>>> > >>>> 3. KIP documentation: Thanks for providing the links to previous KIPs. > >> I > >>>> will be adding the three use cases and javadoc. I will also document > >> the > >>>> risks when it relates to IQ and Join. > >>>> > >>>> Best, > >>>> Shay > >>>> > >>>> On Fri, Jul 21, 2023 at 5:55 PM Matthias J. Sax <mj...@apache.org> > >> wrote: > >>>> > >>>>> I agree that it could easily be misused. There is a few Jira tickets > >> for > >>>>> cases when people want to "cancel" a repartition step. I would hope > >>>>> those tickets are linked to the KIP (if not, we should do this, and > >>>>> maybe even c&p those cases as motivation into the KIP itself)? > >>>>> > >>>>> It's always a tricky question to what extend we want to guide users, > >> and > >>>>> to what extend we need to give levers for advances case (and how to > >>>>> design those levers...) It's for sure a good idea to call out "use > >> with > >>>>> case" in the JavaDocs for the new method. > >>>>> > >>>>> > >>>>> -Matthias > >>>>> > >>>>> On 7/21/23 3:34 PM, Sophie Blee-Goldman wrote: > >>>>>> I guess I felt a bit uneasy about how this could be used/abused > >> while > >>>>>> reading the KIP, but if we truly believe this is an advanced > >> feature, > >>>> I'm > >>>>>> fine with the way things currently are. It doesn't feel like the > >> best > >>>>> API, > >>>>>> but it does seem to be the best *possible* API given the way things > >>>> are. > >>>>>> > >>>>>> W.r.t the KTable notes, that all makes sense to me. I just wanted > >> to > >>>> lay > >>>>>> out all the potential cases to make sure we had our bases covered. > >>>>>> > >>>>>> I still think an example or two would help, but the only thing I > >> will > >>>>>> actually wait on before feeling comfortable enough to vote on this > >>>> would > >>>>> be > >>>>>> a clear method signature (and maybe sample javadocs) in the "Public > >>>>>> Interfaces" section. > >>>>>> > >>>>>> Thanks again for the KIP Shay! Hope I haven't dragged it out too > >> much > >>>>>> > >>>>>> On Fri, Jul 21, 2023 at 3:19 PM Matthias J. Sax <mj...@apache.org> > >>>>> wrote: > >>>>>> > >>>>>>> Some thought about the API question. > >>>>>>> > >>>>>>> > >>>>>>>>> A. kstream.groupBy(...).aggregate(...) > >>>>>>> > >>>>>>> This can be re-writtten as > >>>>>>> > >>>>>>> kstream.selectKey(...) > >>>>>>> .markAsRepartitioned() > >>>>>>> .groupByKey() > >>>>>>> .aggregate() > >>>>>>> > >>>>>>> Given that `markAsRepartitoned` is an advanced feature, I think it > >>>> would > >>>>>>> be ok? > >>>>>>> > >>>>>>> > >>>>>>>>> B. ktable.groupBy(...).aggregate(...) > >>>>>>> > >>>>>>> For KTable aggregation, not sure how useful it would be? In the > >> end, > >>>> an > >>>>>>> table aggregation does only make sense if we pick something from > >> the > >>>>>>> value, ie, we indeed change the key? > >>>>>>> > >>>>>>> > >>>>>>>>> C. kstream.selectKey(...).join(ktable) > >>>>>>> > >>>>>>> We can just insert a `markAsRepartitioned()` after `selectKey` to > >>>> avoid > >>>>>>> repartitioning of the left input KStream. > >>>>>>> > >>>>>>> > >>>>>>>> KStream.selectKey(...).toTable().join(...) > >>>>>>> > >>>>>>> Not sure if I understand what you try to say with this example? > >> In the > >>>>>>> end, `selectKey(...).toTable()` would repartiton. If I know that > >> one > >>>> can > >>>>>>> upsert directly, one inserts a `markAsRepartitioned()` in between. > >>>>>>> > >>>>>>> > >>>>>>> In general, the use case seems to be that the key is not in the > >> right > >>>>>>> "format", or there is no key, but data was partitioned by a > >>>>>>> value-attribute upstream and we just want to extract this > >>>>>>> value-attribute into the key. Both seems to be KStream cases? > >>>>>>> > >>>>>>> > >>>>>>> -Matthias > >>>>>>> > >>>>>>> > >>>>>>> > >>>>>>> On 7/15/23 1:43 PM, Sophie Blee-Goldman wrote: > >>>>>>>> Hey Shay, while I don't have any specific concerns about the new > >>>> public > >>>>>>> API > >>>>>>>> in this KIP, I'd like to better understand how this feature will > >> work > >>>>>>>> before I vote. We should document the behavior of this new > >> operator > >>>>>>> clearly > >>>>>>>> in the KIP as well -- you don't necessarily need to write the > >>>> complete > >>>>>>>> javadocs up front, but it should be possible for a user to read > >> the > >>>> KIP > >>>>>>> and > >>>>>>>> then understand how this feature will work and how they would > >> need to > >>>>>>> apply > >>>>>>>> it. > >>>>>>>> > >>>>>>>> To that end, I recommend framing this proposal with a few > >> examples to > >>>>>>> help > >>>>>>>> clarify the semantics. When and where can you apply the > >>>>>>> markAsPartitioned() > >>>>>>>> operator? Some suggestions below. > >>>>>>>> > >>>>>>>> Specific notes: > >>>>>>>> > >>>>>>>> 1. The KIP opens with "Each key changing operation in Kafka > >> Streams > >>>>>>>> (selectKey, map, transform, etc.) now leads to automatic > >> repartition > >>>>>>> before > >>>>>>>> an aggregation." We should change "aggregation" to "stateful > >>>> operation" > >>>>>>> as > >>>>>>>> this is true for things like joins as well as aggregations > >>>>>>>> 2. The callout on IQ makes me a bit uncomfortable -- basically it > >>>> says > >>>>>>> this > >>>>>>>> should not be a concern "if we use markAsPartitioned correctly". > >> Does > >>>>>>> this > >>>>>>>> mean if we, the devs implementing this, write the feature > >> correctly? > >>>> Or > >>>>>>> is > >>>>>>>> it saying that this won't be a problem as long as "we", the > >> users of > >>>>> this > >>>>>>>> feature, use it correctly"? Just wondering if you've put any > >> thought > >>>>> into > >>>>>>>> how this would work yet (I personally have not) > >>>>>>>> 3. The KIP should lay out the proposed API exactly, even if > >> there's > >>>>> only > >>>>>>>> one new method. Check out this KIP > >>>>>>>> < > >>>>>>> > >>>>> > >>>> > >> > https://cwiki.apache.org/confluence/display/KAFKA/KIP-450%3A+Sliding+Window+Aggregations+in+the+DSL > >>>>>>>> > >>>>>>>> (or this KIP > >>>>>>>> < > >>>>>>> > >>>>> > >>>> > >> > https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=128651808 > >>>>>>>> ) > >>>>>>>> for a good reference on what the Public Interfaces section should > >>>>> include > >>>>>>>> 4. Regarding the proposed API itself, I wonder if KStream is > >> really > >>>> the > >>>>>>>> most appropriate interface for the new operator. A repartition > >> can be > >>>>>>>> triggered on just a KTable. Here's where some examples would > >> help. > >>>>>>> Perhaps > >>>>>>>> we could focus on these three cases: > >>>>>>>> > >>>>>>>> A. kstream.groupBy(...).aggregate(...) > >>>>>>>> B. ktable.groupBy(...).aggregate(...) > >>>>>>>> C. kstream.selectKey(...).join(ktable) > >>>>>>>> > >>>>>>>> I'm sure someone will correct me if I'm missing any additional > >> vital > >>>>>>>> examples, but at the very least, these are the three to consider: > >>>>> either > >>>>>>> a > >>>>>>>> KStream or KTable followed by a groupBy/aggregation, or a KStream > >>>> with > >>>>>>>> key-changing operator followed by a join. Note that you could > >> have > >>>>>>>> something like KStream.selectKey(...).toTable().join(...) as > >> well, > >>>> but > >>>>>>>> since there are no pure key-changing operators (like #selectKey) > >> on > >>>>>>>> KTables, only groupBy() which must always be followed by > >> aggregation, > >>>>>>> this > >>>>>>>> 4th case can be reduced to an example like C of a KStream with > >>>>>>> key-changing > >>>>>>>> operation and downstream join -- ie there's no way to do this > >> without > >>>>>>>> #toTable which is more like syntactic sugar for the purposes of > >> this > >>>>>>>> repartitioning discussion. > >>>>>>>> > >>>>>>>> I worry that making this a DSL operator on KStream is too > >> generic, > >>>> and > >>>>> we > >>>>>>>> would also need to add it to KTable for example B, despite > >> KTables > >>>> not > >>>>>>>> having any true pure key-changing operators outside of #groupBy. > >>>> Would > >>>>> we > >>>>>>>> throw an exception if you invoked #markAsPartitioned on a KTable > >> that > >>>>>>>> wasn't followed by a groupBy? If you have multiple key-changing > >>>>>>> operators, > >>>>>>>> would you need to add markAsPartitioned after each one? If not, > >> what > >>>>> are > >>>>>>>> the semantics of that? These are the main questions that got me > >>>>> thinking > >>>>>>>> here, and will definitely need to be clarified in the KIP if we > >> do go > >>>>>>> with > >>>>>>>> the current proposal. But I wanted to throw out another idea for > >> an > >>>>> API I > >>>>>>>> think would help with some of this awkwardness by having clearly > >>>>> defined > >>>>>>>> semantics: > >>>>>>>> > >>>>>>>> Fundamentally it seems to me that these issues are arising from > >> that > >>>>>>> "being > >>>>>>>> partitioned" is conceptually a property of other operations > >> applied > >>>> to > >>>>> a > >>>>>>>> KStream/KTable, rather than an operation itself. So rather than > >>>> making > >>>>>>> this > >>>>>>>> a DSL operator itself, what if we added it to the Grouped and > >> various > >>>>>>>> Joined configuration classes? It would allow us to more > >> carefully hit > >>>>>>> only > >>>>>>>> the relevant parts of the DSL, so there are no questions about > >>>>>>> whether/when > >>>>>>>> to throw errors when the operator is incorrectly applied -- there > >>>> would > >>>>>>> be > >>>>>>>> no way to apply it incorrectly. The main drawback I can think of > >> is > >>>>>>> simply > >>>>>>>> that this touches on a larger surface area of the API. I > >> personally > >>>>> don't > >>>>>>>> believe this is a good enough reason to make it a DSL operator > >> as one > >>>>>>> could > >>>>>>>> make that argument for nearly any kind of KStream or KTable > >> operator > >>>>>>>> configuration going forward, and would explode the > >> KStream/KTable API > >>>>>>>> surface area instead. Perhaps this was discussed during the > >> previous > >>>>>>>> iteration of this KIP, or I'm missing something here, so I just > >>>> wanted > >>>>> to > >>>>>>>> put this out there and see what people think > >>>>>>>> > >>>>>>>> Either way, thanks for picking up this KIP. It's been a long time > >>>>> coming > >>>>>>> :) > >>>>>>>> > >>>>>>>> -Sophie > >>>>>>>> > >>>>>>>> > >>>>>>>> > >>>>>>>> > >>>>>>>> > >>>>>>>> On Mon, Jul 10, 2023 at 2:05 PM Shay Lin <lqxs...@gmail.com> > >> wrote: > >>>>>>>> > >>>>>>>>> Hi all, > >>>>>>>>> > >>>>>>>>> It's been a few days so I went ahead with editing the KIP, the > >> main > >>>>>>> change > >>>>>>>>> is on the method name > >>>>>>>>> > >>>>>>>>> > >>>>>>> > >>>>> > >>>> > >> > https://cwiki.apache.org/confluence/display/KAFKA/KIP-759%3A+Unneeded+repartition+canceling > >>>>>>>>> . > >>>>>>>>> I will follow up with a VOTE separately. > >>>>>>>>> > >>>>>>>>> Best, > >>>>>>>>> Shay > >>>>>>>>> > >>>>>>>>> On Thu, Jun 29, 2023 at 4:52 PM Matthias J. Sax < > >> mj...@apache.org> > >>>>>>> wrote: > >>>>>>>>> > >>>>>>>>>> Shay, > >>>>>>>>>> > >>>>>>>>>> thanks for picking up this KIP. It's a pity that the discussion > >>>>> stalled > >>>>>>>>>> for such a long time. > >>>>>>>>>> > >>>>>>>>>> As expressed previously, I am happy with the name > >>>>> `markAsPartitioned()` > >>>>>>>>>> and also believe it's ok to just document the impact and leave > >> it > >>>> to > >>>>>>> the > >>>>>>>>>> user to do the right thing. > >>>>>>>>>> > >>>>>>>>>> If we really get a lot of users that ask about it, because > >> they did > >>>>> not > >>>>>>>>>> do the right thing, we could still add something (eg, a > >>>>> reverse-mapper > >>>>>>>>>> function) in a follow-up KIP. But we don't know if it's > >> necessary; > >>>>>>> thus, > >>>>>>>>>> making a small incremental step sounds like a good approach to > >> me. > >>>>>>>>>> > >>>>>>>>>> Let's see if others agree or not. > >>>>>>>>>> > >>>>>>>>>> > >>>>>>>>>> -Matthias > >>>>>>>>>> > >>>>>>>>>> On 6/28/23 5:29 PM, Shay Lin wrote: > >>>>>>>>>>> Hi all, > >>>>>>>>>>> > >>>>>>>>>>> Great discussion thread. May I take this KIP up? If it’s > >> alright > >>>> my > >>>>>>>>> plan > >>>>>>>>>> is > >>>>>>>>>>> to update the KIP with the operator `markAsPartitioned()`. > >>>>>>>>>>> > >>>>>>>>>>> As you have discussed and pointed out, there are implications > >> to > >>>>>>>>>> downstream > >>>>>>>>>>> joins or aggregation operations. Still, the operator is > >> intended > >>>> for > >>>>>>>>>>> advanced users so my two cents is it would be a valuable > >> addition > >>>>>>>>>>> nonetheless. We could add this as a caution/consideration as > >> part > >>>> of > >>>>>>>>> the > >>>>>>>>>>> java doc. > >>>>>>>>>>> > >>>>>>>>>>> Let me know, thanks. > >>>>>>>>>>> Shay > >>>>>>>>>>> > >>>>>>>>>> > >>>>>>>>> > >>>>>>>> > >>>>>>> > >>>>>> > >>>>> > >>>> > >>> > >> > > >
