Looks like a bug in flink-training-exercises for the CEP example

Question

I got a example for the CEP in the following URL https://github.com/dataArtisans/flink-training-exercises/blob/master/src/main/java/com/dataartisans/flinktraining/exercises/datastream_java/cep/LongRides.java

And the "goal for this exercise is to emit START events for taxi rides that have not been matched by an END event during the first 2 hours of the ride." However from the code below, it seems get a pattern to find rides have been completed in 2 hours instead of have NOT been completed in 2 hours.

It looks like the pattern firstly find the Start event , then find the End Event(!ride.isStart), and within 2 hours, so doesn't it explains as a pattern to find rides have been completed in 2 hours?

Pattern<TaxiRide, TaxiRide> completedRides =
            Pattern.<TaxiRide>begin("start")
                    .where(new SimpleCondition<TaxiRide>() {
                        @Override
                        public boolean filter(TaxiRide ride) throws Exception {
                            return ride.isStart;
                        }
                    })
                    .next("end")
                    .where(new SimpleCondition<TaxiRide>() {
                        @Override
                        public boolean filter(TaxiRide ride) throws Exception {
                            return !ride.isStart;
                        }
                    });

    // We want to find rides that have NOT been completed within 120 minutes
    PatternStream<TaxiRide> patternStream = CEP.pattern(keyedRides, completedRides.within(Time.minutes(120)));

Pay extra attention to which matches are further processed. It is the timed out patterns that are printed, so there is no mistake in this example. — Dawid Wysakowicz

David Anderson David Anderson · Accepted Answer · 2018-06-01T08:09:22

I've improved the comment in the sample solution to make this clearer.

// We want to find rides that have NOT been completed within 120 minutes.
// This pattern matches rides that ARE completed.
// Below we will ignore rides that match this pattern, and emit those that timeout.
PatternStream<TaxiRide> patternStream = CEP.pattern(keyedRides, completedRides.within(Time.minutes(120)));

OutputTag<TaxiRide> timedout = new OutputTag<TaxiRide>("timedout"){};

SingleOutputStreamOperator<TaxiRide> longRides = patternStream.flatSelect(
        timedout,
        new TaxiRideTimedOut<TaxiRide>(),
        new FlatSelectNothing<TaxiRide>()
);

Looks like a bug in flink-training-exercises for the CEP example

1 Answers