BroadcastStream导致watermark不更新的问题

BroadcastStream导致watermark不更新的问题

业务场景

有业务数据流businessDataStream和规则数据源ruleDataSource, businessDataStream数据来自Kafka，ruleDataSource定时从数据库查询需要更新的规则并广播(broadcast)数据流为ruleBroadcastStream。使用businessDataStream(或者keyBy分组后得到的KeyedStream).connect(ruleBroadcastStream)将两个流汇聚成BroadcastConnectedStream。随后执行算子(BroadcastProcessFunction或KeyedBroadcastProcessFunction)。

示意图如下:

image-20200629223054640.png

代码示例如下:

BroadcastStream<Rule> ruleBroadcastStream = env.addSource(new RuleSource()).broadcast(Descriptors.RuleDesc);

// 业务数据流，分配Timestamp和watermark
DataStream businessDataStream = env.addSource(new BusinessDataSource()).assignTimestampsAndWatermarks(...)

businessDataStream
    .keyBy(...)
    .connect(ruleBroadcastStream)
    .process(new MyBroadcastProcessFunction());

出现问题

查看FlinkWebUI, 算子MyBroadcastProcessFunction的Watermarks一栏的所有SubTask都实现: No Watermark。确认了业务数据一直是有数据流入的，这就奇怪了，为什么watermark会不更新呢？同样通过该算子的后续算子的Watermarks也都为No Watermark。而且伴随着也出现了后续的TimeWindow算子时间窗口到期后也不触发的问题。

分析问题

后续的TimeWindow算子时间窗口到期后也不触发这个问题的原因是由于MyBroadcastProcessFunction之后的Watermark一直没有更新的缘故，因为执行环境设置了

 env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

因此时间窗口的触发依赖于Trigger类中的onEventTime方法，该方法依赖于窗口算子的Watermark, watermark不更新，onEventTime方法也就不会触发了。

因此问题专注定位到为什么MyBroadcastProcessFunction的watermark不更新。

每个Operator将数据输出到下游的时候都会分发Watermark，执行类似如下代码:

out.emitWatermark(watermark);

...

该方法会执行:

operator.processWatermark(mark);

而实现接口TwoInputStreamOperator.java的Operator会有两个处理方法，分别为:

    /**
     * Processes a {@link Watermark} that arrived on the first input of this two-input operator.
     * This method is guaranteed to not be called concurrently with other methods of the operator.
     *
     * @see org.apache.flink.streaming.api.watermark.Watermark
     */
    void processWatermark1(Watermark mark) throws Exception;

    /**
     * Processes a {@link Watermark} that arrived on the second input of this two-input operator.
     * This method is guaranteed to not be called concurrently with other methods of the operator.
     *
     * @see org.apache.flink.streaming.api.watermark.Watermark
     */
    void processWatermark2(Watermark mark) throws Exception;

它的实现类之一CoBroadcastWithKeyedOperator.java,它是父类AbstractStreamOperator.java中实现了以上2个方法

    private long combinedWatermark = Long.MIN_VALUE;
    private long input1Watermark = Long.MIN_VALUE;
    private long input2Watermark = Long.MIN_VALUE;
//...

public void processWatermark1(Watermark mark) throws Exception {
   input1Watermark = mark.getTimestamp();
   long newMin = Math.min(input1Watermark, input2Watermark);
   if (newMin > combinedWatermark) {
      combinedWatermark = newMin;
      processWatermark(new Watermark(combinedWatermark));
   }
}

public void processWatermark2(Watermark mark) throws Exception {
   input2Watermark = mark.getTimestamp();
   long newMin = Math.min(input1Watermark, input2Watermark);
   if (newMin > combinedWatermark) {
      combinedWatermark = newMin;
      processWatermark(new Watermark(combinedWatermark));
   }
}

public void processWatermark(Watermark mark) throws Exception {
   if (timeServiceManager != null) {
      timeServiceManager.advanceWatermark(mark);
   }
   output.emitWatermark(mark);
}

经过相关源码追踪分析得知:当主数据流有数据的时候会执行processWatermark1方法，当规则数据流有数据的时候会执行processWatermark2方法。且由这两个方法的逻辑得知，要执行output.emitWatermark(mark)需要2个数据流中最小的watermark值大于之前的watermark值。要达到这个条件需要两个数据流都有watermark更新才行。

可是规则数据流并不是经常会有数据产生，怎么办呢？

一时没有找到解决办法，通过google查找"broadcastStream watermark"之类的关键词找到stackoverflow上有类似的问题: https://stackoverflow.com/questions/57585528/timestamp-watermark-assigning-for-two-input-streams-later-connected-for-dynam

于是恍然大悟。自己怎么没有想到呢，input2Watermark一开始就给它发来一个最大的watermark不就行了，因为取决定作用的是input1Watermark和input2Watermark的最小值，这样整个Operator的值不就根据业务数据流的watermark来更新了吗，达到了我想要的目的。

解决问题

新增一个AssignerWithPeriodicWatermarks类

public class QueryStreamAssigner<T> implements AssignerWithPeriodicWatermarks<T> {

    @Nullable
    @Override
    public Watermark getCurrentWatermark() {
        return Watermark.MAX_WATERMARK;
    }

    @Override
    public long extractTimestamp(T element, long previousElementTimestamp) {
        return 0;
    }
}

在规则数据流ruleBroadcastStream的时候执行assignTimestampsAndWatermarks方法，代码如下:

BroadcastStream<Rule> ruleBroadcastStream = env.addSource(new RuleSource())
.assignTimestampsAndWatermarks(new QueryStreamAssigner<>())
.broadcast(Descriptors.RuleDesc);

// 业务数据流，分配Timestamp和watermark
DataStream businessDataStream = env.addSource(new BusinessDataSource()).assignTimestampsAndWatermarks(...)

businessDataStream
    .keyBy(...)
    .connect(ruleBroadcastStream)
    .process(new MyBroadcastProcessFunction());

问题解决！

总结

• watermark决定onTimer的触发
• 有2个输入流的operator, 它的watermark取2个流的watermark最小值，将其中一个流的watermark取Int.Max可忽略它的影响而由另一个流来更新watermark
• assignTimestampsAndWatermarks可在source之后,sink之前多次执行，重新分配