Apache Flink CEP检测指定事件缺失超时未触发问题求助

• 模式定义错误，无法触发超时逻辑
  Flink CEP的超时仅针对存在部分匹配的模式生效。你当前定义的Pattern仅包含单个匹配节点，只要匹配到符合条件的事件就会直接进入正常匹配逻辑，不存在「匹配了一半、等待剩余条件满足」的中间状态，因此永远不会触发超时回调。你要检测事件缺失，需要定义至少两个节点的连续模式：先匹配到一个目标事件，再等待下一个目标事件在指定窗口内到达，未到达则触发超时。
• 未正确获取超时侧输出流
  你调用flatSelect后返回的DataStream是正常匹配的主流，超时事件全部写入你定义的OutputTag对应的侧输出流中，当前你直接返回主流，自然拿不到超时结果。
• 测试数据源逻辑不符合验证要求
  你当前的测试数据中仅存在1条id为USD的事件，模式匹配到这条事件后就直接结束，没有后续的等待逻辑；且source发送完所有事件后没有发送最大水印，可能导致最后未触发的超时事件被丢弃。

1. 修正Pattern定义 #

将模式修改为连续匹配逻辑，检测两个USD事件的间隔是否超过3000ms：

Pattern<Event, ?> myPattern = Pattern.<Event>begin("first_usd")
        .subtype(CurrencyEvent.class)
        .where(new SimpleCondition<CurrencyEvent>() {
            @Override
            public boolean filter(CurrencyEvent currencyEvent) throws Exception {
                return currencyEvent.getId().equalsIgnoreCase("usd");
            }
        })
        .next("next_usd")
        .subtype(CurrencyEvent.class)
        .where(new SimpleCondition<CurrencyEvent>() {
            @Override
            public boolean filter(CurrencyEvent currencyEvent) throws Exception {
                return currencyEvent.getId().equalsIgnoreCase("usd");
            }
        })
        .within(Time.milliseconds(3000L));

如果需要检测作业启动后3秒内没有第一个USD事件也触发报警，可以额外增加初始超时逻辑，或者用notFollowedBy配合时间窗口实现。

2. 正确获取侧输出流 #

调用flatSelect后，通过侧输出标签获取超时报警流，同时不要忘记在超时回调中调用collect输出报警对象：

PatternFlatTimeoutFunction<Event, Alarm> eventAlarmTimeoutPatternFunction = (patterns, timestamp, ctx) -> {
    System.out.println("New alarm, since after 3 seconds an event with id=usd is not detected");
    ctx.collect(new Alarm()); // 输出报警对象到侧输出流
};

SingleOutputStreamOperator<Alarm> mainStream = patternStream.flatSelect(
        tag,
        eventAlarmTimeoutPatternFunction,
        TypeInformation.of(Alarm.class),
        eventAlarmPatternSelectFunction
);
// 拿到超时报警流，打印或后续处理
DataStream<Alarm> alarmStream = mainStream.getSideOutput(tag);
alarmStream.print();
return alarmStream;

3. 调整测试数据源 #

首先增加多条USD测试事件，间隔大于3000ms；其次在source发送完所有事件后，发送最大水印触发所有剩余超时：

@Override
public void run(SourceContext<Event> sourceContext) throws InterruptedException {
    long timestamp  = this.initialTimestamp;
    for(Event event: this.events){
        sourceContext.collectWithTimestamp(event, timestamp);
        sourceContext.emitWatermark(new Watermark(timestamp));
        timestamp+=this.timeBetweenInMillis;
    }
    // 发送最大水印，触发所有未完成的超时逻辑
    sourceContext.emitWatermark(Watermark.MAX_WATERMARK);
}

内容的提问来源于stack exchange，提问作者nach0