title: Flink Streaming: Data stream that gets controlled by control stream
body: <p>I have a question which is a variation of this question: <a href="https://stackoverflow.com/questions/41064059/flink-how-to-store-state-and-use-in-another-stream">Flink: how to store state and use in another stream?</a></p>
  
  <p>I have two streams:</p>
  
  <ol>
  <li><code>val ipStream: DataStream[IPAddress] = ???</code></li>
  <li><code>val routeStream: DataStream[RoutingTable] = ???</code></li>
  </ol>
  
  <p>I want to find out which route which package uses. Usually this can be done with:</p>
  
  <pre><code>val ip = IPAddress("10.10.10.10")
  val table = RoutingTable(Seq("10.10.10.0/24", "5.5.5.0/24"))
  val route = table.lookup(ip) // == "10.10.10.0/24"
  </code></pre>
  
  <p>The problem here is that I cannot really key the stream here since that requires both the complete table as well as the ip address (and keys must be computed isolated).</p>
  
  <p>For every element from the <code>ipStream</code>, I need the latest <code>routeStream</code> element. Right now I'm using a hack that all of that is processed non-parallel:</p>
  
  <pre><code>ipStream
    .connect(routeStream)
    .keyBy(_ =&gt; 0, _ =&gt; 0)
    .flatMap(new MyRichCoFlatMapFunction) // with ValueState[RoutingTable]
  </code></pre>
  
  <p>This sounds like the use case for a broadcast strategy. However, the routeStream will be updated and is not fixed in a file. The question remains: <strong>Is there a way to have two streams, one of which contains changing control data for the other stream?</strong></p>