Ivan, did you see my last reply? What do you think about my proposal to mix both approaches and try to get best-of-both worlds?
-Matthias On 6/11/19 3:56 PM, Matthias J. Sax wrote: > Thanks for the input John! > >> under your suggestion, it seems that the name is required > > If you want to get the `KStream` as part of the `Map` back using a > `Function`, yes. If you follow the "embedded chaining" pattern using a > `Consumer`, no. > > Allowing for a default name via `split()` can of course be done. > Similarly, using `Named` instead of `String` is possible. > > I wanted to sketch out a high level proposal to merge both patterns > only. Your suggestions to align the new API with the existing API make > totally sense. > > > > One follow up question: Would `Named` be optional or required in > `split()` and `branch()`? It's unclear from your example. > > If both are mandatory, what do we gain by it? The returned `Map` only > contains the corresponding branches, so why should we prefix all of > them? If only `Named` is mandatory in `branch()`, but optional in > `split()`, the same question raises? > > Requiring `Named` in `split()` seems only to make sense, if `Named` is > optional in `branch()` and we generate `-X` suffix using a counter for > different branch name. However, this might lead to the problem of > changing names if branches are added/removed. Also, how would the names > be generated if `Consumer` is mixed in (ie, not all branches are > returned in the `Map`). > > If `Named` is optional for both, it could happen that a user misses to > specify a name for a branch what would lead to runtime issues. > > > Hence, I am actually in favor to not allow a default name but keep > `split()` without parameter and make `Named` in `branch()` required if a > `Function` is used. This makes it explicit to the user that specifying a > name is required if a `Function` is used. > > > > About > >> KBranchedStream#branch(BranchConfig) > > I don't think that the branching predicate is a configuration and hence > would not include it in a configuration object. > >> withChain(...); > > Similar, `withChain()` (that would only take a `Consumer`?) does not > seem to be a configuration. We can also not prevent a user to call > `withName()` in combination of `withChain()` what does not make sense > IMHO. We could of course throw an RTE but not have a compile time check > seems less appealing. Also, it could happen that neither `withChain()` > not `withName()` is called and the branch is missing in the returned > `Map` what lead to runtime issues, too. > > Hence, I don't think that we should add `BranchConfig`. A config object > is helpful if each configuration can be set independently of all others, > but this seems not to be the case here. If we add new configuration > later, we can also just move forward by deprecating the methods that > accept `Named` and add new methods that accepted `BranchConfig` (that > would of course implement `Named`). > > > Thoughts? > > > @Ivan, what do you think about the general idea to blend the two main > approaches of returning a `Map` plus an "embedded chaining"? > > > > -Matthias > > > > On 6/4/19 10:33 AM, John Roesler wrote: >> Thanks for the idea, Matthias, it does seem like this would satisfy >> everyone. Returning the map from the terminal operations also solves >> the problem of merging/joining the branched streams, if we want to add >> support for the compliment later on. >> >> Under your suggestion, it seems that the name is required. Otherwise, >> we wouldn't have keys for the map to return. I this this is actually >> not too bad, since experience has taught us that, although names for >> operations are not required to define stream processing logic, it does >> significantly improve the operational experience when you can map the >> topology, logs, metrics, etc. back to the source code. Since you >> wouldn't (have to) reference the name to chain extra processing onto >> the branch (thanks to the second argument), you can avoid the >> "unchecked name" problem that Ivan pointed out. >> >> In the current implementation of Branch, you can name the branch >> operator itself, and then all the branches get index-suffixed names >> built from the branch operator name. I guess under this proposal, we >> could naturally append the branch name to the branching operator name, >> like this: >> >> stream.split(Named.withName("mysplit")) //creates node "mysplit" >> .branch(..., ..., "abranch") // creates node "mysplit-abranch" >> .defaultBranch(...) // creates node "mysplit-default" >> >> It does make me wonder about the DSL syntax itself, though. >> >> We don't have a defined grammar, so there's plenty of room to debate >> the "best" syntax in the context of each operation, but in general, >> the KStream DSL operators follow this pattern: >> >> operator(function, config_object?) OR operator(config_object) >> >> where config_object is often just Named in the "function" variant. >> Even when the config_object isn't a Named, but some other config >> class, that config class _always_ implements NamedOperation. >> >> Here, we're introducing a totally different pattern: >> >> operator(function, function, string) >> >> where the string is the name. >> My first question is whether the name should instead be specified with >> the NamedOperation interface. >> >> My second question is whether we should just roll all these arguments >> up into a config object like: >> >> KBranchedStream#branch(BranchConfig) >> >> interface BranchConfig extends NamedOperation { >> withPredicate(...); >> withChain(...); >> withName(...); >> } >> >> Although I guess we'd like to call BranchConfig something more like >> "Branched", even if I don't particularly like that pattern. >> >> This makes the source code a little noisier, but it also makes us more >> future-proof, as we can deal with a wide range of alternatives purely >> in the config interface, and never have to deal with adding overloads >> to the KBranchedStream if/when we decide we want the name to be >> optional, or the KStream->KStream to be optional. >> >> WDYT? >> >> Thanks, >> -John >> >> On Fri, May 24, 2019 at 5:25 PM Michael Drogalis >> <michael.droga...@confluent.io> wrote: >>> >>> Matthias: I think that's pretty reasonable from my point of view. Good >>> suggestion. >>> >>> On Thu, May 23, 2019 at 9:50 PM Matthias J. Sax <matth...@confluent.io> >>> wrote: >>> >>>> Interesting discussion. >>>> >>>> I am wondering, if we cannot unify the advantage of both approaches: >>>> >>>> >>>> >>>> KStream#split() -> KBranchedStream >>>> >>>> // branch is not easily accessible in current scope >>>> KBranchedStream#branch(Predicate, Consumer<KStream>) >>>> -> KBranchedStream >>>> >>>> // assign a name to the branch and >>>> // return the sub-stream to the current scope later >>>> // >>>> // can be simple as `#branch(p, s->s, "name")` >>>> // or also complex as `#branch(p, s->s.filter(...), "name")` >>>> KBranchedStream#branch(Predicate, Function<KStream,KStream>, String) >>>> -> KBranchedStream >>>> >>>> // default branch is not easily accessible >>>> // return map of all named sub-stream into current scope >>>> KBranchedStream#default(Cosumer<KStream>) >>>> -> Map<String,KStream> >>>> >>>> // assign custom name to default-branch >>>> // return map of all named sub-stream into current scope >>>> KBranchedStream#default(Function<KStream,KStream>, String) >>>> -> Map<String,KStream> >>>> >>>> // assign a default name for default >>>> // return map of all named sub-stream into current scope >>>> KBranchedStream#defaultBranch(Function<KStream,KStream>) >>>> -> Map<String,KStream> >>>> >>>> // return map of all names sub-stream into current scope >>>> KBranchedStream#noDefaultBranch() >>>> -> Map<String,KStream> >>>> >>>> >>>> >>>> Hence, for each sub-stream, the user can pick to add a name and return >>>> the branch "result" to the calling scope or not. The implementation can >>>> also check at runtime that all returned names are unique. The returned >>>> Map can be empty and it's also optional to use the Map. >>>> >>>> To me, it seems like a good way to get best of both worlds. >>>> >>>> Thoughts? >>>> >>>> >>>> >>>> -Matthias >>>> >>>> >>>> >>>> >>>> On 5/6/19 5:15 PM, John Roesler wrote: >>>>> Ivan, >>>>> >>>>> That's a very good point about the "start" operator in the dynamic case. >>>>> I had no problem with "split()"; I was just questioning the necessity. >>>>> Since you've provided a proof of necessity, I'm in favor of the >>>>> "split()" start operator. Thanks! >>>>> >>>>> Separately, I'm interested to see where the present discussion leads. >>>>> I've written enough Javascript code in my life to be suspicious of >>>>> nested closures. You have a good point about using method references (or >>>>> indeed function literals also work). It should be validating that this >>>>> was also the JS community's first approach to flattening the logic when >>>>> their nested closure situation got out of hand. Unfortunately, it's >>>>> replacing nesting with redirection, both of which disrupt code >>>>> readability (but in different ways for different reasons). In other >>>>> words, I agree that function references is *the* first-order solution if >>>>> the nested code does indeed become a problem. >>>>> >>>>> However, the history of JS also tells us that function references aren't >>>>> the end of the story either, and you can see that by observing that >>>>> there have been two follow-on eras, as they continue trying to cope with >>>>> the consequences of living in such a callback-heavy language. First, you >>>>> have Futures/Promises, which essentially let you convert nested code to >>>>> method-chained code (Observables/FP is a popular variation on this). >>>>> Most lately, you have async/await, which is an effort to apply language >>>>> (not just API) syntax to the problem, and offer the "flattest" possible >>>>> programming style to solve the problem (because you get back to just one >>>>> code block per functional unit). >>>>> >>>>> Stream-processing is a different domain, and Java+KStreams is nowhere >>>>> near as callback heavy as JS, so I don't think we have to take the JS >>>>> story for granted, but then again, I think we can derive some valuable >>>>> lessons by looking sideways to adjacent domains. I'm just bringing this >>>>> up to inspire further/deeper discussion. At the same time, just like JS, >>>>> we can afford to take an iterative approach to the problem. >>>>> >>>>> Separately again, I'm interested in the post-branch merge (and I'd also >>>>> add join) problem that Paul brought up. We can clearly punt on it, by >>>>> terminating the nested branches with sink operators. But is there a DSL >>>>> way to do it? >>>>> >>>>> Thanks again for your driving this, >>>>> -John >>>>> >>>>> On Thu, May 2, 2019 at 7:39 PM Paul Whalen <pgwha...@gmail.com >>>>> <mailto:pgwha...@gmail.com>> wrote: >>>>> >>>>> Ivan, I’ll definitely forfeit my point on the clumsiness of the >>>>> branch(predicate, consumer) solution, I don’t see any real drawbacks >>>>> for the dynamic case. >>>>> >>>>> IMO the one trade off to consider at this point is the scope >>>>> question. I don’t know if I totally agree that “we rarely need them >>>>> in the same scope” since merging the branches back together later >>>>> seems like a perfectly plausible use case that can be a lot nicer >>>>> when the branched streams are in the same scope. That being said, >>>>> for the reasons Ivan listed, I think it is overall the better >>>>> solution - working around the scope thing is easy enough if you need >>>>> to. >>>>> >>>>> > On May 2, 2019, at 7:00 PM, Ivan Ponomarev >>>>> <iponoma...@mail.ru.invalid> wrote: >>>>> > >>>>> > Hello everyone, thank you all for joining the discussion! >>>>> > >>>>> > Well, I don't think the idea of named branches, be it a >>>>> LinkedHashMap (no other Map will do, because order of definition >>>>> matters) or `branch` method taking name and Consumer has more >>>>> advantages than drawbacks. >>>>> > >>>>> > In my opinion, the only real positive outcome from Michael's >>>>> proposal is that all the returned branches are in the same scope. >>>>> But 1) we rarely need them in the same scope 2) there is a >>>>> workaround for the scope problem, described in the KIP. >>>>> > >>>>> > 'Inlining the complex logic' is not a problem, because we can use >>>>> method references instead of lambdas. In real world scenarios you >>>>> tend to split the complex logic to methods anyway, so the code is >>>>> going to be clean. >>>>> > >>>>> > The drawbacks are strong. The cohesion between predicates and >>>>> handlers is lost. We have to define predicates in one place, and >>>>> handlers in another. This opens the door for bugs: >>>>> > >>>>> > - what if we forget to define a handler for a name? or a name for >>>>> a handler? >>>>> > - what if we misspell a name? >>>>> > - what if we copy-paste and duplicate a name? >>>>> > >>>>> > What Michael propose would have been totally OK if we had been >>>>> writing the API in Lua, Ruby or Python. In those languages the >>>>> "dynamic naming" approach would have looked most concise and >>>>> beautiful. But in Java we expect all the problems related to >>>>> identifiers to be eliminated in compile time. >>>>> > >>>>> > Do we have to invent duck-typing for the Java API? >>>>> > >>>>> > And if we do, what advantage are we supposed to get besides having >>>>> all the branches in the same scope? Michael, maybe I'm missing your >>>>> point? >>>>> > >>>>> > --- >>>>> > >>>>> > Earlier in this discussion John Roesler also proposed to do >>>>> without "start branching" operator, and later Paul mentioned that in >>>>> the case when we have to add a dynamic number of branches, the >>>>> current KIP is 'clumsier' compared to Michael's 'Map' solution. Let >>>>> me address both comments here. >>>>> > >>>>> > 1) "Start branching" operator (I think that *split* is a good name >>>>> for it indeed) is critical when we need to do a dynamic branching, >>>>> see example below. >>>>> > >>>>> > 2) No, dynamic branching in current KIP is not clumsy at all. >>>>> Imagine a real-world scenario when you need one branch per enum >>>>> value (say, RecordType). You can have something like this: >>>>> > >>>>> > /*John:if we had to start with stream.branch(...) here, it would >>>>> have been much messier.*/ >>>>> > KBranchedStream branched = stream.split(); >>>>> > >>>>> > /*Not clumsy at all :-)*/ >>>>> > for (RecordType recordType : RecordType.values()) >>>>> > branched = branched.branch((k, v) -> v.getRecType() == >>>>> recordType, >>>>> > recordType::processRecords); >>>>> > >>>>> > Regards, >>>>> > >>>>> > Ivan >>>>> > >>>>> > >>>>> > 02.05.2019 14:40, Matthias J. Sax пишет: >>>>> >> I also agree with Michael's observation about the core problem of >>>>> >> current `branch()` implementation. >>>>> >> >>>>> >> However, I also don't like to pass in a clumsy Map object. My >>>>> thinking >>>>> >> was more aligned with Paul's proposal to just add a name to each >>>>> >> `branch()` statement and return a `Map<String,KStream>`. >>>>> >> >>>>> >> It makes the code easier to read, and also make the order of >>>>> >> `Predicates` (that is essential) easier to grasp. >>>>> >> >>>>> >>>>>> Map<String, KStream<K, V>> branches = stream.split() >>>>> >>>>>> .branch("branchOne", Predicate<K, V>) >>>>> >>>>>> .branch( "branchTwo", Predicate<K, V>) >>>>> >>>>>> .defaultBranch("defaultBranch"); >>>>> >> An open question is the case for which no defaultBranch() should >>>> be >>>>> >> specified. Atm, `split()` and `branch()` would return >>>>> `BranchedKStream` >>>>> >> and the call to `defaultBranch()` that returns the `Map` is >>>> mandatory >>>>> >> (what is not the case atm). Or is this actually not a real >>>> problem, >>>>> >> because users can just ignore the branch returned by >>>>> `defaultBranch()` >>>>> >> in the result `Map` ? >>>>> >> >>>>> >> >>>>> >> About "inlining": So far, it seems to be a matter of personal >>>>> >> preference. I can see arguments for both, but no "killer >>>>> argument" yet >>>>> >> that clearly make the case for one or the other. >>>>> >> >>>>> >> >>>>> >> -Matthias >>>>> >> >>>>> >>> On 5/1/19 6:26 PM, Paul Whalen wrote: >>>>> >>> Perhaps inlining is the wrong terminology. It doesn’t require >>>>> that a lambda with the full downstream topology be defined inline - >>>>> it can be a method reference as with Ivan’s original suggestion. >>>>> The advantage of putting the predicate and its downstream logic >>>>> (Consumer) together in branch() is that they are required to be near >>>>> to each other. >>>>> >>> >>>>> >>> Ultimately the downstream code has to live somewhere, and deep >>>>> branch trees will be hard to read regardless. >>>>> >>> >>>>> >>>> On May 1, 2019, at 1:07 PM, Michael Drogalis >>>>> <michael.droga...@confluent.io >>>>> <mailto:michael.droga...@confluent.io>> wrote: >>>>> >>>> >>>>> >>>> I'm less enthusiastic about inlining the branch logic with its >>>>> downstream >>>>> >>>> functionality. Programs that have deep branch trees will >>>>> quickly become >>>>> >>>> harder to read as a single unit. >>>>> >>>> >>>>> >>>>> On Tue, Apr 30, 2019 at 8:34 PM Paul Whalen >>>>> <pgwha...@gmail.com <mailto:pgwha...@gmail.com>> wrote: >>>>> >>>>> >>>>> >>>>> Also +1 on the issues/goals as Michael outlined them, I think >>>>> that sets a >>>>> >>>>> great framework for the discussion. >>>>> >>>>> >>>>> >>>>> Regarding the SortedMap solution, my understanding is that the >>>>> current >>>>> >>>>> proposal in the KIP is what is in my PR which (pending naming >>>>> decisions) is >>>>> >>>>> roughly this: >>>>> >>>>> >>>>> >>>>> stream.split() >>>>> >>>>> .branch(Predicate<K, V>, Consumer<KStream<K, V>>) >>>>> >>>>> .branch(Predicate<K, V>, Consumer<KStream<K, V>>) >>>>> >>>>> .defaultBranch(Consumer<KStream<K, V>>); >>>>> >>>>> >>>>> >>>>> Obviously some ordering is necessary, since branching as a >>>>> construct >>>>> >>>>> doesn't work without it, but this solution seems like it >>>>> provides as much >>>>> >>>>> associativity as the SortedMap solution, because each branch() >>>>> call >>>>> >>>>> directly associates the "conditional" with the "code block." >>>>> The value it >>>>> >>>>> provides over the KIP solution is the accessing of streams in >>>>> the same >>>>> >>>>> scope. >>>>> >>>>> >>>>> >>>>> The KIP solution is less "dynamic" than the SortedMap solution >>>>> in the sense >>>>> >>>>> that it is slightly clumsier to add a dynamic number of >>>>> branches, but it is >>>>> >>>>> certainly possible. It seems to me like the API should favor >>>>> the "static" >>>>> >>>>> case anyway, and should make it simple and readable to >>>>> fluently declare and >>>>> >>>>> access your branches in-line. It also makes it impossible to >>>>> ignore a >>>>> >>>>> branch, and it is possible to build an (almost) identical >>>>> SortedMap >>>>> >>>>> solution on top of it. >>>>> >>>>> >>>>> >>>>> I could also see a middle ground where instead of a raw >>>>> SortedMap being >>>>> >>>>> taken in, branch() takes a name and not a Consumer. Something >>>>> like this: >>>>> >>>>> >>>>> >>>>> Map<String, KStream<K, V>> branches = stream.split() >>>>> >>>>> .branch("branchOne", Predicate<K, V>) >>>>> >>>>> .branch( "branchTwo", Predicate<K, V>) >>>>> >>>>> .defaultBranch("defaultBranch", Consumer<KStream<K, V>>); >>>>> >>>>> >>>>> >>>>> Pros for that solution: >>>>> >>>>> - accessing branched KStreams in same scope >>>>> >>>>> - no double brace initialization, hopefully slightly more >>>>> readable than >>>>> >>>>> SortedMap >>>>> >>>>> >>>>> >>>>> Cons >>>>> >>>>> - downstream branch logic cannot be specified inline which >>>>> makes it harder >>>>> >>>>> to read top to bottom (like existing API and SortedMap, but >>>>> unlike the KIP) >>>>> >>>>> - you can forget to "handle" one of the branched streams (like >>>>> existing >>>>> >>>>> API and SortedMap, but unlike the KIP) >>>>> >>>>> >>>>> >>>>> (KBranchedStreams could even work *both* ways but perhaps >>>>> that's overdoing >>>>> >>>>> it). >>>>> >>>>> >>>>> >>>>> Overall I'm curious how important it is to be able to easily >>>>> access the >>>>> >>>>> branched KStream in the same scope as the original. It's >>>>> possible that it >>>>> >>>>> doesn't need to be handled directly by the API, but instead >>>>> left up to the >>>>> >>>>> user. I'm sort of in the middle on it. >>>>> >>>>> >>>>> >>>>> Paul >>>>> >>>>> >>>>> >>>>> >>>>> >>>>> >>>>> >>>>> On Tue, Apr 30, 2019 at 8:48 PM Sophie Blee-Goldman >>>>> <sop...@confluent.io <mailto:sop...@confluent.io>> >>>>> >>>>> wrote: >>>>> >>>>> >>>>> >>>>>> I'd like to +1 what Michael said about the issues with the >>>>> existing >>>>> >>>>> branch >>>>> >>>>>> method, I agree with what he's outlined and I think we should >>>>> proceed by >>>>> >>>>>> trying to alleviate these problems. Specifically it seems >>>>> important to be >>>>> >>>>>> able to cleanly access the individual branches (eg by mapping >>>>> >>>>>> name->stream), which I thought was the original intention of >>>>> this KIP. >>>>> >>>>>> >>>>> >>>>>> That said, I don't think we should so easily give in to the >>>>> double brace >>>>> >>>>>> anti-pattern or force ours users into it if at all possible to >>>>> >>>>> avoid...just >>>>> >>>>>> my two cents. >>>>> >>>>>> >>>>> >>>>>> Cheers, >>>>> >>>>>> Sophie >>>>> >>>>>> >>>>> >>>>>> On Tue, Apr 30, 2019 at 12:59 PM Michael Drogalis < >>>>> >>>>>> michael.droga...@confluent.io >>>>> <mailto:michael.droga...@confluent.io>> wrote: >>>>> >>>>>> >>>>> >>>>>>> I’d like to propose a different way of thinking about this. >>>>> To me, >>>>> >>>>> there >>>>> >>>>>>> are three problems with the existing branch signature: >>>>> >>>>>>> >>>>> >>>>>>> 1. If you use it the way most people do, Java raises unsafe >>>> type >>>>> >>>>>> warnings. >>>>> >>>>>>> 2. The way in which you use the stream branches is >>>>> positionally coupled >>>>> >>>>>> to >>>>> >>>>>>> the ordering of the conditionals. >>>>> >>>>>>> 3. It is brittle to extend existing branch calls with >>>>> additional code >>>>> >>>>>>> paths. >>>>> >>>>>>> >>>>> >>>>>>> Using associative constructs instead of relying on ordered >>>>> constructs >>>>> >>>>>> would >>>>> >>>>>>> be a stronger approach. Consider a signature that instead >>>>> looks like >>>>> >>>>>> this: >>>>> >>>>>>> Map<String, KStream<K,V>> KStream#branch(SortedMap<String, >>>>> Predicate<? >>>>> >>>>>>> super K,? super V>>); >>>>> >>>>>>> >>>>> >>>>>>> Branches are given names in a map, and as a result, the API >>>>> returns a >>>>> >>>>>>> mapping of names to streams. The ordering of the >>>> conditionals is >>>>> >>>>>> maintained >>>>> >>>>>>> because it’s a sorted map. Insert order determines the order >>>> of >>>>> >>>>>> evaluation. >>>>> >>>>>>> This solves problem 1 because there are no more varargs. It >>>>> solves >>>>> >>>>>> problem >>>>> >>>>>>> 2 because you no longer lean on ordering to access the >>>>> branch you’re >>>>> >>>>>>> interested in. It solves problem 3 because you can introduce >>>>> another >>>>> >>>>>>> conditional by simply attaching another name to the >>>>> structure, rather >>>>> >>>>>> than >>>>> >>>>>>> messing with the existing indices. >>>>> >>>>>>> >>>>> >>>>>>> One of the drawbacks is that creating the map inline is >>>>> historically >>>>> >>>>>>> awkward in Java. I know it’s an anti-pattern to use >>>>> voluminously, but >>>>> >>>>>>> double brace initialization would clean up the aesthetics. >>>>> >>>>>>> >>>>> >>>>>>> On Tue, Apr 30, 2019 at 9:10 AM John Roesler >>>>> <j...@confluent.io <mailto:j...@confluent.io>> >>>>> >>>>> wrote: >>>>> >>>>>>>> Hi Ivan, >>>>> >>>>>>>> >>>>> >>>>>>>> Thanks for the update. >>>>> >>>>>>>> >>>>> >>>>>>>> FWIW, I agree with Matthias that the current "start >>>> branching" >>>>> >>>>> operator >>>>> >>>>>>> is >>>>> >>>>>>>> confusing when named the same way as the actual branches. >>>>> "Split" >>>>> >>>>> seems >>>>> >>>>>>>> like a good name. Alternatively, we can do without a "start >>>>> >>>>> branching" >>>>> >>>>>>>> operator at all, and just do: >>>>> >>>>>>>> >>>>> >>>>>>>> stream >>>>> >>>>>>>> .branch(Predicate) >>>>> >>>>>>>> .branch(Predicate) >>>>> >>>>>>>> .defaultBranch(); >>>>> >>>>>>>> >>>>> >>>>>>>> Tentatively, I think that this branching operation should be >>>>> >>>>> terminal. >>>>> >>>>>>> That >>>>> >>>>>>>> way, we don't create ambiguity about how to use it. That >>>>> is, `branch` >>>>> >>>>>>>> should return `KBranchedStream`, while `defaultBranch` is >>>>> `void`, to >>>>> >>>>>>>> enforce that it comes last, and that there is only one >>>>> definition of >>>>> >>>>>> the >>>>> >>>>>>>> default branch. Potentially, we should log a warning if >>>>> there's no >>>>> >>>>>>> default, >>>>> >>>>>>>> and additionally log a warning (or throw an exception) if a >>>>> record >>>>> >>>>>> falls >>>>> >>>>>>>> though with no default. >>>>> >>>>>>>> >>>>> >>>>>>>> Thoughts? >>>>> >>>>>>>> >>>>> >>>>>>>> Thanks, >>>>> >>>>>>>> -John >>>>> >>>>>>>> >>>>> >>>>>>>> On Fri, Apr 26, 2019 at 3:40 AM Matthias J. Sax < >>>>> >>>>> matth...@confluent.io <mailto:matth...@confluent.io> >>>>> >>>>>>>> wrote: >>>>> >>>>>>>> >>>>> >>>>>>>>> Thanks for updating the KIP and your answers. >>>>> >>>>>>>>> >>>>> >>>>>>>>> >>>>> >>>>>>>>>> this is to make the name similar to String#split >>>>> >>>>>>>>>>> that also returns an array, right? >>>>> >>>>>>>>> The intend was to avoid name duplication. The return type >>>>> should >>>>> >>>>>> _not_ >>>>> >>>>>>>>> be an array. >>>>> >>>>>>>>> >>>>> >>>>>>>>> The current proposal is >>>>> >>>>>>>>> >>>>> >>>>>>>>> stream.branch() >>>>> >>>>>>>>> .branch(Predicate) >>>>> >>>>>>>>> .branch(Predicate) >>>>> >>>>>>>>> .defaultBranch(); >>>>> >>>>>>>>> >>>>> >>>>>>>>> IMHO, this reads a little odd, because the first >>>>> `branch()` does >>>>> >>>>> not >>>>> >>>>>>>>> take any parameters and has different semantics than the >>>> later >>>>> >>>>>>>>> `branch()` calls. Note, that from the code snippet above, >>>> it's >>>>> >>>>> hidden >>>>> >>>>>>>>> that the first call is `KStream#branch()` while the others >>>> are >>>>> >>>>>>>>> `KBranchedStream#branch()` what makes reading the code >>>> harder. >>>>> >>>>>>>>> >>>>> >>>>>>>>> Because I suggested to rename `addBranch()` -> `branch()`, >>>>> I though >>>>> >>>>>> it >>>>> >>>>>>>>> might be better to also rename `KStream#branch()` to avoid >>>> the >>>>> >>>>> naming >>>>> >>>>>>>>> overlap that seems to be confusing. The following reads >>>> much >>>>> >>>>> cleaner >>>>> >>>>>> to >>>>> >>>>>>>> me: >>>>> >>>>>>>>> stream.split() >>>>> >>>>>>>>> .branch(Predicate) >>>>> >>>>>>>>> .branch(Predicate) >>>>> >>>>>>>>> .defaultBranch(); >>>>> >>>>>>>>> >>>>> >>>>>>>>> Maybe there is a better alternative to `split()` though to >>>>> avoid >>>>> >>>>> the >>>>> >>>>>>>>> naming overlap. >>>>> >>>>>>>>> >>>>> >>>>>>>>> >>>>> >>>>>>>>>> 'default' is, however, a reserved word, so unfortunately >>>> we >>>>> >>>>> cannot >>>>> >>>>>>> have >>>>> >>>>>>>>> a method with such name :-) >>>>> >>>>>>>>> >>>>> >>>>>>>>> Bummer. Didn't consider this. Maybe we can still come up >>>>> with a >>>>> >>>>> short >>>>> >>>>>>>> name? >>>>> >>>>>>>>> >>>>> >>>>>>>>> Can you add the interface `KBranchedStream` to the KIP >>>>> with all >>>>> >>>>> it's >>>>> >>>>>>>>> methods? It will be part of public API and should be >>>>> contained in >>>>> >>>>> the >>>>> >>>>>>>>> KIP. For example, it's unclear atm, what the return type of >>>>> >>>>>>>>> `defaultBranch()` is. >>>>> >>>>>>>>> >>>>> >>>>>>>>> >>>>> >>>>>>>>> You did not comment on the idea to add a >>>>> `KBranchedStream#get(int >>>>> >>>>>>> index) >>>>> >>>>>>>>> -> KStream` method to get the individually >>>>> branched-KStreams. Would >>>>> >>>>>> be >>>>> >>>>>>>>> nice to get your feedback about it. It seems you suggest >>>>> that users >>>>> >>>>>>>>> would need to write custom utility code otherwise, to >>>>> access them. >>>>> >>>>> We >>>>> >>>>>>>>> should discuss the pros and cons of both approaches. It >>>> feels >>>>> >>>>>>>>> "incomplete" to me atm, if the API has no built-in support >>>>> to get >>>>> >>>>> the >>>>> >>>>>>>>> branched-KStreams directly. >>>>> >>>>>>>>> >>>>> >>>>>>>>> >>>>> >>>>>>>>> >>>>> >>>>>>>>> -Matthias >>>>> >>>>>>>>> >>>>> >>>>>>>>> >>>>> >>>>>>>>>> On 4/13/19 2:13 AM, Ivan Ponomarev wrote: >>>>> >>>>>>>>>> Hi all! >>>>> >>>>>>>>>> >>>>> >>>>>>>>>> I have updated the KIP-418 according to the new vision. >>>>> >>>>>>>>>> >>>>> >>>>>>>>>> Matthias, thanks for your comment! >>>>> >>>>>>>>>> >>>>> >>>>>>>>>>> Renaming KStream#branch() -> #split() >>>>> >>>>>>>>>> I can see your point: this is to make the name similar to >>>>> >>>>>>> String#split >>>>> >>>>>>>>>> that also returns an array, right? But is it worth the >>>>> loss of >>>>> >>>>>>>> backwards >>>>> >>>>>>>>>> compatibility? We can have overloaded branch() as well >>>>> without >>>>> >>>>>>>> affecting >>>>> >>>>>>>>>> the existing code. Maybe the old array-based `branch` >>>> method >>>>> >>>>> should >>>>> >>>>>>> be >>>>> >>>>>>>>>> deprecated, but this is a subject for discussion. >>>>> >>>>>>>>>> >>>>> >>>>>>>>>>> Renaming KBranchedStream#addBranch() -> >>>>> >>>>> BranchingKStream#branch(), >>>>> >>>>>>>>>> KBranchedStream#defaultBranch() -> >>>> BranchingKStream#default() >>>>> >>>>>>>>>> >>>>> >>>>>>>>>> Totally agree with 'addBranch->branch' rename. 'default' >>>> is, >>>>> >>>>>>> however, a >>>>> >>>>>>>>>> reserved word, so unfortunately we cannot have a method >>>>> with such >>>>> >>>>>>> name >>>>> >>>>>>>>> :-) >>>>> >>>>>>>>>>> defaultBranch() does take an `Predicate` as argument, >>>> but I >>>>> >>>>> think >>>>> >>>>>>> that >>>>> >>>>>>>>>> is not required? >>>>> >>>>>>>>>> >>>>> >>>>>>>>>> Absolutely! I think that was just copy-paste error or >>>>> something. >>>>> >>>>>>>>>> >>>>> >>>>>>>>>> Dear colleagues, >>>>> >>>>>>>>>> >>>>> >>>>>>>>>> please revise the new version of the KIP and Paul's PR >>>>> >>>>>>>>>> (https://github.com/apache/kafka/pull/6512) >>>>> >>>>>>>>>> >>>>> >>>>>>>>>> Any new suggestions/objections? >>>>> >>>>>>>>>> >>>>> >>>>>>>>>> Regards, >>>>> >>>>>>>>>> >>>>> >>>>>>>>>> Ivan >>>>> >>>>>>>>>> >>>>> >>>>>>>>>> >>>>> >>>>>>>>>> 11.04.2019 11:47, Matthias J. Sax пишет: >>>>> >>>>>>>>>>> Thanks for driving the discussion of this KIP. It seems >>>> that >>>>> >>>>>>> everybody >>>>> >>>>>>>>>>> agrees that the current branch() method using arrays is >>>> not >>>>> >>>>>> optimal. >>>>> >>>>>>>>>>> I had a quick look into the PR and I like the overall >>>>> proposal. >>>>> >>>>>>> There >>>>> >>>>>>>>>>> are some minor things we need to consider. I would >>>>> recommend the >>>>> >>>>>>>>>>> following renaming: >>>>> >>>>>>>>>>> >>>>> >>>>>>>>>>> KStream#branch() -> #split() >>>>> >>>>>>>>>>> KBranchedStream#addBranch() -> BranchingKStream#branch() >>>>> >>>>>>>>>>> KBranchedStream#defaultBranch() -> >>>>> BranchingKStream#default() >>>>> >>>>>>>>>>> >>>>> >>>>>>>>>>> It's just a suggestion to get slightly shorter method >>>> names. >>>>> >>>>>>>>>>> >>>>> >>>>>>>>>>> In the current PR, defaultBranch() does take an >>>>> `Predicate` as >>>>> >>>>>>>> argument, >>>>> >>>>>>>>>>> but I think that is not required? >>>>> >>>>>>>>>>> >>>>> >>>>>>>>>>> Also, we should consider KIP-307, that was recently >>>>> accepted and >>>>> >>>>>> is >>>>> >>>>>>>>>>> currently implemented: >>>>> >>>>>>>>>>> >>>>> >>>>> >>>>> >>>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-307%3A+Allow+to+define+custom+processor+names+with+KStreams+DSL >>>>> >>>>>>>>>>> Ie, we should add overloads that accepted a `Named` >>>>> parameter. >>>>> >>>>>>>>>>> >>>>> >>>>>>>>>>> >>>>> >>>>>>>>>>> For the issue that the created `KStream` object are in >>>>> different >>>>> >>>>>>>> scopes: >>>>> >>>>>>>>>>> could we extend `KBranchedStream` with a `get(int >>>>> index)` method >>>>> >>>>>>> that >>>>> >>>>>>>>>>> returns the corresponding "branched" result `KStream` >>>>> object? >>>>> >>>>>> Maybe, >>>>> >>>>>>>> the >>>>> >>>>>>>>>>> second argument of `addBranch()` should not be a >>>>> >>>>>> `Consumer<KStream>` >>>>> >>>>>>>> but >>>>> >>>>>>>>>>> a `Function<KStream,KStream>` and `get()` could return >>>>> whatever >>>>> >>>>>> the >>>>> >>>>>>>>>>> `Function` returns? >>>>> >>>>>>>>>>> >>>>> >>>>>>>>>>> >>>>> >>>>>>>>>>> Finally, I would also suggest to update the KIP with the >>>>> current >>>>> >>>>>>>>>>> proposal. That makes it easier to review. >>>>> >>>>>>>>>>> >>>>> >>>>>>>>>>> >>>>> >>>>>>>>>>> -Matthias >>>>> >>>>>>>>>>> >>>>> >>>>>>>>>>> >>>>> >>>>>>>>>>> >>>>> >>>>>>>>>>>> On 3/31/19 12:22 PM, Paul Whalen wrote: >>>>> >>>>>>>>>>>> Ivan, >>>>> >>>>>>>>>>>> >>>>> >>>>>>>>>>>> I'm a bit of a novice here as well, but I think it >>>>> makes sense >>>>> >>>>>> for >>>>> >>>>>>>> you >>>>> >>>>>>>>> to >>>>> >>>>>>>>>>>> revise the KIP and continue the discussion. Obviously >>>>> we'll >>>>> >>>>> need >>>>> >>>>>>>> some >>>>> >>>>>>>>>>>> buy-in from committers that have actual binding votes on >>>>> >>>>> whether >>>>> >>>>>>> the >>>>> >>>>>>>>> KIP >>>>> >>>>>>>>>>>> could be adopted. It would be great to hear if they >>>>> think this >>>>> >>>>>> is >>>>> >>>>>>> a >>>>> >>>>>>>>> good >>>>> >>>>>>>>>>>> idea overall. I'm not sure if that happens just by >>>>> starting a >>>>> >>>>>>> vote, >>>>> >>>>>>>>> or if >>>>> >>>>>>>>>>>> there is generally some indication of interest >>>> beforehand. >>>>> >>>>>>>>>>>> >>>>> >>>>>>>>>>>> That being said, I'll continue the discussion a bit: >>>>> assuming >>>>> >>>>> we >>>>> >>>>>> do >>>>> >>>>>>>>> move >>>>> >>>>>>>>>>>> forward the solution of "stream.branch() returns >>>>> >>>>>> KBranchedStream", >>>>> >>>>>>> do >>>>> >>>>>>>>> we >>>>> >>>>>>>>>>>> deprecate "stream.branch(...) returns KStream[]"? I >>>> would >>>>> >>>>> favor >>>>> >>>>>>>>>>>> deprecating, since having two mutually exclusive APIs >>>> that >>>>> >>>>>>> accomplish >>>>> >>>>>>>>> the >>>>> >>>>>>>>>>>> same thing is confusing, especially when they're fairly >>>>> similar >>>>> >>>>>>>>> anyway. We >>>>> >>>>>>>>>>>> just need to be sure we're not making something >>>>> >>>>>>> impossible/difficult >>>>> >>>>>>>>> that >>>>> >>>>>>>>>>>> is currently possible/easy. >>>>> >>>>>>>>>>>> >>>>> >>>>>>>>>>>> Regarding my PR - I think the general structure would >>>> work, >>>>> >>>>> it's >>>>> >>>>>>>> just a >>>>> >>>>>>>>>>>> little sloppy overall in terms of naming and clarity. In >>>>> >>>>>>> particular, >>>>> >>>>>>>>>>>> passing in the "predicates" and "children" lists which >>>> get >>>>> >>>>>> modified >>>>> >>>>>>>> in >>>>> >>>>>>>>>>>> KBranchedStream but read from all the way >>>>> KStreamLazyBranch is >>>>> >>>>> a >>>>> >>>>>>> bit >>>>> >>>>>>>>>>>> complicated to follow. >>>>> >>>>>>>>>>>> >>>>> >>>>>>>>>>>> Thanks, >>>>> >>>>>>>>>>>> Paul >>>>> >>>>>>>>>>>> >>>>> >>>>>>>>>>>> On Fri, Mar 29, 2019 at 5:37 AM Ivan Ponomarev < >>>>> >>>>>> iponoma...@mail.ru <mailto:iponoma...@mail.ru> >>>>> >>>>>>>>> wrote: >>>>> >>>>>>>>>>>>> Hi Paul! >>>>> >>>>>>>>>>>>> >>>>> >>>>>>>>>>>>> I read your code carefully and now I am fully >>>>> convinced: your >>>>> >>>>>>>> proposal >>>>> >>>>>>>>>>>>> looks better and should work. We just have to document >>>> the >>>>> >>>>>> crucial >>>>> >>>>>>>>> fact >>>>> >>>>>>>>>>>>> that KStream consumers are invoked as they're added. >>>>> And then >>>>> >>>>>> it's >>>>> >>>>>>>> all >>>>> >>>>>>>>>>>>> going to be very nice. >>>>> >>>>>>>>>>>>> >>>>> >>>>>>>>>>>>> What shall we do now? I should re-write the KIP and >>>>> resume the >>>>> >>>>>>>>>>>>> discussion here, right? >>>>> >>>>>>>>>>>>> >>>>> >>>>>>>>>>>>> Why are you telling that your PR 'should not be even a >>>>> >>>>> starting >>>>> >>>>>>>> point >>>>> >>>>>>>>> if >>>>> >>>>>>>>>>>>> we go in this direction'? To me it looks like a good >>>>> starting >>>>> >>>>>>> point. >>>>> >>>>>>>>> But >>>>> >>>>>>>>>>>>> as a novice in this project I might miss some important >>>>> >>>>> details. >>>>> >>>>>>>>>>>>> Regards, >>>>> >>>>>>>>>>>>> >>>>> >>>>>>>>>>>>> Ivan >>>>> >>>>>>>>>>>>> >>>>> >>>>>>>>>>>>> >>>>> >>>>>>>>>>>>> 28.03.2019 17:38, Paul Whalen пишет: >>>>> >>>>>>>>>>>>>> Ivan, >>>>> >>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>> Maybe I’m missing the point, but I believe the >>>>> >>>>> stream.branch() >>>>> >>>>>>>>> solution >>>>> >>>>>>>>>>>>> supports this. The couponIssuer::set* consumers will be >>>>> >>>>> invoked >>>>> >>>>>> as >>>>> >>>>>>>>> they’re >>>>> >>>>>>>>>>>>> added, not during streamsBuilder.build(). So the user >>>>> still >>>>> >>>>>> ought >>>>> >>>>>>> to >>>>> >>>>>>>>> be >>>>> >>>>>>>>>>>>> able to call couponIssuer.coupons() afterward and >>>>> depend on >>>>> >>>>> the >>>>> >>>>>>>>> branched >>>>> >>>>>>>>>>>>> streams having been set. >>>>> >>>>>>>>>>>>>> The issue I mean to point out is that it is hard to >>>>> access >>>>> >>>>> the >>>>> >>>>>>>>> branched >>>>> >>>>>>>>>>>>> streams in the same scope as the original stream (that >>>>> is, not >>>>> >>>>>>>> inside >>>>> >>>>>>>>> the >>>>> >>>>>>>>>>>>> couponIssuer), which is a problem with both proposed >>>>> >>>>> solutions. >>>>> >>>>>> It >>>>> >>>>>>>>> can be >>>>> >>>>>>>>>>>>> worked around though. >>>>> >>>>>>>>>>>>>> [Also, great to hear additional interest in 401, I’m >>>>> excited >>>>> >>>>> to >>>>> >>>>>>>> hear >>>>> >>>>>>>>>>>>> your thoughts!] >>>>> >>>>>>>>>>>>>> Paul >>>>> >>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>> On Mar 28, 2019, at 4:00 AM, Ivan Ponomarev < >>>>> >>>>>> iponoma...@mail.ru <mailto:iponoma...@mail.ru> >>>>> >>>>>>>>> wrote: >>>>> >>>>>>>>>>>>>>> Hi Paul! >>>>> >>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>> The idea to postpone the wiring of branches to the >>>>> >>>>>>>>>>>>> streamsBuilder.build() also looked great for me at >>>> first >>>>> >>>>> glance, >>>>> >>>>>>> but >>>>> >>>>>>>>> --- >>>>> >>>>>>>>>>>>>>>> the newly branched streams are not available in the >>>>> same >>>>> >>>>>> scope >>>>> >>>>>>> as >>>>> >>>>>>>>> each >>>>> >>>>>>>>>>>>> other. That is, if we wanted to merge them back >>>> together >>>>> >>>>> again >>>>> >>>>>> I >>>>> >>>>>>>>> don't see >>>>> >>>>>>>>>>>>> a way to do that. >>>>> >>>>>>>>>>>>>>> You just took the words right out of my mouth, I was >>>>> just >>>>> >>>>>> going >>>>> >>>>>>> to >>>>> >>>>>>>>>>>>> write in details about this issue. >>>>> >>>>>>>>>>>>>>> Consider the example from Bill's book, p. 101: say >>>>> we need >>>>> >>>>> to >>>>> >>>>>>>>> identify >>>>> >>>>>>>>>>>>> customers who have bought coffee and made a purchase >>>>> in the >>>>> >>>>>>>>> electronics >>>>> >>>>>>>>>>>>> store to give them coupons. >>>>> >>>>>>>>>>>>>>> This is the code I usually write under these >>>>> circumstances >>>>> >>>>>> using >>>>> >>>>>>>> my >>>>> >>>>>>>>>>>>> 'brancher' class: >>>>> >>>>>>>>>>>>>>> @Setter >>>>> >>>>>>>>>>>>>>> class CouponIssuer{ >>>>> >>>>>>>>>>>>>>> private KStream<....> coffePurchases; >>>>> >>>>>>>>>>>>>>> private KStream<....> electronicsPurchases; >>>>> >>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>> KStream<...> coupons(){ >>>>> >>>>>>>>>>>>>>> return >>>>> >>>>>>>>> coffePurchases.join(electronicsPurchases...)...whatever >>>>> >>>>>>>>>>>>>>> /*In the real world the code here can be >>>>> complex, so >>>>> >>>>>>>>> creation of >>>>> >>>>>>>>>>>>> a separate CouponIssuer class is fully justified, in >>>>> order to >>>>> >>>>>>>> separate >>>>> >>>>>>>>>>>>> classes' responsibilities.*/ >>>>> >>>>>>>>>>>>>>> } >>>>> >>>>>>>>>>>>>>> } >>>>> >>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>> CouponIssuer couponIssuer = new CouponIssuer(); >>>>> >>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>> new KafkaStreamsBrancher<....>() >>>>> >>>>>>>>>>>>>>> .branch(predicate1, >>>> couponIssuer::setCoffePurchases) >>>>> >>>>>>>>>>>>>>> .branch(predicate2, >>>>> >>>>>> couponIssuer::setElectronicsPurchases) >>>>> >>>>>>>>>>>>>>> .onTopOf(transactionStream); >>>>> >>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>> /*Alas, this won't work if we're going to wire up >>>>> everything >>>>> >>>>>>>> later, >>>>> >>>>>>>>>>>>> without the terminal operation!!!*/ >>>>> >>>>>>>>>>>>>>> couponIssuer.coupons()... >>>>> >>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>> Does this make sense? In order to properly >>>>> initialize the >>>>> >>>>>>>>> CouponIssuer >>>>> >>>>>>>>>>>>> we need the terminal operation to be called before >>>>> >>>>>>>>> streamsBuilder.build() >>>>> >>>>>>>>>>>>> is called. >>>>> >>>>>>>>>>>>>>> [BTW Paul, I just found out that your KIP-401 is >>>>> essentially >>>>> >>>>>> the >>>>> >>>>>>>>> next >>>>> >>>>>>>>>>>>> KIP I was going to write here. I have some thoughts >>>>> based on >>>>> >>>>> my >>>>> >>>>>>>>> experience, >>>>> >>>>>>>>>>>>> so I will join the discussion on KIP-401 soon.] >>>>> >>>>>>>>>>>>>>> Regards, >>>>> >>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>> Ivan >>>>> >>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>> 28.03.2019 6:29, Paul Whalen пишет: >>>>> >>>>>>>>>>>>>>>> Ivan, >>>>> >>>>>>>>>>>>>>>> I tried to make a very rough proof of concept of a >>>>> fluent >>>>> >>>>> API >>>>> >>>>>>>> based >>>>> >>>>>>>>>>>>> off of >>>>> >>>>>>>>>>>>>>>> KStream here >>>>> (https://github.com/apache/kafka/pull/6512), >>>>> >>>>>> and >>>>> >>>>>>> I >>>>> >>>>>>>>> think >>>>> >>>>>>>>>>>>> I >>>>> >>>>>>>>>>>>>>>> succeeded at removing both cons. >>>>> >>>>>>>>>>>>>>>> - Compatibility: I was incorrect earlier about >>>>> >>>>>>> compatibility >>>>> >>>>>>>>>>>>> issues, >>>>> >>>>>>>>>>>>>>>> there aren't any direct ones. I was unaware >>>>> that Java >>>>> >>>>> is >>>>> >>>>>>>> smart >>>>> >>>>>>>>>>>>> enough to >>>>> >>>>>>>>>>>>>>>> distinguish between a branch(varargs...) >>>>> returning one >>>>> >>>>>>> thing >>>>> >>>>>>>>> and >>>>> >>>>>>>>>>>>> branch() >>>>> >>>>>>>>>>>>>>>> with no arguments returning another thing. >>>>> >>>>>>>>>>>>>>>> - Requiring a terminal method: We don't actually >>>>> need >>>>> >>>>> it. >>>>> >>>>>>> We >>>>> >>>>>>>>> can >>>>> >>>>>>>>>>>>> just >>>>> >>>>>>>>>>>>>>>> build up the branches in the KBranchedStream who >>>>> shares >>>>> >>>>>> its >>>>> >>>>>>>>> state >>>>> >>>>>>>>>>>>> with the >>>>> >>>>>>>>>>>>>>>> ProcessorSupplier that will actually do the >>>>> branching. >>>>> >>>>>>> It's >>>>> >>>>>>>>> not >>>>> >>>>>>>>>>>>> terribly >>>>> >>>>>>>>>>>>>>>> pretty in its current form, but I think it >>>>> demonstrates >>>>> >>>>>> its >>>>> >>>>>>>>>>>>> feasibility. >>>>> >>>>>>>>>>>>>>>> To be clear, I don't think that pull request should >>>> be >>>>> >>>>> final >>>>> >>>>>> or >>>>> >>>>>>>>> even a >>>>> >>>>>>>>>>>>>>>> starting point if we go in this direction, I just >>>>> wanted to >>>>> >>>>>> see >>>>> >>>>>>>> how >>>>> >>>>>>>>>>>>>>>> challenging it would be to get the API working. >>>>> >>>>>>>>>>>>>>>> I will say though, that I'm not sure the existing >>>>> solution >>>>> >>>>>>> could >>>>> >>>>>>>> be >>>>> >>>>>>>>>>>>>>>> deprecated in favor of this, which I had originally >>>>> >>>>> suggested >>>>> >>>>>>>> was a >>>>> >>>>>>>>>>>>>>>> possibility. The reason is that the newly branched >>>>> streams >>>>> >>>>>> are >>>>> >>>>>>>> not >>>>> >>>>>>>>>>>>>>>> available in the same scope as each other. That >>>>> is, if we >>>>> >>>>>>> wanted >>>>> >>>>>>>>> to >>>>> >>>>>>>>>>>>> merge >>>>> >>>>>>>>>>>>>>>> them back together again I don't see a way to do >>>>> that. The >>>>> >>>>>> KIP >>>>> >>>>>>>>>>>>> proposal >>>>> >>>>>>>>>>>>>>>> has the same issue, though - all this means is that >>>> for >>>>> >>>>>> either >>>>> >>>>>>>>>>>>> solution, >>>>> >>>>>>>>>>>>>>>> deprecating the existing branch(...) is not on the >>>>> table. >>>>> >>>>>>>>>>>>>>>> Thanks, >>>>> >>>>>>>>>>>>>>>> Paul >>>>> >>>>>>>>>>>>>>>>> On Wed, Mar 27, 2019 at 12:08 PM Ivan Ponomarev < >>>>> >>>>>>>>> iponoma...@mail.ru <mailto:iponoma...@mail.ru>> >>>>> >>>>>>>>>>>>> wrote: >>>>> >>>>>>>>>>>>>>>>> OK, let me summarize what we have discussed up to >>>> this >>>>> >>>>>> point. >>>>> >>>>>>>>>>>>>>>>> First, it seems that it's commonly agreed that >>>>> branch API >>>>> >>>>>>> needs >>>>> >>>>>>>>>>>>>>>>> improvement. Motivation is given in the KIP. >>>>> >>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>> There are two potential ways to do it: >>>>> >>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>> 1. (as origianlly proposed) >>>>> >>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>> new KafkaStreamsBrancher<..>() >>>>> >>>>>>>>>>>>>>>>> .branch(predicate1, ks ->..) >>>>> >>>>>>>>>>>>>>>>> .branch(predicate2, ks->..) >>>>> >>>>>>>>>>>>>>>>> .defaultBranch(ks->..) //optional >>>>> >>>>>>>>>>>>>>>>> .onTopOf(stream).mapValues(...).... //onTopOf >>>>> returns >>>>> >>>>>> its >>>>> >>>>>>>>> argument >>>>> >>>>>>>>>>>>>>>>> PROS: 1) Fully backwards compatible. 2) The code >>>> won't >>>>> >>>>> make >>>>> >>>>>>>> sense >>>>> >>>>>>>>>>>>> until >>>>> >>>>>>>>>>>>>>>>> all the necessary ingredients are provided. >>>>> >>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>> CONS: The need to create a KafkaStreamsBrancher >>>>> instance >>>>> >>>>>>>>> contrasts the >>>>> >>>>>>>>>>>>>>>>> fluency of other KStream methods. >>>>> >>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>> 2. (as Paul proposes) >>>>> >>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>> stream >>>>> >>>>>>>>>>>>>>>>> .branch(predicate1, ks ->...) >>>>> >>>>>>>>>>>>>>>>> .branch(predicate2, ks->...) >>>>> >>>>>>>>>>>>>>>>> .defaultBranch(ks->...) //or noDefault(). Both >>>>> >>>>>>>>> defaultBranch(..) >>>>> >>>>>>>>>>>>> and >>>>> >>>>>>>>>>>>>>>>> noDefault() return void >>>>> >>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>> PROS: Generally follows the way KStreams interface >>>> is >>>>> >>>>>> defined. >>>>> >>>>>>>>>>>>>>>>> CONS: We need to define two terminal methods >>>>> >>>>>>>> (defaultBranch(ks->) >>>>> >>>>>>>>> and >>>>> >>>>>>>>>>>>>>>>> noDefault()). And for a user it is very easy to >>>>> miss the >>>>> >>>>>> fact >>>>> >>>>>>>>> that one >>>>> >>>>>>>>>>>>>>>>> of the terminal methods should be called. If these >>>>> methods >>>>> >>>>>> are >>>>> >>>>>>>> not >>>>> >>>>>>>>>>>>>>>>> called, we can throw an exception in runtime. >>>>> >>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>> Colleagues, what are your thoughts? Can we do >>>> better? >>>>> >>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>> Regards, >>>>> >>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>> Ivan >>>>> >>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>> 27.03.2019 18:46, Ivan Ponomarev пишет: >>>>> >>>>>>>>>>>>>>>>>> 25.03.2019 17:43, Ivan Ponomarev пишет: >>>>> >>>>>>>>>>>>>>>>>>> Paul, >>>>> >>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>> I see your point when you are talking about >>>>> >>>>>>>>>>>>>>>>>>> stream..branch..branch...default.. >>>>> >>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>> Still, I believe that this cannot not be >>>>> implemented the >>>>> >>>>>>> easy >>>>> >>>>>>>>> way. >>>>> >>>>>>>>>>>>>>>>>>> Maybe we all should think further. >>>>> >>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>> Let me comment on two of your ideas. >>>>> >>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>> user could specify a terminal method that >>>> assumes >>>>> >>>>> nothing >>>>> >>>>>>>> will >>>>> >>>>>>>>>>>>> reach >>>>> >>>>>>>>>>>>>>>>>>>> the default branch, >>>>> >>>>>>>>>>>>>>>>>>> throwing an exception if such a case occurs. >>>>> >>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>> 1) OK, apparently this should not be the only >>>> option >>>>> >>>>>> besides >>>>> >>>>>>>>>>>>>>>>>>> `default`, because there are scenarios when we >>>>> want to >>>>> >>>>>> just >>>>> >>>>>>>>> silently >>>>> >>>>>>>>>>>>>>>>>>> drop the messages that didn't match any >>>>> predicate. 2) >>>>> >>>>>>> Throwing >>>>> >>>>>>>>> an >>>>> >>>>>>>>>>>>>>>>>>> exception in the middle of data flow processing >>>>> looks >>>>> >>>>>> like a >>>>> >>>>>>>> bad >>>>> >>>>>>>>>>>>> idea. >>>>> >>>>>>>>>>>>>>>>>>> In stream processing paradigm, I would prefer to >>>>> emit a >>>>> >>>>>>>> special >>>>> >>>>>>>>>>>>>>>>>>> message to a dedicated stream. This is exactly >>>> where >>>>> >>>>>>> `default` >>>>> >>>>>>>>> can >>>>> >>>>>>>>>>>>> be >>>>> >>>>>>>>>>>>>>>>>>> used. >>>>> >>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>> it would be fairly easily for the >>>>> >>>>> InternalTopologyBuilder >>>>> >>>>>>> to >>>>> >>>>>>>>> track >>>>> >>>>>>>>>>>>>>>>>>>> dangling >>>>> >>>>>>>>>>>>>>>>>>> branches that haven't been terminated and raise >>>>> a clear >>>>> >>>>>>> error >>>>> >>>>>>>>>>>>> before it >>>>> >>>>>>>>>>>>>>>>>>> becomes an issue. >>>>> >>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>> You mean a runtime exception, when the program is >>>>> >>>>> compiled >>>>> >>>>>>> and >>>>> >>>>>>>>> run? >>>>> >>>>>>>>>>>>>>>>>>> Well, I'd prefer an API that simply won't >>>>> compile if >>>>> >>>>> used >>>>> >>>>>>>>>>>>>>>>>>> incorrectly. Can we build such an API as a >>>>> method chain >>>>> >>>>>>>> starting >>>>> >>>>>>>>>>>>> from >>>>> >>>>>>>>>>>>>>>>>>> KStream object? There is a huge cost difference >>>>> between >>>>> >>>>>>>> runtime >>>>> >>>>>>>>> and >>>>> >>>>>>>>>>>>>>>>>>> compile-time errors. Even if a failure uncovers >>>>> >>>>> instantly >>>>> >>>>>> on >>>>> >>>>>>>>> unit >>>>> >>>>>>>>>>>>>>>>>>> tests, it costs more for the project than a >>>>> compilation >>>>> >>>>>>>> failure. >>>>> >>>>>>>>>>>>>>>>>>> Regards, >>>>> >>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>> Ivan >>>>> >>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>> 25.03.2019 0:38, Paul Whalen пишет: >>>>> >>>>>>>>>>>>>>>>>>>> Ivan, >>>>> >>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>> Good point about the terminal operation being >>>>> required. >>>>> >>>>>>> But >>>>> >>>>>>>> is >>>>> >>>>>>>>>>>>> that >>>>> >>>>>>>>>>>>>>>>>>>> really >>>>> >>>>>>>>>>>>>>>>>>>> such a bad thing? If the user doesn't want a >>>>> >>>>>> defaultBranch >>>>> >>>>>>>>> they >>>>> >>>>>>>>>>>>> can >>>>> >>>>>>>>>>>>>>>>>>>> call >>>>> >>>>>>>>>>>>>>>>>>>> some other terminal method (noDefaultBranch()?) >>>>> just as >>>>> >>>>>>>>> easily. In >>>>> >>>>>>>>>>>>>>>>>>>> fact I >>>>> >>>>>>>>>>>>>>>>>>>> think it creates an opportunity for a nicer API >>>> - a >>>>> >>>>> user >>>>> >>>>>>>> could >>>>> >>>>>>>>>>>>> specify >>>>> >>>>>>>>>>>>>>>>> a >>>>> >>>>>>>>>>>>>>>>>>>> terminal method that assumes nothing will reach >>>> the >>>>> >>>>>> default >>>>> >>>>>>>>> branch, >>>>> >>>>>>>>>>>>>>>>>>>> throwing an exception if such a case occurs. >>>> That >>>>> >>>>> seems >>>>> >>>>>>> like >>>>> >>>>>>>>> an >>>>> >>>>>>>>>>>>>>>>>>>> improvement over the current branch() API, >>>>> which allows >>>>> >>>>>> for >>>>> >>>>>>>> the >>>>> >>>>>>>>>>>>> more >>>>> >>>>>>>>>>>>>>>>>>>> subtle >>>>> >>>>>>>>>>>>>>>>>>>> behavior of records unexpectedly getting >>>> dropped. >>>>> >>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>> The need for a terminal operation certainly has >>>>> to be >>>>> >>>>>> well >>>>> >>>>>>>>>>>>>>>>>>>> documented, but >>>>> >>>>>>>>>>>>>>>>>>>> it would be fairly easily for the >>>>> >>>>> InternalTopologyBuilder >>>>> >>>>>>> to >>>>> >>>>>>>>> track >>>>> >>>>>>>>>>>>>>>>>>>> dangling >>>>> >>>>>>>>>>>>>>>>>>>> branches that haven't been terminated and raise >>>>> a clear >>>>> >>>>>>> error >>>>> >>>>>>>>>>>>> before it >>>>> >>>>>>>>>>>>>>>>>>>> becomes an issue. Especially now that there is >>>> a >>>>> >>>>> "build >>>>> >>>>>>>> step" >>>>> >>>>>>>>>>>>> where >>>>> >>>>>>>>>>>>>>>>> the >>>>> >>>>>>>>>>>>>>>>>>>> topology is actually wired up, when >>>>> >>>>>> StreamsBuilder.build() >>>>> >>>>>>> is >>>>> >>>>>>>>>>>>> called. >>>>> >>>>>>>>>>>>>>>>>>>> Regarding onTopOf() returning its argument, I >>>> agree >>>>> >>>>> that >>>>> >>>>>>> it's >>>>> >>>>>>>>>>>>>>>>>>>> critical to >>>>> >>>>>>>>>>>>>>>>>>>> allow users to do other operations on the input >>>>> stream. >>>>> >>>>>>> With >>>>> >>>>>>>>> the >>>>> >>>>>>>>>>>>>>>>> fluent >>>>> >>>>>>>>>>>>>>>>>>>> solution, it ought to work the same way all >>>> other >>>>> >>>>>>> operations >>>>> >>>>>>>>> do - >>>>> >>>>>>>>>>>>> if >>>>> >>>>>>>>>>>>>>>>> you >>>>> >>>>>>>>>>>>>>>>>>>> want to process off the original KStream >>>> multiple >>>>> >>>>> times, >>>>> >>>>>>> you >>>>> >>>>>>>>> just >>>>> >>>>>>>>>>>>>>>>>>>> need the >>>>> >>>>>>>>>>>>>>>>>>>> stream as a variable so you can call as many >>>>> operations >>>>> >>>>>> on >>>>> >>>>>>> it >>>>> >>>>>>>>> as >>>>> >>>>>>>>>>>>> you >>>>> >>>>>>>>>>>>>>>>>>>> desire. >>>>> >>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>> Thoughts? >>>>> >>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>> Best, >>>>> >>>>>>>>>>>>>>>>>>>> Paul >>>>> >>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>> On Sun, Mar 24, 2019 at 2:02 PM Ivan Ponomarev < >>>>> >>>>>>>>> iponoma...@mail.ru <mailto:iponoma...@mail.ru> >>>>> >>>>>>>>>>>>>>>>>>>> wrote: >>>>> >>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>> Hello Paul, >>>>> >>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>> I afraid this won't work because we do not >>>>> always need >>>>> >>>>>> the >>>>> >>>>>>>>>>>>>>>>>>>>> defaultBranch. And without a terminal >>>> operation we >>>>> >>>>> don't >>>>> >>>>>>>> know >>>>> >>>>>>>>>>>>> when to >>>>> >>>>>>>>>>>>>>>>>>>>> finalize and build the 'branch switch'. >>>>> >>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>> In my proposal, onTopOf returns its argument, >>>>> so we >>>>> >>>>> can >>>>> >>>>>> do >>>>> >>>>>>>>>>>>> something >>>>> >>>>>>>>>>>>>>>>>>>>> more with the original branch after branching. >>>>> >>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>> I understand your point that the need of >>>> special >>>>> >>>>> object >>>>> >>>>>>>>>>>>> construction >>>>> >>>>>>>>>>>>>>>>>>>>> contrasts the fluency of most KStream methods. >>>> But >>>>> >>>>> here >>>>> >>>>>> we >>>>> >>>>>>>>> have a >>>>> >>>>>>>>>>>>>>>>>>>>> special case: we build the switch to split the >>>>> flow, >>>>> >>>>> so >>>>> >>>>>> I >>>>> >>>>>>>>> think >>>>> >>>>>>>>>>>>> this >>>>> >>>>>>>>>>>>>>>>> is >>>>> >>>>>>>>>>>>>>>>>>>>> still idiomatic. >>>>> >>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>> Regards, >>>>> >>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>> Ivan >>>>> >>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>> 24.03.2019 4:02, Paul Whalen пишет: >>>>> >>>>>>>>>>>>>>>>>>>>>> Ivan, >>>>> >>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>> I think it's a great idea to improve this >>>>> API, but I >>>>> >>>>>> find >>>>> >>>>>>>> the >>>>> >>>>>>>>>>>>>>>>>>>>>> onTopOff() >>>>> >>>>>>>>>>>>>>>>>>>>>> mechanism a little confusing since it >>>>> contrasts the >>>>> >>>>>>> fluency >>>>> >>>>>>>>> of >>>>> >>>>>>>>>>>>> other >>>>> >>>>>>>>>>>>>>>>>>>>>> KStream method calls. Ideally I'd like to >>>>> just call >>>>> >>>>> a >>>>> >>>>>>>>> method on >>>>> >>>>>>>>>>>>> the >>>>> >>>>>>>>>>>>>>>>>>>>> stream >>>>> >>>>>>>>>>>>>>>>>>>>>> so it still reads top to bottom if the branch >>>>> cases >>>>> >>>>> are >>>>> >>>>>>>>> defined >>>>> >>>>>>>>>>>>>>>>>>>>>> fluently. >>>>> >>>>>>>>>>>>>>>>>>>>>> I think the addBranch(predicate, handleCase) >>>>> is very >>>>> >>>>>> nice >>>>> >>>>>>>>> and the >>>>> >>>>>>>>>>>>>>>>>>>>>> right >>>>> >>>>>>>>>>>>>>>>>>>>> way >>>>> >>>>>>>>>>>>>>>>>>>>>> to do things, but what if we flipped around >>>>> how we >>>>> >>>>>>> specify >>>>> >>>>>>>>> the >>>>> >>>>>>>>>>>>> source >>>>> >>>>>>>>>>>>>>>>>>>>>> stream. >>>>> >>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>> Like: >>>>> >>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>> stream.branch() >>>>> >>>>>>>>>>>>>>>>>>>>>> .addBranch(predicate1, >>>> this::handle1) >>>>> >>>>>>>>>>>>>>>>>>>>>> .addBranch(predicate2, >>>> this::handle2) >>>>> >>>>>>>>>>>>>>>>>>>>>> .defaultBranch(this::handleDefault); >>>>> >>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>> Where branch() returns a KBranchedStreams or >>>>> >>>>>>>> KStreamBrancher >>>>> >>>>>>>>> or >>>>> >>>>>>>>>>>>>>>>>>>>> something, >>>>> >>>>>>>>>>>>>>>>>>>>>> which is added to by addBranch() and >>>>> terminated by >>>>> >>>>>>>>>>>>> defaultBranch() >>>>> >>>>>>>>>>>>>>>>>>>>>> (which >>>>> >>>>>>>>>>>>>>>>>>>>>> returns void). This is obviously >>>>> incompatible with >>>>> >>>>> the >>>>> >>>>>>>>> current >>>>> >>>>>>>>>>>>>>>>>>>>>> API, so >>>>> >>>>>>>>>>>>>>>>>>>>> the >>>>> >>>>>>>>>>>>>>>>>>>>>> new stream.branch() would have to have a >>>>> different >>>>> >>>>>> name, >>>>> >>>>>>>> but >>>>> >>>>>>>>> that >>>>> >>>>>>>>>>>>>>>>>>>>>> seems >>>>> >>>>>>>>>>>>>>>>>>>>>> like a fairly small problem - we could call it >>>>> >>>>>> something >>>>> >>>>>>>> like >>>>> >>>>>>>>>>>>>>>>>>>>>> branched() >>>>> >>>>>>>>>>>>>>>>>>>>> or >>>>> >>>>>>>>>>>>>>>>>>>>>> branchedStreams() and deprecate the old API. >>>>> >>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>> Does this satisfy the motivations of your >>>>> KIP? It >>>>> >>>>>> seems >>>>> >>>>>>>>> like it >>>>> >>>>>>>>>>>>>>>>>>>>>> does to >>>>> >>>>>>>>>>>>>>>>>>>>>> me, allowing for clear in-line branching >>>>> while also >>>>> >>>>>>>> allowing >>>>> >>>>>>>>> you >>>>> >>>>>>>>>>>>> to >>>>> >>>>>>>>>>>>>>>>>>>>>> dynamically build of branches off of >>>>> KBranchedStreams >>>>> >>>>>> if >>>>> >>>>>>>>> desired. >>>>> >>>>>>>>>>>>>>>>>>>>>> Thanks, >>>>> >>>>>>>>>>>>>>>>>>>>>> Paul >>>>> >>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>> On Sat, Mar 23, 2019 at 4:28 PM Ivan Ponomarev >>>>> >>>>>>>>>>>>>>>>>>>>> <iponoma...@mail.ru.invalid> >>>>> >>>>>>>>>>>>>>>>>>>>>> wrote: >>>>> >>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>> Hi Bill, >>>>> >>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>> Thank you for your reply! >>>>> >>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>> This is how I usually do it: >>>>> >>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>> void handleFirstCase(KStream<String, String> >>>>> ks){ >>>>> >>>>>>>>>>>>>>>>>>>>>>> ks.filter(....).mapValues(...) >>>>> >>>>>>>>>>>>>>>>>>>>>>> } >>>>> >>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>> void handleSecondCase(KStream<String, >>>>> String> ks){ >>>>> >>>>>>>>>>>>>>>>>>>>>>> ks.selectKey(...).groupByKey()... >>>>> >>>>>>>>>>>>>>>>>>>>>>> } >>>>> >>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>> ...... >>>>> >>>>>>>>>>>>>>>>>>>>>>> new KafkaStreamsBrancher<String, String>() >>>>> >>>>>>>>>>>>>>>>>>>>>>> .addBranch(predicate1, >>>>> this::handleFirstCase) >>>>> >>>>>>>>>>>>>>>>>>>>>>> .addBranch(predicate2, >>>>> this::handleSecondCase) >>>>> >>>>>>>>>>>>>>>>>>>>>>> .onTopOf(....) >>>>> >>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>> Regards, >>>>> >>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>> Ivan >>>>> >>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>> 22.03.2019 1:34, Bill Bejeck пишет: >>>>> >>>>>>>>>>>>>>>>>>>>>>>> Hi Ivan, >>>>> >>>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>>> Thanks for the KIP. >>>>> >>>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>>> I have one question, the >>>> KafkaStreamsBrancher >>>>> >>>>> takes a >>>>> >>>>>>>>> Consumer >>>>> >>>>>>>>>>>>> as a >>>>> >>>>>>>>>>>>>>>>>>>>>>> second >>>>> >>>>>>>>>>>>>>>>>>>>>>>> argument which returns nothing, and the >>>>> example in >>>>> >>>>>> the >>>>> >>>>>>>> KIP >>>>> >>>>>>>>>>>>> shows >>>>> >>>>>>>>>>>>>>>>>>>>>>>> each >>>>> >>>>>>>>>>>>>>>>>>>>>>>> stream from the branch using a terminal node >>>>> >>>>>>>>> (KafkaStreams#to() >>>>> >>>>>>>>>>>>>>>>>>>>>>>> in this >>>>> >>>>>>>>>>>>>>>>>>>>>>>> case). >>>>> >>>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>>> Maybe I've missed something, but how would >>>> we >>>>> >>>>> handle >>>>> >>>>>>> the >>>>> >>>>>>>>> case >>>>> >>>>>>>>>>>>>>>>>>>>>>>> where the >>>>> >>>>>>>>>>>>>>>>>>>>>>>> user has created a branch but wants to >>>> continue >>>>> >>>>>>>> processing >>>>> >>>>>>>>> and >>>>> >>>>>>>>>>>>> not >>>>> >>>>>>>>>>>>>>>>>>>>>>>> necessarily use a terminal node on the >>>> branched >>>>> >>>>>> stream >>>>> >>>>>>>>>>>>> immediately? >>>>> >>>>>>>>>>>>>>>>>>>>>>>> For example, using today's logic as is if >>>>> we had >>>>> >>>>>>>> something >>>>> >>>>>>>>> like >>>>> >>>>>>>>>>>>>>>>>>>>>>>> this: >>>>> >>>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>>> KStream<String, String>[] branches = >>>>> >>>>>>>>>>>>>>>>>>>>>>>> originalStream.branch(predicate1, >>>>> >>>>>>>>>>>>>>>>>>>>>>>> predicate2); >>>>> >>>>>>>>>>>>>>>>>>>>>>>> branches[0].filter(....).mapValues(...).. >>>>> >>>>>>>>>>>>>>>>>>>>>>>> branches[1].selectKey(...).groupByKey()..... >>>>> >>>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>>> Thanks! >>>>> >>>>>>>>>>>>>>>>>>>>>>>> Bill >>>>> >>>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>>> On Thu, Mar 21, 2019 at 6:15 PM Bill Bejeck >>>> < >>>>> >>>>>>>>> bbej...@gmail.com <mailto:bbej...@gmail.com> >>>>> >>>>>>>>>>>>>>>>>>>>>>>> wrote: >>>>> >>>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>>>> All, >>>>> >>>>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>>>> I'd like to jump-start the discussion for >>>> KIP- >>>>> >>>>> 418. >>>>> >>>>>>>>>>>>>>>>>>>>>>>>> Here's the original message: >>>>> >>>>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>>>> Hello, >>>>> >>>>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>>>> I'd like to start a discussion about >>>> KIP-418. >>>>> >>>>> Please >>>>> >>>>>>>> take >>>>> >>>>>>>>> a >>>>> >>>>>>>>>>>>> look >>>>> >>>>>>>>>>>>>>>>> at >>>>> >>>>>>>>>>>>>>>>>>>>> the >>>>> >>>>>>>>>>>>>>>>>>>>>>>>> KIP if you can, I would appreciate any >>>>> feedback :) >>>>> >>>>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>>>> KIP-418: >>>>> >>>>> >>>>> >>>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-418%3A+A+method-chaining+way+to+branch+KStream >>>>> >>>>>>>>>>>>>>>>>>>>>>>>> JIRA KAFKA-5488: >>>>> >>>>>>>>>>>>> https://issues.apache.org/jira/browse/KAFKA-5488 >>>>> >>>>>>>>>>>>>>>>>>>>>>>>> PR#6164: >>>>> >>>>> https://github.com/apache/kafka/pull/6164 >>>>> >>>>>>>>>>>>>>>>>>>>>>>>> Regards, >>>>> >>>>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>>>> Ivan Ponomarev >>>>> >>>>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>>>>>>>>>>>>>>>>>> >>>>> >>>>>>>>> >>>>> > >>>>> >>>> >
signature.asc
Description: OpenPGP digital signature