ValueJoiner combines the matching values from two Kafka Streams records and returns the value emitted by a join. Its apply method can accept different value types and produce a third type. section 7.3.1 section 7.3.3
Declare the input and result types with ValueJoiner<V1, V2, R>, then implement apply. The join supplies both values for the matching key; the join operation separately defines timing, windows, serialization, and state. section 7.3.1 section 7.3.3
Calculating promotion points
A PromotionJoiner receives a CoffeePurchase and a RetailPurchase, sums coffee spending with purchased-item totals, adds 50 points when store spending exceeds $50, and returns a Promotion containing customer, drink, and purchase information. section 7.3.3
Use the joiner with these practical constraints:
- Use a lambda for simple logic or a concrete implementation when separate unit testing is useful. section 7.3.3
- Use ValueJoinerWithKey when result calculation needs the key; treat that key as read-only. section 7.3.3
Listing 7.14 ValueJoiner implementation
public class PromotionJoiner
implements ValueJoiner<CoffeePurchase,
RetailPurchase,
Promotion> {
@Override
public Promotion apply(
CoffeePurchase coffeePurchase,
RetailPurchase retailPurchase) {
double coffeeSpend = coffeePurchase.getPrice(); #1
double storeSpend = retailPurchase.getPurchasedItemsList() #2
.stream()
.mapToDouble(pi -> pi.getPrice() * pi.getQuantity()).sum();
double promotionPoints = coffeeSpend + storeSpend; #3
if (storeSpend > 50.00) {
promotionPoints += 50.00;
}
return Promotion.newBuilder() #4
.setCustomerId(retailPurchase.getCustomerId())
.setDrink(coffeePurchase.getDrink())
.setItemsPurchased(retailPurchase.getPurchasedItemsCount())
.setPoints(promotionPoints).build();
}
Inspect the complete Java implementation of PromotionJoiner. section 7.3.3
