闪烁是否保存带有水印的关闭事件时间窗口的历史？

Question

我有一个flink任务，它使用带事件时间和水印的键控翻滚窗口来聚合数据。

我的问题是，flink是否保持着他已经关闭的窗口的状态？否则，我没有其他解释为什么属于以前从未打开过的窗口的事件会打开一个窗口而不会立即删除它。

假设我们的窗口是1小时，禁止自动关闭是10分钟

让我们举个例子:

event1=（"2022-01-01T08:25:00Z"）=

event2=（"2022-01-01T09:25:00Z"）=

event3=（"2022-01-01T05:25:00Z"）=

event 4 =(" 2022-01-01t 05:40:00Z ")=

    val stream = env
      .fromSource(
        kafkaSource,
        WatermarkStrategy
          .forBoundedOutOfOrderness[(String, EnrichedProcess, KafkaHeaders)](Duration.ofMinutes(outOfOrdernessMinutes)) //Watermark max time for late events
          .withIdleness(Duration.ofSeconds(idleness))
          .withTimestampAssigner(new SerializableTimestampAssigner[(String, EnrichedProcess, KafkaHeaders)] {
            override def extractTimestamp(element: (String, EnrichedProcess, KafkaHeaders), recordTimestamp: Long)
                : Long = {
              logger.info(
                LogMessage(
                  element._3.orgId,
                  s"Received incoming EnrichedProcess update_time: ${element._3.updateTime}, process time: ${recordTimestamp.asDate}",
                  element._3.flowId
                )
              )
              element._3.updateTime.asEpoch
            }
          }),
        s"Source - $kConsumeTopic"
      )

    stream
      .keyBy(element => (element._2.orgId -> element._2.procUid))                                                                     
      .window(TumblingEventTimeWindows.of(Time.hours(tumblingWindowHours), Time.minutes(windowStartingOffsetMinutes)))
      .reduce(new ReduceFunc)                                                                                         
      .name("Aggregated EnrichedProcess")
      .sinkTo(kafkaConnector.createKafkaSink(kProducerServers, kProduceTopic))
      .name(s"Sink -> $kProduceTopic")

编辑：我测试的方式是使用docker compose进行集成测试。我正在为Kafka制作一个事件=

当我在发送事件3和事件4之间放置30秒的睡眠时，事件4被丢弃了。这就是我所期望的行为。

    val producer = new Producer(producerTopic)

    val consumer = new Consumer(consumerTopic, groupId)
    producer.send(event1)
    producer.send(event2)
    Thread.sleep(30000)
    producer.send(event3)
    Thread.sleep(30000)
    producer.send(event4)

    val received: Iterable[(String, EnrichedProcess)] = consumer.receive[EnrichedProcess]()

但是现在更奇怪的是，为什么当我把睡眠时间设置为10秒而不是30秒时，我只收到第一种情况(水印应该已经更新了(定期水印生成器的默认值是200毫秒)

Answer 1

执行摘要:

使用Flink的基于事件时间的逻辑中的不确定性来自于处理时间与事件时间的混合——就像周期性水印生成器和空闲检测一样。只有在没有延迟事件或空闲源的情况下，才能确保确定性结果。

更多详情：

虽然你会期待

event3 = ("2022-01-01T05:25:00Z")

要迟到，只有当足够大的水印设法先到达时，它才会真正迟到。使用forBoundedOutOfOrness策略不能保证这一点——这是一个周期性的水印生成器，每200毫秒产生一次水印。因此，可能是基于event2的时间戳的水印在event3和event4之间产生。

这是一种可能的解释。根据具体情况，可能还有其他情况。例如，随着所有这些Hibernate状态的进行，水印生成器的一个并行度实例至少空闲了一分钟，这可能涉及产生正在观察到的行为（取决于空闲值等）。

更多背景：

其中平行度为

具有多个输入信道(例如键控窗口)的操作者将通过将(来自所有非空闲信道的)输入水印中的最小值作为他们自己的水印来组合这些水印。

回答原问题：

Flink是否保留已经关闭的窗口的状态？不会。一旦允许的延迟(如果有)过期，事件时间窗口的状态将被清除。