当水印小于窗口结束时，触发基于事件时间的Flink窗口操作

我在Flink测试事件时间和水印。以下是我的代码。

object WatermarkTest {

 def main(args: Array[String]): Unit = {
     val env = StreamExecutionEnvironment.getExecutionEnvironment
     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

     val properties = new Properties()
     properties.setProperty("bootstrap.servers", "127.0.0.1:9092")
     properties.setProperty("group.id", "enven-test")

     env.getConfig.setAutoWatermarkInterval(1L)
     val input = env.addSource(new FlinkKafkaConsumer011[String]("event-time-topic", new SimpleStringSchema(), properties))

     val inputMap = input.map(f=> {
       val arr = f.split(",")
       val code = arr(0)
       val time = arr(1).toLong
       MyEvent(code, time)
    })

    val watermark = inputMap.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessGenerator())
    val window = watermark
       .keyBy(_.code)
       .window(TumblingEventTimeWindows.of(Time.seconds(5)))
       .apply(new WindowFunctionTest)

    window.print()

   env.execute()
}

class WindowFunctionTest extends WindowFunction[MyEvent,(String, Int,String,String,String,String),String,TimeWindow]{

   override def apply(key: String, window: TimeWindow, input: Iterable[MyEvent], out: Collector[(String, Int,String,String,String,String)]): Unit = {
   val list = input.toList.sortBy(_.time)
   val format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS")

out.collect(key,input.size,format.format(list.head.time),format.format(list.last.time),format.format(window.getStart),format.format(window.getEnd))

}

}
}
以下是事件时间和水印生成器：

class BoundedOutOfOrdernessGenerator extends

  AssignerWithPeriodicWatermarks[MyEvent] {
  val maxOutOfOrderness = 10000L

  var currentMaxTimestamp: Long = _

  val format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS")

  var watermark: Watermark = null
  var timestamp: Long = _

  override def extractTimestamp(element: MyEvent, previousElementTimestamp: Long): Long = {
       timestamp = element.time
       currentMaxTimestamp = if (timestamp > currentMaxTimestamp) timestamp else currentMaxTimestamp
       println("timestamp:" + element.code +","+ element.time + "|" +format.format(element.time) +", currentMaxTimestamp: "+  currentMaxTimestamp + "|"+ format.format(currentMaxTimestamp) + ", watermark: "+ format.format(watermark.getTimestamp))
       timestamp;
  }

  override def getCurrentWatermark(): Watermark = {
       watermark = new Watermark((currentMaxTimestamp - maxOutOfOrderness)/1000*1000);
      watermark
  }

}
在我看来，第一次计算应该在水印之后：2016-04-27 19：34：25.000。测试结果显示水印水印后触发计算：2016-04-27 19：34：24.000。任何人都能解释一下吗？