A predicate is a condition that evaluates a record and returns true or false. In Kafka Streams, it decides whether a record continues through a filter, enters a branch, or participates in data-driven aggregation. section 6.7 section 10.3
Pass a predicate to KStream.filter to retain matching records; records that fail the test are dropped. filterNot reverses that choice, forwarding only records that do not match. A filter predicate receives a key and value through its test method, and can inspect only the value when the key is irrelevant. section 6.6.1 section 6.7
Use predicates as logical gates when dividing a stream:
- Call split(), then add a branch with a predicate and its processing logic. A true result lets the record flow into that branch. section 6.6.2
- Finish with defaultBranch() to handle records that matched no predicate, or noDefaultBranch() to drop them. section 6.6.2
- Keep branch key and value types unchanged from the parent stream; branches can then process records independently. section 6.6.2
Route purchase departments
Define isCoffee and isElectronics predicates from a purchase department, then split the purchase stream. Send each matching branch to its topic and route all remaining purchases through the default branch. section 6.6.2
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
This filter lambda totals purchased-item quantities times prices and retains only purchases above $10 before mapping them to reward accumulators. section 6.6.1
Predicates also support custom Processor API behavior. Separate predicates can decide when a sensor record starts or joins an aggregation and when a record rejects it, closing the aggregation decision path. section 10.3
