title: Processing time windows doesn't work on finite data sources in Apache Flink
body: <p>I'm trying to apply a very simple window function to a finite data stream in Apache Flink (<em>locally, no cluster</em>). Here's the example:</p>
  
  <pre><code>val env = StreamExecutionEnvironment.getExecutionEnvironment
  env
    .fromCollection(List("a", "b", "c", "d", "e"))
  
    .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(1)))
    .trigger(ProcessingTimeTrigger.create)
    .process(new ProcessAllWindowFunction[String, String, TimeWindow] {
      override def process(context: Context, elements: Iterable[String], out: Collector[String]): Unit = {
        out.collect(elements.toList.sorted.toString())
      }
    })
  
    .print()
  
  env.execute()
  </code></pre>
  
  <p>Here, I try to group all the elements that arrive into the window within a second and then just print these groups. </p>
  
  <p>I assumed that all the elements would be produces in much less than one second and get into one window, so there will be one incoming element in <code>print()</code>. However, <strong>nothing is printed</strong> at all when I run this.</p>
  
  <p>If I remove all the windowing stuff, like</p>
  
  <pre><code>val env = StreamExecutionEnvironment.getExecutionEnvironment
  env
    .fromCollection(List("a", "b", "c", "d", "e"))
    .print()
  </code></pre>
  
  <p>I see the elements printed after the run. I also tried this with the file source, no difference.</p>
  
  <p>The default parallelism on my machine is 6. If I experiment with the level of parallelism and delays, like this</p>
  
  <pre><code>val env = StreamExecutionEnvironment.createLocalEnvironment(2)
  env
    .fromCollection(List("a", "b", "c", "d", "e"))
    .map { x =&gt; Thread.sleep(1500); x }
  </code></pre>
  
  <p>I able to get some--not all--elements into groups, which are printed.</p>
  
  <p>My first assumption is that the source finishes much faster than 1 second and the task is shut down before the window's timer fires. Debugging showed that <a href="https://github.com/apache/flink/blob/8674b69964eae50cad024f2c5caf92a71bf21a09/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ProcessingTimeTrigger.java#L36" rel="nofollow noreferrer">the timer setting line in <code>ProcessingTimeTrigger</code></a> is reached. Shouldn't all started timers finish before a task shuts down (at least this is the impression I got from <a href="https://github.com/apache/flink/blob/8674b69964eae50cad024f2c5caf92a71bf21a09/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java#L327" rel="nofollow noreferrer">the code</a>)?</p>
  
  <p>Could you please help me understand this and make this more deterministic?</p>
  
  <p><strong>Update #1, 23.09.2018:</strong></p>
  
  <p>I also experimented with event time windows instead of processing time windows. If I do this:</p>
  
  <pre><code>val env = StreamExecutionEnvironment.getExecutionEnvironment
  env
    .fromCollection(List("a", "b", "c", "d", "e"))
    .assignTimestampsAndWatermarks(new AscendingTimestampExtractor[String] {
      override def extractAscendingTimestamp(element: String): Long = {
        element.charAt(0).toInt
      }
    })
  
    .windowAll(TumblingEventTimeWindows.of(Time.seconds(1)))
    .trigger(EventTimeTrigger.create)
    .process(new ProcessAllWindowFunction[String, String, TimeWindow] {
      override def process(context: Context, elements: Iterable[String], out: Collector[String]): Unit = {
        out.collect(elements.toList.toString())
      }
    })
  
    .print()
  
  env.execute()
  </code></pre>
  
  <p>Then again nothing is printed. The debugger shows that the trigger's <code>onElement</code> is called for every element, but <code>onEventTime</code> is never called.</p>
  
  <p>Also, if I modify the the timestamp extractor to make bigger steps:</p>
  
  <pre><code>element.charAt(0).toInt * 1000
  </code></pre>
  
  <p>all the elements are printed (one element per group, which is expected), apart from the last.</p>
  
  <p><strong>Update #2, 23.09.2018:</strong></p>
  
  <p>Update #1 is answered in <a href="https://stackoverflow.com/questions/52467413/processing-time-windows-doesnt-work-on-finite-data-sources-in-apache-flink?noredirect=1#comment91879894_52467413">this comment</a>.</p>