...

builder.addStateStore(Stores.create("Counts").withStringKeys().withIntegerValues().persistent().enableCaching().build(), "Process");

The implication of adding caching to the processor API is that the caching layer will also do forwarding. Hence if a store's cache is enabled, the use of context.forward is not needed anymore. If it is used, it will be a noop.

Public Interfaces

This KIP adds the cache.max.bytes.buffering configuration to the streams configuration as described above. 

...