ContextualProcessor is a Kafka Streams base class for record-processing nodes. Subclasses focus on process(), while the class initializes ProcessorContext and supplies lifecycle-method overrides. section 10.1.2
Use ContextualProcessor to:
- Override process() to transform incoming records and forward results to selected child nodes. section 10.1.2
- Use context() to access ProcessorContext after the topology initializes the processor. section 10.1.2
- Override init() when setup requires a state store or scheduled processing; otherwise, rely on inherited initialization. section 10.2.1
Currency conversion and routing
A BeerPurchaseProcessor extends ContextualProcessor<String, BeerPurchase, String, BeerPurchase>. Its process() method converts euro or pound amounts to dollars, then forwards international sales to an international-sales topic and domestic sales to a domestic-sales topic. The topology attaches this processor to a source node with addProcessor(). section 10.1.2
Listing 10.3 BeerPurchaseProcessor
public class BeerPurchaseProcessor extends
ContextualProcessor<String, BeerPurchase, String, BeerPurchase> {
private final String domesticSalesNode;
private final String internationalSalesNode;
private final Map<String, Double> conversionRates;
public BeerPurchaseProcessor(String domesticSalesNode,
String internationalSalesNode,
Map<String,Double> conversionRates) {
this.domesticSalesNode = domesticSalesNode; #1
this.internationalSalesNode = internationalSalesNode; #1
this.conversionRates = conversionRates; #1
} #1
@Override #1
public void process(
Record<String, BeerPurchase> beerPurchaseRecord) { #2
BeerPurchase beerPurchase = beerPurchaseRecord.value();
String key = beerPurchaseRecord.key();
BeerPurchase.Currency transactionCurrency =
beerPurchase.getCurrency();
if (transactionCurrency != BeerPurchase.Currency.DOLLAR) {
BeerPurchase.Builder builder =
BeerPurchase.newBuilder(beerPurchase);
double internationalSaleAmount = beerPurchase.getTotalSale();
String pattern = "###.##";
DecimalFormat decimalFormat = new DecimalFormat(pattern);
builder.setCurrency(BeerPurchase.Currency.DOLLAR);
builder.setTotalSale(Double.parseDouble(decimalFormat.format(
convertToDollars(transactionCurrency.name(),
internationalSaleAmount)))); #3
Record<String, BeerPurchase> convertedBeerPurchaseRecord =
new Record<>(key,builder.build(),
beerPurchaseRecord.timestamp());
context().forward(convertedBeerPurchaseRecord, #4
internationalSalesNode);
} else {
context().forward(
beerPurchaseRecord,
domesticSalesNode); #5
}
}
Inspect the processor implementation to see constructor configuration, generic types, process(), context(), and forwarding logic together. section 10.1.2
Listing 10.7 init() method tasks
@Override
public void init(ProcessorContext<String, StockPerformance> context) {
super.init(processorContext); #1
keyValueStore = context().getStateStore(stateStoreName); #2
StockPerformancePunctuator punctuator =
new StockPerformancePunctuator(differentialThreshold,
context(),
keyValueStore); #3
context().schedule(10000,
PunctuationType.STREAM_TIME,
punctuator); #4
}
}
Compare the minimal processor with one that calls super.init(), retrieves a state store, and schedules a stream-time punctuator. section 10.2.1
