从kafka主题使用空闲分区时，水印生成在本地/独立模式下的工作方式有所不同吗？

Question

我正在考虑这个水印：

class MyWatermarker(val maxTimeLag: Long = 0)
    extends AssignerWithPeriodicWatermarks[MyEvent] {
  var maxTs: Long = 0

  override def extractTimestamp(e: MyEvent, previousElementTimestamp: Long): Long = {
    val timestamp = e.timestamp
    maxTs = maxTs.max(timestamp)
    timestamp
  }

  override def getCurrentWatermark: WatermarkOld = {
    println(s"event watermark: ${maxTs - maxTimeLag}")
    new WatermarkOld(maxTs - maxTimeLag)
  }

基础事件来自 kafka 源，然后传递给流程函数。实现与问题无关，我将只分享相关的位：

  override def processElement(
    event: MyEvent,
    ctx: KeyedProcessFunction[String, MyEvent, MyEvent]#Context,
    out: Collector[StreamEvent]
  ): Unit = {
    println(
      s"In process function, got event: $event, ctx.timestamp: ${ctx.timestamp}, currentWatermark: ${ctx.timerService.currentWatermark}"
    )
  ...
  }

当我在一个真实的kubernetes集群上运行这个应用程序时，使用了一个具有空闲分区的kafka源主题，水印如预期的那样保持为0:

In process function, got event: xxx, ctx.timestamp: 1601475710619, currentWatermark: 0

我还可以看到水印中生成的这些日志：

event watermark: 1601475710619
event watermark: 0
event watermark: 1601475710619
event watermark: 0

有趣的是，当我在IntelliJ上本地运行同一个应用程序，并且同一主题也有空闲的kafka分区时，我还从水印中获得了上述日志，水印在0和最近接收到的元素的ts之间振荡，因为maxLag=0。然而，对于我来说，出乎意料的是，来自process函数的日志显示水印仍在前进：

In process function, got event: xxx, ctx.timestamp: 1601475710619, currentWatermark: 1601475710618

为什么会发生这种情况？仅供参考，我使用的是 Flink 1.10，环境并行度设置为 2，并且在这两种情况下都有事件时间语义。

Answer 1

如果要使用每个分区的水印，可以直接在Flink Kafka消费者[1]上调用assignTimestampsAndWatermarks，那么我相信空闲分区将始终保留整个水印。

使用每个分区水印，每个Kafka源任务将分别将水印应用到它正在处理的每个分区，然后发出每个分区水印中最小的水印作为其水印。因此，至少有一个Kafka源任务的水印为0，假设您在水印之后和进程函数之前有一个keyBy，这将在进程函数处保留水印。

否则，如果将水印应用于 Kafka 源任务的输出，则水印任务的水印是否为 0 取决于其相应的 Kafka 源任务是否具有任何非空闲分区。如果将分区分配给实例不是确定性的，这可以解释为什么您在IntelliJ中看到不同的结果。

请注意，在Flink 1.11 [2]中，对空闲资源的处理已经重新进行了，与此相关的错误修复仍然悬而未决[3]。

[1]https://ci . Apache . org/projects/flink/flink-docs-release-1.10/dev/connectors/Kafka . html # Kafka-consumers-and-timestamp-extraction watermark-emission。< br >[2]https://ci . Apache . org/projects/flink/flink-docs-release-1.11/dev/event _ timestamps _ watermark s . html #处理-idle-sources。[3]https://issues.apache.org/jira/browse/FLINK-18934