You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by dw...@apache.org on 2017/09/17 16:37:38 UTC
flink git commit: [hotfix] [cep] Fix afterMatchStrategy parameter
missing issue
Repository: flink
Updated Branches:
refs/heads/master 258b385f4 -> 1269f75ee
[hotfix] [cep] Fix afterMatchStrategy parameter missing issue
This closes #4673
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/1269f75e
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/1269f75e
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/1269f75e
Branch: refs/heads/master
Commit: 1269f75ee86f9c3c5dfcd0c05ee39c974bd73fa5
Parents: 258b385
Author: Yestin <87...@qq.com>
Authored: Thu Sep 14 12:01:53 2017 -0400
Committer: Dawid Wysakowicz <dw...@apache.org>
Committed: Sun Sep 17 18:36:35 2017 +0200
----------------------------------------------------------------------
.../AbstractKeyedCEPPatternOperator.java | 2 +-
.../java/org/apache/flink/cep/CEPITCase.java | 35 ++++++++++++++++++++
2 files changed, 36 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/1269f75e/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
index 01aad76..2415a2e 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
@@ -329,7 +329,7 @@ public abstract class AbstractKeyedCEPPatternOperator<IN, KEY, OUT, F extends Fu
*/
private void processEvent(NFA<IN> nfa, IN event, long timestamp) {
Tuple2<Collection<Map<String, List<IN>>>, Collection<Tuple2<Map<String, List<IN>>, Long>>> patterns =
- nfa.process(event, timestamp);
+ nfa.process(event, timestamp, afterMatchSkipStrategy);
try {
processMatchedSequences(patterns.f0, timestamp);
http://git-wip-us.apache.org/repos/asf/flink/blob/1269f75e/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/CEPITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/CEPITCase.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/CEPITCase.java
index 78e485e..81b83a3 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/CEPITCase.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/CEPITCase.java
@@ -21,6 +21,7 @@ package org.apache.flink.cep;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.cep.nfa.AfterMatchSkipStrategy;
import org.apache.flink.cep.nfa.NFA;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
@@ -679,4 +680,38 @@ public class CEPITCase extends StreamingMultipleProgramsTestBase {
return Double.compare(o1.getPrice(), o2.getPrice());
}
}
+
+ @Test
+ public void testSimpleAfterMatchSkip() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ DataStream<Tuple2<Integer, String>> input = env.fromElements(
+ new Tuple2<>(1, "a"),
+ new Tuple2<>(2, "a"),
+ new Tuple2<>(3, "a"),
+ new Tuple2<>(4, "a"));
+
+ Pattern<Tuple2<Integer, String>, ?> pattern =
+ Pattern.<Tuple2<Integer, String>>begin("start", AfterMatchSkipStrategy.skipPastLastEvent())
+ .where(new SimpleCondition<Tuple2<Integer, String>>() {
+ @Override
+ public boolean filter(Tuple2<Integer, String> rec) throws Exception {
+ return rec.f1.equals("a");
+ }
+ }).times(2);
+
+ PatternStream<Tuple2<Integer, String>> pStream = CEP.pattern(input, pattern);
+
+ DataStream<Tuple2<Integer, String>> result = pStream.select(new PatternSelectFunction<Tuple2<Integer, String>, Tuple2<Integer, String>>() {
+ @Override
+ public Tuple2<Integer, String> select(Map<String, List<Tuple2<Integer, String>>> pattern) throws Exception {
+ return pattern.get("start").get(0);
+ }
+ });
+
+ result.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
+
+ expected = "(1,a)\n(3,a)";
+
+ env.execute();
+ }
}