title: Binding GlobalStateStore into Processor with spring-cloud-stream-binder-kafka
body: <p>Initial Question:
  I have a question how I can bind my GlobalStateStore to a processor. My Application has a GlobalStateStore with an own processor (&quot;GlobalConfigProcessor&quot;) to keep the Store up to date. Also, I have another Processor (&quot;MyClassProcessor&quot;) which is called in my Consumer Function. Now I try to access the store from MyClassProcessor, but I get an exception saying : Invalid topology: StateStore config_statestore is not added yet.</p>
  <p>Update on current situation:
  I setup a test repository to give a better overview over my situation. This can be found here: <a href="https://github.com/fx42/store-example" rel="nofollow noreferrer">https://github.com/fx42/store-example</a></p>
  <p>As you can see in the repo, I have two Consumers which both consume different topics. The Config-Topic provides an event which I want to write to a GlobalStateStore. Here are the StateStoreUpdateConsumer.java and the StateStoreProcessor.java involved.
  With the MyClassEventConsumer.java I process another Input-Topic and want to read values from the GlobalStateStore.
  As provided in this <a href="https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/blob/main/docs/src/main/asciidoc/kafka-streams.adoc#streamsbuilderfactorybean-customizer" rel="nofollow noreferrer">doc</a> I can't initialize GlobalStateStores just as StateStoreBean but instead I have to add this actively with the StreamsBuilderFactoryBeanCustomizer Bean. This Code is currently commented out in the StreamConfig.java. Without this code I get the Exception</p>
  <pre><code>org.springframework.kafka.KafkaException: Could not start stream: ; nested exception is org.apache.kafka.streams.errors.TopologyException: Invalid topology: Topology has no stream threads and no global threads, must subscribe to at least one source topic or global table.
  </code></pre>
  <p>If the code is in use I get the exception:</p>
  <pre><code>org.springframework.kafka.KafkaException: Could not start stream: ; nested exception is org.apache.kafka.streams.errors.StreamsException: Unable to initialize state, this can happen if multiple instances of Kafka Streams are running in the same state directory
  </code></pre>
  <p>So this leads my to the decision, that I have a configuration problem so the topology is messed up.</p>
  <p>Questions:</p>
  <ol>
  <li>When I provide the Processor for the GlobalStateStore directly via</li>
  </ol>
  <pre><code>streamBuilder.addGlobalStore(storeBuilder, configInputTopic,
                              Consumed.with(Serdes.String(), Serdes.String()), () -&gt; new StateStoreProcessor(statestoreName));
  </code></pre>
  <p>do I have to provide a Consumer Function for this Processor or do I even have to mention it in the function configuration/application.yml?</p>
  <ol start="2">
  <li><p>Is there a way NOT to provide a ProcessorSupplier into the <code>addGlobalStore</code> call and just use the functional way for this?</p>
  </li>
  <li><p>How can I handle this GlobalStateStore if there are two different topologies for both the defined functions?</p>
  </li>
  </ol>
  <p>Here is the commented out StreamBuilderFactoryCustomizer Bean which I use to add the GlobalStateStore to the FactoryBean:</p>
  <pre><code>@Bean
  StreamsBuilderFactoryBeanCustomizer streamsBuilderFactoryBeanCustomizer(
              StoreBuilder&lt;KeyValueStore&lt;String, String&gt;&gt; storeBuilder) {
  
          return factoryBean -&gt; {
              try {
                  var streamBuilder = factoryBean.getObject();
                  streamBuilder.addGlobalStore(storeBuilder, configInputTopic,
                          Consumed.with(Serdes.String(), Serdes.String()), () -&gt; new StateStoreProcessor(statestoreName));
  
              } catch (Exception e) {
                  e.printStackTrace();
              }
          };
      } };
  }
  </code></pre>