title: Flink's broadcast state behavior
body: <p>I am trying to play with flink's broacast state with a simple case.</p>
  
  <p>I juste want to multiply an integer stream by another integer into a broadcast stream.</p>
  
  <p>The behavior of my Broadcast is "weird", if I put too few elements in my input stream (like 10), nothing happen and my <code>MapState</code> is empty, but if I put more elements (like 100) I have the behavior I want (multiply the integer stream by 2 here).</p>
  
  <p><strong>Why broadcast stream is not taking into account if I gave too few elements ?</strong></p>
  
  <p><strong>How can I control when the broadcast stream is working ?</strong></p>
  
  <p><em>Optional: I want to keep only the last element of my broadcast stream, is <code>.clear()</code> the good way ?</em></p>
  
  <p>Thank you!</p>
  
  <p>Here's my <code>BroadcastProcessFunction</code>:</p>
  
  <pre><code>import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction
  import org.apache.flink.util.Collector
  import scala.collection.JavaConversions._
  
  class BroadcastProcess extends BroadcastProcessFunction[Int, Int, Int] {
    override def processElement(value: Int, ctx: BroadcastProcessFunction[Int, Int, Int]#ReadOnlyContext, out: Collector[Int]) = {
      val currentBroadcastState = ctx.getBroadcastState(State.mapState).immutableEntries()
      if (currentBroadcastState.isEmpty) {
        out.collect(value)
      } else {
        out.collect(currentBroadcastState.last.getValue * value)
      }
    }
  
    override def processBroadcastElement(value: Int, ctx: BroadcastProcessFunction[Int, Int, Int]#Context, out: Collector[Int]) = {
      // Keep only last state
      ctx.getBroadcastState(State.mapState).clear()
      // Add state
      ctx.getBroadcastState(State.mapState).put("key", value)
    }
  }
  </code></pre>
  
  <p>And my <code>MapState</code>:</p>
  
  <pre><code>import org.apache.flink.api.common.state.MapStateDescriptor
  import org.apache.flink.api.scala._
  
  object State {
    val mapState: MapStateDescriptor[String, Int] =
      new MapStateDescriptor(
        "State",
        createTypeInformation[String],
        createTypeInformation[Int]
      )
  }
  </code></pre>
  
  <p>And my <code>Main</code>:</p>
  
  <pre><code>import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
  import org.apache.flink.api.scala._
  
  object Broadcast {
    def main(args: Array[String]): Unit = {
      val numberElements = 100
      val env = StreamExecutionEnvironment.getExecutionEnvironment
      env.setParallelism(1)
      val broadcastStream = env.fromElements(2).broadcast(State.mapState)
      val input = (1 to numberElements).toList
      val inputStream = env.fromCollection(input)
      val outputStream = inputStream
        .connect(broadcastStream)
        .process(new BroadcastProcess())
      outputStream.print()
      env.execute()
    }
  }
  </code></pre>
  
  <p>Edit: I use Flink 1.5, and Broadcast State documentation is <a href="https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/stream/state/broadcast_state.html" rel="nofollow noreferrer">here</a>.</p>