title: Apache Flink: Count window with timeout
body: <p>Here is a simple code example to illustrate my question:</p>
  
  <pre><code>case class Record( key: String, value: Int )
  
  object Job extends App
  {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val data = env.fromElements( Record("01",1), Record("02",2), Record("03",3), Record("04",4), Record("05",5) )
    val step1 = data.filter( record =&gt; record.value % 3 != 0  ) // introduces some data loss
    val step2 = data.map( r =&gt; Record( r.key, r.value * 2 ) )
    val step3 = data.map( r =&gt; Record( r.key, r.value * 3 ) )
    val merged = step1.union( step2, step3 )
    val keyed = merged.keyBy(0)
    val windowed = keyed.countWindow( 3 )
    val summed = windowed.sum( 1 )
    summed.print()
    env.execute("test")
  }
  </code></pre>
  
  <p>This produces the following result:</p>
  
  <pre><code>Record(01,6)
  Record(02,12)
  Record(04,24)
  Record(05,30)
  </code></pre>
  
  <p>As expected, no result is produced for key "03" because the count window expects 3 elements and only two are present in the stream.</p>
  
  <p>What I would like is some kind of count window with timeout so that, after a certain timeout, if the number of elements expected by the count window is not reached, a partial result is produced with the existing elements.</p>
  
  <p>With this behavior, in my example, a Record(03,15) would be produced when the timeout is reached.</p>