Filtering selectively processes records in a Kafka Streams topology. KStream#filter drops records whose predicate returns false, allowing downstream operators to handle only qualifying events. section 6.6 section 6.7
Use filtering according to the record type and desired topology behavior:
- Place KStream.filter before downstream operations such as mapValues; its predicate receives the record key and value. section 6.6.1
- Use filterNot when records failing the predicate should continue instead. section 6.6.1
- On a KTable, nonmatching records produce tombstone markers rather than simply disappearing, while null input values retain delete semantics. section 8.3
Reward purchases over $10
A retailer can filter purchases before mapping them into reward objects. The predicate sums each purchased item's quantity multiplied by price and retains the record only when the total exceeds 10.00. section 6.6 section 6.6.1
Listing 6.16 Adding a filter to drop purchases not meeting rewards criteria
KStream<String, RewardAccumulator> rewardsKStream = #1
retailPurchaseKStream.filter((key, value) -> #2
value.getPurchasedItemsList().stream()
.mapToDouble((item -> item.getQuantity()
* item.getPrice()))
.sum() > 10.00)
.mapValues(rewardObjectMapper); #3
Inspect the Java topology fragment that filters qualifying purchases before applying mapValues. section 6.6.1
