flink-cep

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class LoginFailureDetection {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<LoginEvent> loginEvents = env.fromElements(
            new LoginEvent("user1", true),
            new LoginEvent("user1", false),
            new LoginEvent("user1", false),
            new LoginEvent("user1", false)
        );

        Pattern<LoginEvent, ?> pattern = Pattern.<LoginEvent>begin("first")
            .where(new SimpleCondition<LoginEvent>() {
                @Override
                public boolean filter(LoginEvent event) {
                    return !event.isSuccess();
                }
            })
            .next("second")
            .where(new SimpleCondition<LoginEvent>() {
                @Override
                public boolean filter(LoginEvent event) {
                    return !event.isSuccess();
                }
            })
            .next("third")
            .where(new SimpleCondition<LoginEvent>() {
                @Override
                public boolean filter(LoginEvent event) {
                    return !event.isSuccess();
                }
            });

        PatternStream<LoginEvent> patternStream = CEP.pattern(loginEvents, pattern);

        patternStream.select(pattern -> {
            System.out.println("User " + pattern.get("first").getUserId() + " has three consecutive failed login attempts.");
            return null;
        }).print();

        env.execute("Login Failure Detection");
    }

    public static class LoginEvent {
        private String userId;
        private boolean success;

        public LoginEvent(String userId, boolean success) {
            this.userId = userId;
            this.success = success;
        }

        public String getUserId() {
            return userId;
        }

        public boolean isSuccess() {
            return success;
        }
    }
}