title: Using Custom Kafka State Stores in kstreams application
body: <p>We are using Kafka-streams included within the spring cloud stream Hoxton RC7 project (and therefore use the Kafka-streams and Kafka-client versions provided [2.3.1])</p>
  <pre><code>
  ext {
      set('springCloudVersion', 'Hoxton.SR7')
  }
  ...
  
  dependencies {
      // spring cloud stream
      implementation 'org.springframework.cloud:spring-cloud-stream-binder-kafka-streams'
      implementation 'org.springframework.cloud:spring-cloud-stream-binder-kafka'
      implementation(&quot;org.springframework.cloud:spring-cloud-stream&quot;)
      // redis 
      implementation 'io.lettuce:lettuce-core'
      implementation 'org.springframework.data:spring-data-redis'
      testCompile 'it.ozimov:embedded-redis:0.7.2'
      ...
  
  </code></pre>
  <p>We have implemented a kstreams application</p>
  <pre><code>@Bean
  public Consumer&lt;KStream&lt;String, IncomingEvent&gt;&gt; process() {
  
      return input -&gt; {
  
  </code></pre>
  <p>Where we do some aggregation within like:</p>
  <pre><code>.aggregate(Foo::new, (key, value1, aggregate) -&gt;
                  (aggregate == null || aggregate.getLastModified() == null || this.mustProcess(key, value1))
                          ? value1
                          : aggregate,
          materialized
  
  )
  </code></pre>
  <p>Now materialized should be a custom external state store (Redis):</p>
  <pre><code>Materialized&lt;String, Foo, KeyValueStore&lt;Bytes, byte[]&gt;&gt; materialized =
          Materialized.as(&quot;redis-store&quot;);
  </code></pre>
  <p>Which is provided by a StoreBuilder Bean:</p>
  <pre><code>@Bean
  public StoreBuilder&lt;KeyValueStore&lt;String, Foo&gt;&gt; builder(RedisKeyValueStoreBytes redisKeyValueStoreBytes){
      return Stores.keyValueStoreBuilder(supplier(redisKeyValueStoreBytes),
              new Serdes.StringSerde(),
              new SomeFooSerde());
  }
  
  
  public static KeyValueBytesStoreSupplier supplier(RedisKeyValueStoreBytes redisKeyValueStoreBytes) {
  
      return new KeyValueBytesStoreSupplier() {
          @Override
          public String name() {
              return &quot;redis-store&quot;;
          }
  
          @Override
          public KeyValueStore&lt;Bytes, byte[]&gt; get() {
              return redisKeyValueStoreBytes;
          }
  
          @Override
          public String metricsScope() {
              return &quot;redis-session-state&quot;;
          }
      };
  }
  </code></pre>
  <p>I now test the application with an EmbeddedKafka:</p>
  <pre><code>@ActiveProfiles(&quot;test&quot;)
  @RunWith(SpringRunner.class)
  @DirtiesContext(classMode = DirtiesContext.ClassMode.BEFORE_EACH_TEST_METHOD)
  @SpringBootTest(classes = {TestConfigurationTests.class})
  @EmbeddedKafka(count = 3, ports = {29901, 29902, 29903}, zookeeperPort = 33991)
  public class TestKafkaIntegration {
  </code></pre>
  <p>Where I try to access the state store and query the items added:</p>
  <pre><code>ReadOnlyKeyValueStore&lt;String, Foo&gt; queryableStore = interactiveQueryService.getQueryableStore(
          &quot;redis-store&quot;, QueryableStoreTypes.keyValueStore());
  return queryableStore;
  </code></pre>
  <p>But when I run my test I receive an error:</p>
  <pre><code>Caused by: org.springframework.kafka.KafkaException: Could not start stream: ; nested exception is org.springframework.kafka.KafkaException: Could not start stream: ; nested exception is org.apache.kafka.streams.errors.TopologyException: Invalid topology: StateStore redis-store is already added.
  </code></pre>
  <p>Several questions:</p>
  <ul>
  <li>The examples for using custom state stores explained by [1] use it within a Processor. Does this automatically mean, I am not able to use a custom state store in an aggregation?</li>
  <li>When it is not possible to use it within an aggregation, what is the point of using custom state stores anyway?</li>
  <li>When I slightly change the code above for the kstreams and define a processor instead of using materialized in the aggregate method, the error changes, it then complains about a missing state &quot;redis-store&quot; store while trying to execute getQueryableStore. But indeed i can see, that the addStateStoreBeans registers the 'redis-store'. How can this happen?</li>
  </ul>
  <p>The reason why I want to use a custom state store is, that I am not (really easily) able to have a dedicated hard disk for the application instance. To have a fast startup for the application I want to avoid processing of the complete changelog on each startup of the application (which should preferably take place several times a day and currently takes more than an hour). So now the last question:</p>
  <ul>
  <li>When a custom external state store is used, am I able to resume to the last state on application restart?</li>
  </ul>
  <p>[1] <a href="https://spring.io/blog/2019/12/09/stream-processing-with-spring-cloud-stream-and-apache-kafka-streams-part-6-state-stores-and-interactive-queries" rel="nofollow noreferrer">https://spring.io/blog/2019/12/09/stream-processing-with-spring-cloud-stream-and-apache-kafka-streams-part-6-state-stores-and-interactive-queries</a></p>