title: Replacing RocksDB with In-memory state store in Kafka Streams
body: <p>I'm using Kafka Streams 0.10.1.1 release.</p>
  
  <p>the RocksDB implementation for state store can't handle our 50k/msg rate so I want to change the state store to be the in-memory one. This should be possible according to the docs: <a href="http://docs.confluent.io/3.1.0/streams/architecture.html#state" rel="nofollow noreferrer">http://docs.confluent.io/3.1.0/streams/architecture.html#state</a></p>
  
  <p>However, when I implement this:</p>
  
  <pre class="lang-scala prettyprint-override"><code>val stateStore = Stores.create(stateStoreName).withStringKeys().withStringKeys().inMemory().build()
  
  val procSuppl: KStreamAggregate = ... // I'll spare the implementation details
  
  streamBuilder.addSource(
    "mysource",
    new StringDeserializer(),
    new StringDeserializer(),
    "input_topic"
  ).addProcessor("proc", procSuppl,  "mysource").addStateStore(stateStore, "proc")
  </code></pre>
  
  <p>I end up with this error in runtime:</p>
  
  <pre><code>Caused by: java.lang.ClassCastException: org.apache.kafka.streams.state.internals.MeteredKeyValueStore cannot be cast to org.apache.kafka.streams.state.internals.CachedStateStore
  2017-01-23T13:19:11.830674020Z  at org.apache.kafka.streams.kstream.internals.KStreamAggregate$KStreamAggregateProcessor.init(KStreamAggregate.java:62)
  </code></pre>
  
  <p>The implementation of the above method is:</p>
  
  <pre class="lang-java prettyprint-override"><code>public void init(ProcessorContext context) {
          super.init(context);
          store = (KeyValueStore&lt;K, T&gt;) context.getStateStore(storeName);
          ((CachedStateStore) store).setFlushListener(new ForwardingCacheFlushListener&lt;K, V&gt;(context, sendOldValues));
      }
  </code></pre>
  
  <p>Why is it trying to cast the state store to a <code>CachedStateStore</code> instance? How can I implement a simple in-memory state store which should be possible according to docs?</p>
  
  <p>Thanks</p>