How to add a custom StateStore to the Kafka Streams DSL processor?

apache-kafka-streams

Solution

You need to register you store outside of you processor using `StreamsBuilder` (or `KStreamBuilder` in older releases). First you create the store, than you registers it to `StreamsBuilder` (`KStreamBuilder`), and when you add the processor you provide the store name to connect the processor and the store.

StreamsBuilder builder = new StreamsBuilder();

// create store
StoreBuilder storeBuilder = Stores.keyValueStoreBuilder(
    Stores.persistentKeyValueStore("invStore"),
    Serdes.String(),
    invSerde));
// register store
builder.addStateStore(storeBuilder);

KStream<String, String> source = builder.stream(source_stream);
source.selectKey(new MyKeyValueMapper())
        .filterNot((k,v) -> k.equals("UnknownGroup"))
        .process(() -> new MyProcessor(), "invStore"); // connect store to processor by providing store name


// older API:

KStreamBuilder builder = new KStreamBuilder();

// create store
StateStoreSupplier storeSupplier = (KeyValueStore)Stores.create("invStore")
    .withKeys(Serdes.String())
    .withValues(invSerde)
    .persistent()
    .build();
// register store
builder.addStateStore(storeSupplier);

KStream<String, String> source = builder.stream(source_stream);
source.selectKey(new MyKeyValueMapper())
        .filterNot((k,v) -> k.equals("UnknownGroup"))
        .process(() -> new MyProcessor(), "invStore"); // connect store to processor by providing store name

Problem

For one of my Kafka streams apps, I need to use the features of both DSL and Processor API. My streaming app flow is ``` source -> selectKey -> filter -> aggregate (on a window) -> sink ``` After aggregation I need to send a SINGLE aggregated message to the sink. So I define my topology as below ``` KStreamBuilder builder = new KStreamBuilder(); KStream<String, String> source = builder.stream(source_stream); source.selectKey(new MyKeyValueMapper()) .filterNot((k,v) -> k.equals("UnknownGroup")) .process(() -> new MyProcessor()); ``` I define a custom `StateStore` and register it with my processor as below ``` public class MyProcessor implements Processor<String, String> { private ProcessorContext context = null; Serde<HashMapStore> invSerde = Serdes.serdeFrom(invJsonSerializer, invJsonDeserializer); KeyValueStore<String, HashMapStore> invStore = (KeyValueStore) Stores.create("invStore") .withKeys(Serdes.String()) .withValues(invSerde) .persistent() .build() .get(); public MyProcessor() { } @Override public void init(ProcessorContext context) { this.context = context; this.context.register(invStore, false, null); // register the store this.context.schedule(10 * 60 * 1000L); } @Override public void process(String partitionKey, String message) { try { MessageModel smb = new MessageModel(message); HashMapStore oldStore = invStore.get(partitionKey); if (oldStore == null) { oldStore = new HashMapStore(); } oldStore.addSmb(smb); invStore.put(partitionKey, oldStore); } catch (Exception e) { e.printStackTrace(); } } @Override public void punctuate(long timestamp) { // processes all the messages in the state store and sends single aggregate message } @Override public void close() { invStore.close(); } } ``` When I run the app, I get `java.lang.NullPointerException` Exception in thread "StreamThread-18" java.lang.NullPointerException at org.apache.kafka.streams.state.internals.MeteredKeyValueStore.flush(MeteredKeyValueStore.java:167) at org.apache.kafka.streams.processor.internals.ProcessorStateManager.flush(ProcessorStateManager.java:332) at org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:252) at org.apache.kafka.streams.processor.internals.StreamThread.commitOne(StreamThread.java:446) at org.apache.kafka.streams.processor.internals.StreamThread.commitAll(StreamThread.java:434) at org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:422) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:340) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:218) Any idea what's going wrong here?

Original source