KAFKA-5488: Add type-safe split() operator

[inponomarev]

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[inponomarev]

⚠️ Two differences with KIP specification, discussion needed⚠️

1. Instead of multiple overloaded variants of Branched.with we now have Branched.withFunction and Branched.withConsumer. This is because of compiler warnings about overloading (Function and Consumer being indistinguishable when supplied as lambdas)

2. 'Fully covariant' signatures like Consumer<? super KStream<? super K, ? super V>> don't work as expected. Used Consumer<? super KStream<K, V>> instead

[vvcephei]

Hey @mjsax , do you have time to give this a first pass?

[mjsax]

I'll put it into my backlog. But I am the main reviewer for two other KIPs (216 and 466) that I should review first as they got approve earlier and PRs are open for longer already.

[mjsax]

Thanks for the PR @inponomarev and sorry for the long wait for a review

Many comments are about JavaDocs, so it's mostly small suggestions. A few comments about the code structure are there, too.

[mjsax]

nit supposed -> used ?

this function -> the provided function

[mjsax]

Actually, I am wondering if we should allow to pass in null? Thoughts?

[inponomarev]

See my reply below, where we discuss null consumers: #9107 (comment)

(in short: I agree, I think we shouldn't)

[mjsax]

If a non-null branch is provided here? (branch -> consumer?)

But I would propose to simplify it, and just use By default (as passing in a non-null consumer should be the "default" usage).

[mjsax]

As above, should we even allow a not-null consumer?

[inponomarev]

see #9107 (comment)

[mjsax]

Should we call branch((k,v) -> true, branched) instead to just add a predicate and branch? This way, the default branch is nothing special at runtime any longer.

[inponomarev]

The default branch should have index 0 (so it will be stable when branches are added or removed), but it should always be checked after all other branches. And when we come to the default branch during message processing, there is actually no need in dereferncing a predicate and calling test... that's why I treat the default branch differently.

[mjsax]

I guess it's fine both ways. -- The point about the index is a good one that I missed. But would still be doable I guess.

I don't think that there would be any measurable runtime difference if you use a "default predicate" (what we also do in the current implementation) -- the code is just a little "cleaner" as we don't need an extra "if" at the end -- but it's also not the end of the world as the process method is fairly simply anyway.

[mjsax]

result -> outputBranches ?

[mjsax]

I am wondering if it might be better to move this code into a build method that would be called within defaultBranch() / noDefaultBranch() ?

The pattern to pass in empty list that we modify later seems undesirable, and we should first build the list, and than pass them in -- otherwise, we make assumptions how ProcessorParameters and ProcessorGraphNode might be implemented what we should avoid.

[inponomarev]

I clearly remember that something made me to write it this way, but I have to recall...

[mjsax]

Would love to learn about it. -- In general, it's easier to follow the same pattern throughout the code base. It easier to reason about the code that way, and also easier for people to learn the code base.

[mjsax]

Just saw your other comment: #9107 (comment)

[inponomarev]

Hi @mjsax, thanks for your thorough revew! I have fixed everything according to your comments, except:

• KAFKA-5488: Add type-safe split() operator #9107 (comment) -- just want to be sure!
• KAFKA-5488: Add type-safe split() operator #9107 (comment) -- see my explanation
• KAFKA-5488: Add type-safe split() operator #9107 (comment) -- I agree that this is ugly but there was a good reason why I made it this way. Just give me a day and I'll return with comments (or will fix it).

[inponomarev]

OK @mjsax concerning #9107 (comment) I remembered why it was implemented this way!

The problem is that it is not necessary to invoke defaultBranch() / noDefaultBranch() when we use consumers, like in this simple example (I just added a new unit test for this case):

source.split()
    .branch(isCoffee, Branched.withConsumer(issuer::setCoffeePurchases))
    .branch(isElectronics, Branched.withConsumer(issuer::setElectronicsPurchases));

[mjsax]

About the original comment: #9107 (comment)

I am fine with those changes.

About #9107 (comment) -- that is a good point. Thanks for explaining. I guess it's a "philosophical" question if we want to allow this pattern though, or if we want to require that either defaultBranch() or noDefaultBranch() is called? -- I did consider calling branch() like a builder pattern, and the final [noD|d]efaultBranch call is basically build()?

Curious to hear what @vvcephei thinks about it.

[mjsax]

@inponomarev -- Can you also update the docs for Kafka Streams and the 2.8 upgrade guide in this PR.

[inponomarev]

Can you also update the docs for Kafka Streams and the 2.8 upgrade guide in this PR.

The documentation had been already updated (see changes in docs/streams/developer-guide/dsl-api.html)

I also modified docs/upgrade.html -- should I add something more here, like code examples?

Another question: CI checks fail because of usage of deprecated branch method in streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KStream.scala. Since I'm not a Scala user, I have no idea of what should be done here.

Most likely we should deprecate the branch method and add a wrapper for the new split method, but I don't know how to do this correctly.

[mjsax]

To make the build pass, for now, it should be sufficient to just deprecate the method via @nowarn("cat=deprecation") -- But it seems we should update the Scala API, too. If you cannot handle it, we can do a follow up PR.

It seems, we need to add split() to KStream.scala and introduce a new BranchedKStream.scala and Branch.scala classes and maybe some translations from Java Consumer/Function to their Scala variants. But I also don't really now Scala; @vvcephei should nkow better.

[inponomarev]

As far as I can judge from the name, @nowarn is not for deprecation, but rather for a warning suppression 🤔 apparently we need to mirror the changes in Java KStream interface here. Never wrote anything in Scala before. OK, it's better to wait for @vvcephei !

[mjsax]

Maybe I miss-understood you question. I thought the build fails because we are using some deprecated method -- for this case, we can make the build pass by suppressing the warning. If you want to deprecate a method in the Scala API, you just add @deprecated similar to Java. -- I guess it makes sense to also deprecate the KStream.scala#branch() method, but suppressing the warning should also make the build pass and we can deprecate this method when we add the new split() method.

[vvcephei]

Hey @inponomarev and @mjsax ! I'm glad to see this is moving along.

Regarding #9107 (comment) :

My understanding was that defaultBranch/noDefaultBranch were the terminal operators, in that they close out the context of a BranchedKStream, and you can't add any more branches after one of those methods.

But also, the whole branching construct is an incremental builder like the rest of the Kafka Streams API. In other words, just like this is a valid program:

builder.stream("input")
       .filter(myPredicate)

so would be Ivan's example:

builder.split()
       .branch("myBranch", ...)

What I mean by "incremental builder" is that each time you call a chained method in the DSL, it immediately adds nodes to the program, as opposed to having to call any kind of build() method to actually add stuff to the program. I think there are pros and cons to this design, but it seems more in line with the rest of the DSL not to require the terminal operators.

[inponomarev]

Hi @vvcephei , thank you for your comment. There's another question that we were unable to solve without you -- see #9107 (comment) from the words 'CI checks fail' and further discussion. Can you clarify, what's expected from KStream.scala ?

[mjsax]

@vvcephei -- hope you are also ok with the proposed changes to the KIP as per the PR description on top: #9107 (comment)

[inponomarev]

@vvcephei @mjsax I added full Scala wrapper for the new API: split method, BranchedKStream and Branched. Also added Scala unit tests that verify main use cases

[vvcephei]

Hey @inponomarev , I just took a look at the Scala API. Thanks for adding that!

I figured it'd be just easier to push a few tweaks than to describe what needs to be done.

• You asked me offline if we could avoid the overloads in Branched, and indeed, we can with a default argument of null for the name.
• The Scala test was inadvertently using the Java Branched class, but you meant to test the Scala one.
• I happened to notice a small typo: The opposite of "prefix" is "suffix", not "postfix"
• I also noticed that your files all had CRLF (windows) return characters, so I fixed them. You might want to configure git for autocrlf (git config --global core.autocrlf true) (see https://www.git-scm.com/book/en/v2/Customizing-Git-Git-Configuration)

These are all separate commits above, so you can scrutinize each one. This PR is your work, so feel free to protest any of my suggestions.

[inponomarev]

Hi @vvcephei thank you for your commits! Is everything else OK, especially #9107 (comment)?

@mjsax I pushed small fixes to Javadoc/Scaladoc, and AFAICS only tests not related to the changes are failing.

[vvcephei]

Thanks @inponomarev ,

Ah, I didn't notice that method signature name. I actually prefer it this way :)

Thanks also for pointing out the covariance change. This is also fine. Java's type system only contains a partial implementation of variance, so we do best we can.

Did you already update the KIP? If not, please do.

I'm +1 on this PR.

[inponomarev]

Thank you @vvcephei, I have updated the KIP and now it reflects the actual implementation.

I just wasn't sure if it's ok to edit specification text after it has been formally approved :-)

[inponomarev]

Hi @mjsax , I have rebased and manually merged conflicts, and also removed FunctionConverters

JDK8 build still fails, but this time much later -- something related to integration testing

[mjsax]

Wait failure do you see exactly? Seem Jenkins in still running.

[inponomarev]

I was talking about build 17 (triggered by Commit db573f5, see https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-9107/)

Where did build 18 come from, why did it take 8 hours and then timed out -- I can't understand 😃

[mjsax]

Ah I see -- well, we do have from flaky tests, so nothing to worry about I guess. The last run timed out, so I retriggered the build. However, I could build it locally with Java8/Scale 2.12 and so I guess we can merge. Just waiting for @vvcephei to take a quick look at the last Scala commit.

[vvcephei]

Thanks, all, that Scala fix looks perfect to me.

[mjsax]

Merged to trunk.

Congrats for getting this into the 2.8.0 release @inponomarev -- great work!