title: Flinkcep does not process the whole datastream
body: <p>I'm writing a project in which I want to receive events from a kafka source and apply a cep pattern to find the possible matches.</p>
  <pre><code>StreamExecutionEnvironment qEnv = StreamExecutionEnvironment.getExecutionEnvironment();
          KafkaSource&lt;String&gt; qSource = KafkaSource.&lt;String&gt;builder()
                  .setBootstrapServers(broker1.getConfig())
                  .setTopics(Query.setQueryTopic(newQuery.getID()))
                  .setGroupId(&quot;group2&quot;)
                  .setStartingOffsets(OffsetsInitializer.earliest())
                  .setValueOnlyDeserializer(new SimpleStringSchema())
                  .build();
  
          DataStream&lt;String&gt; stringQueryInput = qEnv.fromSource(qSource, WatermarkStrategy.noWatermarks(), &quot;KafkaTopic&quot;);
  
          DataStream&lt;DataEvent&gt; qInput = stringQueryInput.map(new MapFunction&lt;String, DataEvent&gt;() {
  
              @Override
              public DataEvent map(String s) throws Exception {
                  DataEvent d = ExtractDataEvent(s);
  
                  return d;
  
              }
          }).assignTimestampsAndWatermarks(WatermarkStrategy.noWatermarks());
  
  //qInput.print();
  
          Pattern&lt;DataEvent, ?&gt; qPattern= Pattern.&lt;DataEvent&gt;begin(&quot;start&quot;).where(new SimpleCondition&lt;DataEvent&gt;() {
              @Override
              public boolean filter(DataEvent e) throws Exception {
  
  
                  SE de0 = newQuery.getEtList().get(0);
  
                  System.out.println(&quot;SE : &quot;+de0.getEventType() + &quot; + current event &quot; + DataEvent.DataEventToString(e));
                  return e.getType().equals(de0.getEventType()) &amp;&amp; SE.checkThreshold(de0.getOperator(), e.getValue(),de0.getValue()) ;
                  
              }
          });
  
                  DataStream&lt;DataEvent&gt; result =  CEP.pattern(qInput, qPattern).inProcessingTime().process(new PatternProcessFunction&lt;DataEvent, DataEvent&gt;() {
  
              @Override
              public void processMatch(Map&lt;String, List&lt;DataEvent&gt;&gt; map, Context context, Collector&lt;DataEvent&gt; collector) throws Exception {
                  System.out.println(&quot;map event list : &quot; + DataEvent.DataEventToString(map.get(&quot;start&quot;).get(0)));
                  //System.out.println(&quot;map event list : &quot; + DataEvent.DataEventToString(map.get(&quot;start&quot;).get(1)));
                  ArrayList&lt;AttributeValue&gt; avc=new ArrayList&lt;AttributeValue&gt;();
                  avc.add(new AttributeValue(&quot;QueryID&quot;, newQuery.getID()));
                  avc.add(new AttributeValue(&quot;TimeStamp&quot;, String.valueOf(context.timestamp())));
                  ArrayList&lt;DataEvent&gt; uncertainties = new ArrayList&lt;&gt;();
  
                  uncertainties.add(map.get(&quot;start&quot;).get(0));
                  for (int j=1;j&gt;map.size();j++) {
                      uncertainties.add(map.get(String.valueOf(j)).get(0));
                  }
  
                  avc.add(new AttributeValue(&quot;Uncertainty&quot;, DataEvent.CumulativeUncertainty(uncertainties)));
                  //avc.add(new AttributeValue(&quot;EndToEndLatency&quot;, String.valueOf(context.timestamp()-Long.parseLong(map.get(&quot;start&quot;).get(0).getTimeStamp()))));
  
                  DataEvent devent = new DataEvent(avc);
                  System.out.println(&quot;process match &quot;+DataEvent.DataEventToString(devent));
                  collector.collect(devent);
                  System.out.println(collector);
                  
              }
          });
  
  </code></pre>
  <p>My problem is when I run the code with a query that receives 3 distinct events from the kafka source and then apply the pattern over these event stream, only the last event in the eventstream is processed and the previous events are not checked.
  The next problem is the result of process match function will be fed again to pattern matching in the input element (map).
  The result is something like this:</p>
  <pre><code>[Map -&gt; Sink Unnamed (1/6)#0] INFO org.apache.flink.runtime.taskmanager.Task - Map -&gt; Sink Unnamed (1/6)#0 (702146db2e5a34d4b727441e028b0810) switched from INITIALIZING to RUNNING.
  [Map -&gt; Sink Unnamed (2/6)#0] INFO org.apache.flink.runtime.taskmanager.Task - Map -&gt; Sink Unnamed (2/6)#0 (241186254bed8dace8ea516fa65aae64) switched from INITIALIZING to RUNNING.
  [Map -&gt; Sink Unnamed (3/6)#0] INFO org.apache.flink.runtime.taskmanager.Task - Map -&gt; Sink Unnamed (3/6)#0 (758427479573e154eed7d040aa244e32) switched from INITIALIZING to RUNNING.
  [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Map -&gt; Sink Unnamed (2/6) (241186254bed8dace8ea516fa65aae64) switched from INITIALIZING to RUNNING.
  [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Map -&gt; Sink Unnamed (3/6) (758427479573e154eed7d040aa244e32) switched from INITIALIZING to RUNNING.
  [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Map -&gt; Sink Unnamed (1/6) (702146db2e5a34d4b727441e028b0810) switched from INITIALIZING to RUNNING.
  [GlobalCepOperator (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - GlobalCepOperator (1/1)#0 (3b2fc9bd1850f0fbfb9befea28066507) switched from INITIALIZING to RUNNING.
  [flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - GlobalCepOperator (1/1) (3b2fc9bd1850f0fbfb9befea28066507) switched from INITIALIZING to RUNNING.
  [Map -&gt; Sink Unnamed (5/6)#0] INFO org.apache.flink.runtime.taskmanager.Task - Map -&gt; Sink Unnamed (5/6)#0 (80819f5940c445fb63425f7b1cca7c06) switched from INITIALIZING to RUNNING.
  [Source Data Fetcher for Source: KafkaTopic -&gt; Map -&gt; Timestamps/Watermarks -&gt; Sink: Print to Std. Out (4/6)#0] INFO org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher - Starting split fetcher 0
  [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Map -&gt; Sink Unnamed (5/6) (80819f5940c445fb63425f7b1cca7c06) switched from INITIALIZING to RUNNING.
  [Source Data Fetcher for Source: KafkaTopic -&gt; Map -&gt; Timestamps/Watermarks -&gt; Sink: Print to Std. Out (4/6)#0] INFO org.apache.kafka.clients.consumer.KafkaConsumer - [Consumer clientId=group2-3, groupId=group2] Subscribed to partition(s): q1Topic-0
  [Source Data Fetcher for Source: KafkaTopic -&gt; Map -&gt; Timestamps/Watermarks -&gt; Sink: Print to Std. Out (4/6)#0] INFO org.apache.kafka.clients.consumer.internals.SubscriptionState - [Consumer clientId=group2-3, groupId=group2] Seeking to EARLIEST offset of partition q1Topic-0
  [Source Data Fetcher for Source: KafkaTopic -&gt; Map -&gt; Timestamps/Watermarks -&gt; Sink: Print to Std. Out (4/6)#0] INFO org.apache.kafka.clients.Metadata - [Consumer clientId=group2-3, groupId=group2] Cluster ID: ILvuvQJcQNez2yJJJdjwyg
  [Source Data Fetcher for Source: KafkaTopic -&gt; Map -&gt; Timestamps/Watermarks -&gt; Sink: Print to Std. Out (4/6)#0] INFO org.apache.kafka.clients.consumer.internals.SubscriptionState - [Consumer clientId=group2-3, groupId=group2] Resetting offset for partition q1Topic-0 to position FetchPosition{offset=0, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[majidlotfian-VirtualBox:9092 (id: 0 rack: null)], epoch=0}}.
  4&gt; Events.DataEvent@4c2f64d6
  4&gt; Events.DataEvent@61af870b
  4&gt; Events.DataEvent@ca385a3
  SE : Temperature + current event ProducerID p2 Type Temperature Value 19.4 GroundTruth 20.0 TimeStamp 1646053121798 SendingLatency 0.0 
  event value : 19.4 threshold: 18
  map event list : ProducerID p2 Type Temperature Value 19.4 GroundTruth 20.0 TimeStamp 1646053121798 SendingLatency 0.0 
  process match QueryID q1 TimeStamp 1646053125270 Uncertainty NaN 
  org.apache.flink.streaming.api.operators.TimestampedCollector@5105b38
  SE : Temperature + current event QueryID q1 TimeStamp 1646053125270 Uncertainty NaN 
  SE : Temperature + current event QueryID q1 TimeStamp 1646053125270 Uncertainty NaN 
  5&gt; Events.DataEvent@27006cee
  QueryID q1 TimeStamp 1646053125270 Uncertainty NaN 
  
  </code></pre>
  <p>It seems for the first time, the processMatch method applies the pattern over the last element of the data stream and then it applies the pattern on the newly created event by itself.
  Can someone tell me what is going on in this code?</p>