title: TopologyException on left join
body: <p>I'm trying to do a simple <code>stream.leftJoin(table)</code> but get the following exception at runtime: </p>
  
  <pre><code>TopologyException: Invalid topology: StateStore null is not added yet
  </code></pre>
  
  <p>This is what my code roughly looks like, I commented out the implementation details to keep it short:</p>
  
  <pre><code>val streamsConfiguration: Properties = {
    val p = new Properties()
    // api config
    p.put(StreamsConfig.APPLICATION_ID_CONFIG /**/)
    p.put(StreamsConfig.CLIENT_ID_CONFIG /**/)
    // kafka broker
    p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
    // local state store
    p.put(StreamsConfig.STATE_DIR_CONFIG, "./streams-state")
    p.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
    // serdes
    p.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, classOf[StringSerde])
    p.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, classOf[StringSerde])
    p
  }
  
  val builder = new StreamsBuilderS()
  
  val rawInfoTable: KTableS[String, String] = builder.table("station_info")
  val infoTable: KTableS[String, StationInfo] = rawInfoTable.mapValues{jsonString =&gt;
    /** */
  }.filter(/** */)
    .mapValues((/** */)
  
  val rawStatusStream: KStreamS[String, String] = builder.stream("station_status")
  val statusStream: KStreamS[String, StationStatus] = rawStatusStream.flatMapValues{jsonString =&gt;
    /** */
  }
  
  val outputStream: KStreamS[String, String] = statusStream
    .leftJoin(infoTable, calculateStats)
    .filter((_, availability) =&gt; {
      /** */
    })
    .map((stationId: String, availability) =&gt; {
      /** */
    })
  
  outputStream.to("low_availability")
  
  val streams = new KafkaStreams(builder.build(), streamsConfiguration)
  
  streams.cleanUp()
  streams.start()
  </code></pre>
  
  <p>I even tried to manually add a <code>StateStore</code> via:</p>
  
  <pre><code>val store = Stores.inMemoryKeyValueStore("my-store")
  val storeBuilder = Stores.keyValueStoreBuilder(store, new StringSerde(), new StringSerde())
  val builder = new StreamsBuilderS()
  builder.addStateStore(storeBuilder)
  </code></pre>
  
  <p>But it doesn't seem to change anything. I'm using the kafka streams wrapper from lightbend: <code>"com.lightbend" %% "kafka-streams-scala" % "0.2.1"</code></p>
  
  <p>All the examples I checked don't seem to care about adding a state store, so I'm somewhat confused. Can somebody point me in the right direction? Does this have to do something the <code>STATE_DIR_CONFIG</code>? Or with the Kafka cluster I'm running locally?</p>