cep更适用于重复的模式，在这里它实际上并不起作用。您所说的跳过策略实际上被称为“匹配后跳过策略”，这意味着您要匹配多个模式。
我认为更好的方法是使用过程函数，比如：

class AlertUserOnce extends ProcessFunction[element, element] {

  lazy val turnOff: ValueState[Boolean] = getRuntimeContext
     .getState(new ValueStateDescriptor[Boolean]("turn-function-off", classOf[Boolean]))

  override def processElement(value: element, ctx: Context, out: Collector[element]): Unit = {
   if (turnOff.value == null && element.downloads > 1000) {
     alertUser(value)
     turnOff.update(true)
    }
    out.collect(value)
  }
}

...

stream.keyBy(...).process(new AlertUserOnce)