title: Flink Streaming: From one window, lookup state in another window
body: <p>I have two streams:</p>
  
  <ul>
  <li>Measurements</li>
  <li>WhoMeasured (metadata about who took the measurement)</li>
  </ul>
  
  <p>These are the case classes for them:</p>
  
  <pre><code>case class Measurement(var value: Int, var who_measured_id: Int)
  case class WhoMeasured(var who_measured_id: Int, var name: String)
  </code></pre>
  
  <p>The <code>Measurement</code> stream has a lot of data. The <code>WhoMeasured</code> stream has little. In fact, for each <code>who_measured_id</code> in the <code>WhoMeasured</code> stream, only 1 name is relevant, so old elements can be discarded if one with the same <code>who_measured_id</code> arrives. This is essentially a HashTable that gets filled by the <code>WhoMeasured</code> stream.</p>
  
  <p>In my custom window function</p>
  
  <pre><code>class WFunc extends WindowFunction[Measurement, Long, Int, TimeWindow] {
    override def apply(key: Int, window: TimeWindow, input: Iterable[Measurement], out: Collector[Long]): Unit = {
  
      // Here I need access to the WhoMeasured stream to get the name of the person who took a measurement
      // The following two are equivalent since I keyed by who_measured_id
      val name_who_measured = magic(key)
      val name_who_measured = magic(input.head.who_measured_id)
    }
  }
  </code></pre>
  
  <p>This is my job. Now as you might see, there is something missing: The combination of the two streams.</p>
  
  <pre><code>val who_measured_stream = who_measured_source
    .keyBy(w =&gt; w.who_measured_id)
    .countWindow(1)
  
  val measurement_stream = measurements_source
    .keyBy(m =&gt; m.who_measured_id)
    .timeWindow(Time.seconds(60), Time.seconds(5))
    .apply(new WFunc)
  </code></pre>
  
  <p>So in essence this is sort of a lookup table that gets updated when new elements in the <code>WhoMeasured</code> stream arrive.</p>
  
  <p>So the question is: How to achieve such a lookup from one <code>WindowedStream</code> into another?</p>
  
  <hr>
  
  <p><strong>Follow Up:</strong></p>
  
  <p>After implementing in the way Fabian suggested, the job always fails with some sort of serialization issue:</p>
  
  <pre><code>[info] Loading project definition from /home/jgroeger/Code/MeasurementJob/project
  [info] Set current project to MeasurementJob (in build file:/home/jgroeger/Code/MeasurementJob/)
  [info] Compiling 8 Scala sources to /home/jgroeger/Code/MeasurementJob/target/scala-2.11/classes...
  [info] Running de.company.project.Main dev MeasurementJob
  [error] Exception in thread "main" org.apache.flink.api.common.InvalidProgramException: The implementation of the RichCoFlatMapFunction is not serializable. The object probably contains or references non serializable fields.
  [error]     at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:100)
  [error]     at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.clean(StreamExecutionEnvironment.java:1478)
  [error]     at org.apache.flink.streaming.api.datastream.DataStream.clean(DataStream.java:161)
  [error]     at org.apache.flink.streaming.api.datastream.ConnectedStreams.flatMap(ConnectedStreams.java:230)
  [error]     at org.apache.flink.streaming.api.scala.ConnectedStreams.flatMap(ConnectedStreams.scala:127)
  [error]     at de.company.project.jobs.MeasurementJob.run(MeasurementJob.scala:139)
  [error]     at de.company.project.Main$.main(Main.scala:55)
  [error]     at de.company.project.Main.main(Main.scala)
  [error] Caused by: java.io.NotSerializableException: de.company.project.jobs.MeasurementJob
  [error]     at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184)
  [error]     at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
  [error]     at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
  [error]     at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
  [error]     at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
  [error]     at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
  [error]     at org.apache.flink.util.InstantiationUtil.serializeObject(InstantiationUtil.java:301)
  [error]     at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:81)
  [error]     ... 7 more
  java.lang.RuntimeException: Nonzero exit code returned from runner: 1
      at scala.sys.package$.error(package.scala:27)
  [trace] Stack trace suppressed: run last MeasurementJob/compile:run for the full output.
  [error] (MeasurementJob/compile:run) Nonzero exit code returned from runner: 1
  [error] Total time: 9 s, completed Nov 15, 2016 2:28:46 PM
  
  Process finished with exit code 1
  </code></pre>
  
  <p>The error message:</p>
  
  <pre><code>The implementation of the RichCoFlatMapFunction is not serializable. The object probably contains or references non serializable fields.
  </code></pre>
  
  <p>However, the only field my <code>JoiningCoFlatMap</code> has is the suggested <code>ValueState</code>.</p>
  
  <p>The signature looks like this:</p>
  
  <pre><code>class JoiningCoFlatMap extends RichCoFlatMapFunction[Measurement, WhoMeasured, (Measurement, String)] {
  </code></pre>