ProcessorSupplier is the factory boundary between a Kafka Streams topology and its processor instances. Topology.addProcessor receives one, then uses the supplied processor to handle records from a named parent node. Because it is a single-method interface, a lambda is sufficient when no state-store wiring is required. section 10.1.2
Use a concrete implementation when the processor needs topology-managed state:
- Implement get() to return a new processor instance, commonly passing the state-store name into its constructor. section 10.2.1
- Implement stores() to return the StoreBuilder instances associated with the processor; Kafka Streams then wires those stores to it. section 10.2.1
- Connect the processor to its parent by passing the source or upstream node name to addProcessor. section 10.1.2
Stock-performance processor
A supplier receives a StoreBuilder, returns a StockPerformanceProcessor named with storeBuilder.name(), and exposes the builder through stores(). The processor later retrieves that store from ProcessorContext during init(), then schedules punctuation for periodic processing. section 10.2.1
Listing 10.6 ProcessorSupplier implementation
public class StockPerformanceProcessorSupplier
implements ProcessorSupplier<String, Transaction,
String, StockPerformance> {
StoreBuilder<?> storeBuilder;
public StockPerformanceProcessorSupplier(StoreBuilder<?> storeBuilder) {
this.storeBuilder = storeBuilder;
}
@Override
public Processor<String, Transaction, String, StockPerformance> get() {
return new StockPerformanceProcessor(storeBuilder.name()); #1
}
@Override
public Set<StoreBuilder<?>> stores() {
return Collections.singleton(storeBuilder); #2
}
}
Inspect the complete supplier shape, including generic input and output types, get(), and stores(). section 10.2.1
