BranchedKStream.branch() creates one stream branch by pairing a boolean Predicate with a Branched processing definition. A matching record enters that branch’s processor, while the parent stream retains its original key and value types. section 6.6.2
Build a split topology in this sequence:
- Call KStream.split() to obtain a BranchedKStream. section 6.6.2
- Call branch(predicate, branched) for each condition and destination or transformation. section 6.6.2
- Finish with defaultBranch() to process nonmatching records, or noDefaultBranch() to drop them. section 6.6.2
- Use the returned map of KStream objects when Branched definitions expose newly created branches. section 6.6.2
Route purchase departments
Predicate<String, Purchase> isCoffee = (key, purchase) -> purchase.getDepartment().equalsIgnoreCase("coffee"); purchaseKStream.split() .branch(isCoffee, Branched.withConsumer(stream -> stream.to("coffee-topic"))) .branch(isElectronics, Branched.withConsumer(stream -> stream.to("electronics"))) .defaultBranch(Branched.withConsumer(stream -> stream.to("purchases"))); Coffee and electronics purchases leave through dedicated topics; all other purchases continue through the default topic. section 6.6.2
Listing 6.17 Splitting the stream
Predicate<String, Purchase> isCoffee =
(key, purchase) ->
purchase.getDepartment().equalsIgnoreCase("coffee"); #1
Predicate<String, Purchase> isElectronics =
(key, purchase) ->
purchase.getDepartment().equalsIgnoreCase("electronics");
purchaseKStream.split() #2
.branch(isCoffee,
Branched.withConsumer(coffeeStream -> coffeeStream.to("coffee-topic"))) #3
.branch(isElectronics,
Branched.withConsumer(electronicStream ->
electronicStream.to("electronics")) #4
.defaultBranch(Branched.withConsumer(retailStream ->
retailStream.to("purchases")); #5
Inspect the complete purchase-routing listing showing predicates, Branched.withConsumer(), and defaultBranch(). section 6.6.2
