BroadcastStream导致watermark不更新的问题
业务场景
有业务数据流businessDataStream和规则数据源ruleDataSource, businessDataStream数据来自Kafka，ruleDataSource定时从数据库查询需要更新的规则并广播(broadcast)数据流为ruleBroadcastStream。使用businessDataStream(或者keyBy分组后得到的KeyedStream).connect(ruleBroadcastStream)将两个流汇聚成BroadcastConnectedStream。随后执行算子(BroadcastProcessFunction或KeyedBroadcastProcessFunction)。
示意图如下:
代码示例如下:
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之前多次执行，重新分配
