title: Apache Flink CEP how to detect if event did not occur within x seconds?
body: <p>For example A should be followed by B within 10 seconds. I know how to track if this DID occur (.next, .within), but I want to send an alert if B never happened within the window.</p>
  
  <pre><code>    public static void main(String[] args) throws Exception {
          final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  
          // checkpointing is required for exactly-once or at-least-once guarantees
  //      env.enableCheckpointing(1000);
  
          final RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder()
              .setHost("localhost")
              .setPort(5672)
              .setVirtualHost("/")
              .setUserName("guest")
              .setPassword("guest")
              .build();
  
          final DataStream&lt;String&gt; inputStream = env
              .addSource(new RMQSource&lt;String&gt;(
                  connectionConfig,               // config for the RabbitMQ connection
                  "cep",                          // name of the RabbitMQ queue to consume
                  true,                           // use correlation ids; can be false if only at-least-once is required
                  new SimpleStringSchema()))      // deserialization schema to turn messages into Java objects
              .setParallelism(1);                 // non-parallel source is only required for exactly-once
  
          inputStream.print();
  
          Pattern&lt;String, ?&gt; simplePattern =
                  Pattern.&lt;String&gt;begin("start")
                      .where(new SimpleCondition&lt;String&gt;() {
                          @Override
                          public boolean filter(String event) {
                              return event.equals("A");
                          }
                      })
                      .next("end")
                      .where(new SimpleCondition&lt;String&gt;() {
                          @Override
                          public boolean filter(String event) {
                              return event.equals("B");
                          }
                      });
  
          PatternStream&lt;String&gt; timedOutPatternStream = CEP.pattern(inputStream, simplePattern.within(Time.seconds(10)));
          OutputTag&lt;String&gt; timedout = new OutputTag&lt;String&gt;("timedout"){};
          SingleOutputStreamOperator&lt;String&gt; timedOutNotificationsStream = timedOutPatternStream.flatSelect(
              timedout,
              new TimedOut&lt;String&gt;(),
              new FlatSelectNothing&lt;String&gt;()
          );
          timedOutNotificationsStream.getSideOutput(timedout).print();
  
          env.execute("mynotification");
      }
  
  public static class TimedOut&lt;String&gt; implements PatternFlatTimeoutFunction&lt;String, String&gt; {
      @Override
      public void timeout(Map&lt;java.lang.String, List&lt;String&gt;&gt; pattern, long timeoutTimestamp, Collector&lt;String&gt; out) throws Exception {
          out.collect((String) "LATE!");
      }
  }
  
  public static class FlatSelectNothing&lt;T&gt; implements PatternFlatSelectFunction&lt;T, T&gt; {
      @Override
      public void flatSelect(Map&lt;String, List&lt;T&gt;&gt; pattern, Collector&lt;T&gt; collector) {}
  }
  </code></pre>
  
  <p>Actual behavior:</p>
  
  <pre><code>publish "A"
  (wait 5 seconds)
  publish "B"
  =&gt; (no alert)
  
  publish "A"
  (wait 10 seconds)
  =&gt; (no alert, but should be)
  
  publish "A"
  (wait 10 seconds)
  publish "B"
  =&gt; "LATE!"
  </code></pre>
  
  <p>Expected behavior:</p>
  
  <pre><code>publish "A"
  (wait 10 seconds)
  =&gt; "LATE!"
  </code></pre>