title: InvalidStateStoreException on KStream join using GlobalKtables
body: <p>I have a Kafka Streams application where I am joining a KStream that reads
  from "topic1" with a GlobalKTable that reads from "topic2" and then with
  another GlobalKTable that reads from "topic3".</p>
  
  <p>When I try to push messages to all 3 topics at the same time then I get following exception - </p>
  
  <blockquote>
    <p>org.apache.kafka.streams.errors.InvalidStateStoreException</p>
  </blockquote>
  
  <p>If I push messages one by one in these topics i.e push messages in topic2 then in topic3 and then in topic1, then I do not get this
  exception.</p>
  
  <p>I have also added StateListener before I start KafkaStreams</p>
  
  <pre><code>KafkaStreams.StateListener stateListener = new KafkaStreams.StateListener() {
              @Override
              public void onChange (KafkaStreams.State newState, KafkaStreams.State oldState) {
                  if(newState == KafkaStreams.State.REBALANCING) {
                      try {
                          Thread.sleep(1000);
                      }
                      catch (InterruptedException e) {
                          e.printStackTrace();
                      }
                  }
              }
          };
  streams.setStateListener(stateListener);
  streams.start();
  </code></pre>
  
  <p>Also I wait till the store is queryable after the stream has started by calling following method</p>
  
  <pre><code>public static  &lt;T&gt; T waitUntilStoreIsQueryable(final String storeName,
                                                     final QueryableStoreType&lt;T&gt; queryableStoreType,
                                                     final KafkaStreams streams) throws InterruptedException {
          while (true) {
              try {
                  return streams.store(storeName, queryableStoreType);
              } catch (final InvalidStateStoreException ignored) {
                  // store not yet ready for querying
                  Thread.sleep(100);
              }
          }
      }
  </code></pre>
  
  <p>Following is the Kafka Streams and GlobalKTable join code:</p>
  
  <pre><code>KStream&lt;String, GenericRecord&gt; topic1KStream =
             builder.stream(
                 "topic1",
                 Consumed.with(Serdes.String(), genericRecordSerde)
             );
  GlobalKTable&lt;String, GenericRecord&gt; topic2KTable =
             builder.globalTable(
                 "topic2",
                 Consumed.with(Serdes.String(), genericRecordSerde),
                 Materialized.&lt;String, GenericRecord, KeyValueStore&lt;Bytes, byte[]&gt;&gt;as("topic2-global-store")
                     .withKeySerde(Serdes.String())
                     .withValueSerde(genericRecordSerde)
             );
    GlobalKTable&lt;String, GenericRecord&gt; topic3KTable =
             builder.globalTable(
                 "topic3",
                 Consumed.with(Serdes.String(), genericRecordSerde),
                 Materialized.&lt;String, GenericRecord, KeyValueStore&lt;Bytes, byte[]&gt;&gt;as("topic3-global-store")
                     .withKeySerde(Serdes.String())
                     .withValueSerde(genericRecordSerde)
             );
  
  KStream&lt;String, MergedObj&gt; stream_topic1_topic2 = topic1KStream.join(
             topic2KTable,
             (topic2Id, topic1Obj) -&gt; topic1.get("id").toString(),
             (topic1Obj, topic2Obj) -&gt; new MergedObj(topic1Obj, topic2Obj)
         );
         final KStream&lt;String, GenericRecord&gt; enrichedStream =
          stream_topic1_topic2.join(
             topic3KTable,
             (topic2Id, mergedObj) -&gt; mergedObj.topic3Id(),
             (mergedObj, topic3Obj) -&gt; new Enriched(
                 mergedObj.topic1Obj,
                 mergedObj.topic2Obj,
                 topic3Obj
             ).enrich()
         );
  enrichedStream.to("enrichedStreamTopic", Produced.with(Serdes.String(),getGenericRecordSerde()));
  </code></pre>
  
  <p>The above code is very similar to <a href="https://github.com/confluentinc/kafka-streams-examples/blob/5.2.1-post/src/main/java/io/confluent/examples/streams/GlobalKTablesExample.java" rel="nofollow noreferrer">this</a>.</p>
  
  <p>When I try to push messages to all 3 topics at the same time then I get
  following exception:</p>
  
  <pre><code>org.apache.kafka.streams.errors.StreamsException: Exception caught in process. taskId=0_1, processor=KSTREAM-SOURCE-0000000000, topic=topic1,
  partition=1, offset=61465,
  stacktrace=org.apache.kafka.streams.errors.InvalidStateStoreException:
  Store topic2-global-store is currently closed.
      at
  org.apache.kafka.streams.state.internals.WrappedStateStore.validateStoreOpen(WrappedStateStore.java:66)
      at
  org.apache.kafka.streams.state.internals.CachingKeyValueStore.get(CachingKeyValueStore.java:150)
      at
  org.apache.kafka.streams.state.internals.CachingKeyValueStore.get(CachingKeyValueStore.java:37)
      at
  org.apache.kafka.streams.state.internals.MeteredKeyValueStore.get(MeteredKeyValueStore.java:135)
      at
  org.apache.kafka.streams.processor.internals.ProcessorContextImpl$KeyValueStoreReadOnlyDecorator.get(ProcessorContextImpl.java:245)
      at
  org.apache.kafka.streams.kstream.internals.KTableSourceValueGetterSupplier$KTableSourceValueGetter.get(KTableSourceValueGetterSupplier.java:49)
      at
  org.apache.kafka.streams.kstream.internals.KStreamKTableJoinProcessor.process(KStreamKTableJoinProcessor.java:71)
      at
  org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:117)
      at
  org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:183)
      at
  org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:162)
      at
  org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:122)
      at
  org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:87)
      at
  org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:364)
      at
  org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.process(AssignedStreamsTasks.java:199)
      at
  org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:420)
      at
  org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:890)
      at
  org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:805)
      at
  org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:774)
  </code></pre>