The process method is the Processor API entry point for handling each incoming record. In a stock-performance application, it retrieves a ticker’s state, creates missing state, updates price and volume statistics, and stores the result. section 10.2.3
A typical processing cycle:
- Read the StockPerformance object from the key-value state store using the stock ticker. section 10.2.3
- Create the object when no stored state exists, then add the transaction’s share price and volume. section 10.2.3
- Maintain rolling statistics after 20 trades by removing the oldest value and adding the newest rather than recalculating everything. section 10.2.3
Figure 10.7 Stock analysis process() method walk-through
Follow the stock-analysis walk-through from incoming trades through state updates and rolling calculations. section 10.2.3
KStream.process() places a custom Processor API component inside a Kafka Streams DSL topology. This keeps DSL operators for most of the application while reserving lower-level control for selected logic. section 10.4
Listing 10.14 Example of mixing the DSL and the Processor API
StreamsBuilder builder = new StreamsBuilder();
StoreBuilder<KeyValueStore<String, StockPerformance>> storeBuilder =
Stores.keyValueStoreBuilder(storeSupplier,
Serdes.String(),
stockPerformanceSerde);
builder.stream(INPUT_TOPIC, #1
Consumed.with(stringSerde, stockTransactionSerde))
.process(
new StockPerformanceProcessorSupplier(
storeBuilder)) #2
.peek(printKV("StockPerformance"))
.to(OUTPUT_TOPIC,
Produced.with(stringSerde, stockPerformanceSerde));
Inspect the mixed topology: read the input topic, inject StockPerformanceProcessor, inspect results, and write the output topic. section 10.4
