Apache Flink CEP , pattern not matching

[#1]

Hello Experts,

I am new to Flink CEP and trying to test basic things - In below code my expectation is all the input should matched in patter and should print as matched result. But somehow nothing is matching ('matechedStream.print()') any idea about the reason ?

Any suggestion/help would be much appreciated.

package com.o9.flink;
import com.o9.flink.asyncio.DemandSupply;
import org.apache.flink.cep.CEP;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
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;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;


public class DemandSupplyPattern {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStream<String> keyedInputStream = env.fromElements("AAA","BBB","CCC");

        Pattern<String, ?> dspattern = Pattern.<String>begin("start");

        PatternStream<String> patternStream = CEP.pattern(keyedInputStream, dspattern);
        DataStream<String> matechedStream =  patternStream.process(new PatternProcessFunction<String, String>() {
            @Override
            public void processMatch(Map<String, List<String>> map, Context context, Collector<String> collector) throws Exception {
                collector.collect(map.get("start").toString());
            }
        });

        matechedStream.print();

        env.execute("DemandSupply-CEP");
    }
}

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-runtime</artifactId>
        <version>${flink.version}</version>
        <!--<scope>test</scope>-->
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-core</artifactId>
        <version>${flink.version}</version>
        <!--<scope>provided</scope>-->
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-avro</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java</artifactId>
        <version>${flink.version}</version>
        <!--<scope>provided</scope>-->
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-statebackend-rocksdb</artifactId>
        <version>${flink.version}</version>
        <!--<scope>test</scope>-->
    </dependency>
    <dependency>
        <groupId>org.asynchttpclient</groupId>
        <artifactId>async-http-client</artifactId>
        <version>2.2.0</version>
    </dependency>

    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-cep</artifactId>
        <version>${flink.version}</version>
        <!--<scope>provided</scope>-->
    </dependency>


</dependencies>

[#2]

@hegdemahendra ,

You need to set the time characteristic of the stream for it to work. For example, try setting it to processing time, as shown below:

		DataStream<String> matechedStream =  patternStream
				.inProcessingTime()
				.process(new PatternProcessFunction<String, String>() {
					@Override
					public void processMatch(Map<String, List<String>> map, Context context, Collector<String> collector) throws Exception {
						collector.collect(map.get("start").toString());
					}
				}
		);

Cheers,

André