我正在尝试使用 EventTime . 我将从Kafka消费。我的数据可能会延迟一分钟（因此我的水印会延迟一分钟）并且出现故障。我有30秒的窗口。我有以下设置：
Kafka消费者.scala

object KafkaConsumer {
  def main(args: Array[String]) {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val properties = getServerProperties
    val consumer = new FlinkKafkaConsumer010[ObjectNode](getKafkaTopic, new JsonNodeDeserializationSchema, properties)
    consumer.setStartFromGroupOffsets()

    val stream = env.addSource(consumer)
      .assignTimestampsAndWatermarks(new WMAssigner)

    stream
      .keyBy { jsonEvent =>
        val key = jsonEvent.findValue("key").toString replaceAll("\"","")
        key.toString
      }
      .window(TumblingEventTimeWindows.of(Time.seconds(30)))
      .process { new SessionWindowProcessor }
      .print
    env.execute("EventTime Test")
  }
}

wmassigner.scala文件

class WMAssigner extends AssignerWithPeriodicWatermarks[ObjectNode] {

  var currentMaxTimestamp: Long = 0
  var currentWaterMark: Long = 0

  override def extractTimestamp(element: ObjectNode, previousElementTimestamp: Long): Long = {
    val lStr = element.findValue("ts").toString replaceAll("\"", "")
    currentMaxTimestamp = if(currentMaxTimestamp > lStr.toLong) currentMaxTimestamp else lStr.toLong
    currentMaxTimestamp
  }

  override def getCurrentWatermark: Watermark = {
      currentWaterMark = if(currentMaxTimestamp - 60000 < 0) 0 else currentMaxTimestamp - 60000
      new Watermark(currentWaterMark)

  }
}

会话窗口处理器.scala

class SessionWindowProcessor extends ProcessWindowFunction[ObjectNode,Long,String,TimeWindow] {
  override def process(key: String, context: Context, elements: Iterable[ObjectNode], out: Collector[Long]): Unit = {
    println("Processing!")
    var maxVal: Long = 0
    elements.foreach( value => {
      val valStr = value.findValue("value").toString replaceAll("\"","")
      maxVal = if(valStr.toLong > maxVal) valStr.toLong else maxVal
    })
    out.collect(maxVal)
  }
}

样本数据：

"{\"ts\": 0,     \"key\": \"a\", \"value\": 0}",
"{\"ts\": 15000, \"key\": \"a\", \"value\": 1}",
"{\"ts\": 30000, \"key\": \"a\", \"value\": 2}",
"{\"ts\": 90001, \"key\": \"a\", \"value\": 3}"

我希望在第四条消息传入后，我的第一个窗口将触发，值为 1 或者 2 （我现在不确定包容性是如何发挥作用的）。不幸的是，我甚至没有看到 println 在 SessionWindowProcessor.scala 开火。是不是我做错了什么导致了我的开窗失败？