Apache Flink TwoInputStreamOP运算符的水印行为

Question

有2个指定了时间戳的数据流和定义如下的水印生成器。

val streamA: DataStream[A] = kafkaStreamASourceOutput.assignTimestampsAndWatermarks(
        WatermarkStrategy
          .forBoundedOutOfOrderness[A](Duration.ofSeconds(0))
          .withTimestampAssigner(new SerializableTimestampAssigner[A] {
            override def extractTimestamp(element: A, recordTimestamp: Long): Long = {
              element.lastUpdatedMs
            }
          })
      )

val streamA: DataStream[B] = kafkaStreamBSourceOutput.assignTimestampsAndWatermarks(
        WatermarkStrategy
          .forBoundedOutOfOrderness[B](Duration.ofSeconds(0))
          .withTimestampAssigner(new SerializableTimestampAssigner[B] {
            override def extractTimestamp(element: B, recordTimestamp: Long): Long = {
              element.lastUpdatedMs
            }
          })
      )

当这两个流在一个操作符中连接时，来自streamA或streamB的最小水印作为连接操作符的水印。

class CombineAB extends CoProcessFunction[A, B, C] {
   override def processElement1(elem: A, ctx:Context, out: Collector[C]) {
        out.collect(C(elem.x, elem.y, time.Now()))
   }
   override def processElement2(elem: B, ctx:Context, out: Collector[C]) {
        out.collect(C(elem.x, elem.y, time.Now()))
   }
}

val streamC: DataStream[C] = streamA.connect(streamB)
      .process(new CombineAB)

组合B运算符的水印是A或B中的最小值。基于C类元素是否标记为延迟。

但是，由于我们没有附加任何分配给C的时间戳，这是否意味着AssociineAB运算符中的任何元素都没有被标记为延迟？因此在C上窗口不会有任何延迟记录被删除？

假设我们将分配的时间戳和水印生成器附加到C，如下所示，那么这是否意味着来自A和B的水印被完全忽略，AssociineAB的水印仅取决于C的时间戳字段和C定义的延迟。

     streamC.assignTimestampsAndWatermarks(
        WatermarkStrategy
          .forBoundedOutOfOrderness[C](Duration.ofSeconds(0))
          .withTimestampAssigner(new SerializableTimestampAssigner[C] {
            override def extractTimestamp(element: C, recordTimestamp: Long): Long = {
              element.updatedTime
            }
          })
      )

是否有一种方法可以将时间戳赋值器附加到C，并且组合B的水印仍然是a和B的最小值，并且C的元素根据C的指定时间戳和组合B的wartermark标记得较晚

更新：CombineAB的优化实施

Answer 1

以下几点：

forBoundedOutOfOrness[A]（Duration.of秒（0））是不寻常的。任何乱序事件都会迟到。为什么不使用forMonotonousTimestamps（）？

CombineAB产生的记录将有时间戳；无需对此流应用assignTimestampsAndWatermarks。收集器生成的任何记录的时间戳都是传入记录的时间戳。

如果在流C上调用assignTimestampsAndWatermarks，传入的水印将被过滤掉，您需要生成新的水印。