title: How to add a global state store in a Spring Cloud Stream application
body: <p>I'm trying to configure a global state store in a Spring Cloud Stream application but it fails with :</p>
  <p>Exception in thread &quot;svc-toa-app-datainteg-io-dev-fd2e1a47-0758-4b95-9b2e-022ca4df95ca-
  StreamThread-1&quot; org.apache.kafka.streams.errors.TopologyException: Invalid topology:
  Topic is unknown to the topology. This may happen if different KafkaStreams instances of
  the same application execute different Topologies. Note that Topologies are only identical if all operators are added in the same order.</p>
  <p>Following the official documentation I added the following to configure the state store :</p>
  <pre class="lang-java prettyprint-override"><code> @Bean
      public StreamsBuilderFactoryBeanCustomizer streamsBuilderFactoryBeanCustomizer() {
          return factoryBean -&gt; {
              if (factoryBean.getStreamsConfiguration().getProperty(StreamsConfig.APPLICATION_ID_CONFIG)
                             .equals(applicationId)) {
                  try {
  
                      final var keyValueStore = new KeyValueStoreBuilder&lt;&gt;(
                          Stores.inMemoryKeyValueStore(PRODUCT_INFO_STORE),
                          productInfoKeySerde(),
                          productInfoValueSerde(),
                          Time.SYSTEM
                      );
  
                      final StreamsBuilder streamsBuilder = factoryBean.getObject();
  
                      streamsBuilder.addGlobalStore(
                          keyValueStore.withLoggingDisabled(),
                          productInfoTopic,
                          Consumed.with(productInfoKeySerde(), productInfoValueSerde()),
                          ProductGlobalStateStoreProcessor::new
                      );
                  } catch (Exception e) {
                      e.printStackTrace();
                  }
              }
          };
      }
  </code></pre>
  <p>And I have a processor where I want to use the state store :</p>
  <pre class="lang-java prettyprint-override"><code>public class ProductGlobalStateStoreProcessor implements Processor&lt;ProductInfoKey, ProductInfo, Void, Void&gt; {
      KeyValueStore&lt;ProductInfoKey, ProductInfo&gt; stateStore;
  
      @Override
      public void init(ProcessorContext&lt;Void, Void&gt; context) {
          stateStore = context.getStateStore(ProductGlobalStateStoreConfiguration.PRODUCT_INFO_STORE);
      }
  
      @Override
      public void process(Record&lt;ProductInfoKey, ProductInfo&gt; record) {
          stateStore.put(record.key(), record.value());
      }
  }
  </code></pre>
  <p>I'm a little lost on what I'm missing, I've tried to add a gktable binding in the properties but it does not change anything.</p>