Q:

How to use a KeyValueStore state store in DSL?

KeyValueStore<String, Long> kvStore=(KeyValueStore<String, Long>) 
Stores.create("InterWindowStore1").withKeys(Serdes.String())
                .withValues(Serdes.Long())
                .persistent()
                .build().get();` 

I have created statestore as shown in above code and try to insert into kvStore.put(key, value); but it is throwing me NPE 
Caused by: java.lang.NullPointerException
    at org.apache.kafka.streams.state.internals.MeteredKeyValueStore.put(MeteredKeyValueStore.java:117)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:82)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:202)
    at org.apache.kafka.streams.kstream.internals.ForwardingCacheFlushListener.apply(ForwardingCacheFlushListener.java:42)
    at org.apache.kafka.streams.state.internals.CachingWindowStore.maybeForward(CachingWindowStore.java:103)
    at org.apache.kafka.streams.state.internals.CachingWindowStore.access$200(CachingWindowStore.java:34)
    at org.apache.kafka.streams.state.internals.CachingWindowStore$1.apply(CachingWindowStore.java:86)
    at org.apache.kafka.streams.state.internals.NamedCache.flush(NamedCache.java:131)
    at org.apache.kafka.streams.state.internals.NamedCache.flush(NamedCache.java:95)

A:

As you describe in your comments you are basically doing a window aggregation:
KStream stream = ...
KTable table = stream.groupByKey().aggregate(..., TimeWindow.of(...));

As you KTable stream might contain updates for you window aggregation, you want to modify this stream. For this, you can use a stateful transformer or value-transformer:
StateStoreSupplier myState = State.create("nameOfMyState")....;

KStream result = table.toStream().transform(..., "nameOfMyState");

Finally, you can write your result to the output topic:
result.to("output-topic");

Your Transformer that you provide to transform can get the state via the given context in init() and use within transform() each time a window output is generated/updated.

