StoreBuilder constructs a state store for the Kafka Streams Processor API and lets a topology connect that store to processors. Unlike DSL usage with StoreSuppliers and Materialized, the builder is supplied directly when assembling the topology. section 7.4.6
A typical Processor API setup follows these steps:
- Create a store supplier, such as an in-memory key-value store. section 10.2.1
- Create a typed StoreBuilder with the supplier and key/value Serdes. section 10.2.1
- Pass the builder to a ProcessorSupplier; its stores method returns the builder for automatic processor wiring. section 10.2.1
- Retrieve the resulting state store from ProcessorContext during processor initialization. section 10.2.1
Stock-performance topology
A stock-performance application creates an in-memory store, builds a KeyValueStore<String, StockPerformance>, passes the StoreBuilder into StockPerformanceProcessorSupplier, and retrieves the store in init(). The processor then schedules stream-time punctuation to use the stored state. section 10.2.1
Listing 10.5 Stock-performance application with a custom processor
Topology topology = new Topology();
String stocksStateStore = "stock-performance-store";
double differentialThreshold = 0.02; #1
KeyValueBytesStoreSupplier storeSupplier =
Stores.inMemoryKeyValueStore(stocksStateStore); #2
StoreBuilder<KeyValueStore<String, StockPerformance>> storeBuilder
= Stores.keyValueStoreBuilder(
storeSupplier, Serdes.String(), stockPerformanceSerde); #3
topology.addSource("stocks-source",
stringDeserializer,
stockTransactionDeserializer,
"stock-transactions")
.addProcessor("stocks-processor", #4
new StockPerformanceProcessorSupplier(storeBuilder),
"stocks-source")
.addSink("stocks-sink",
"stock-performance",
stringSerializer,
stockPerformanceSerializer,
"stocks-processor"); #5
Shows the StoreBuilder declaration alongside the source, processor, and sink connections in a complete topology. section 10.2.1
