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 2018/10/01 07:46:48 UTC
[flink] branch master updated: [FLINK-10417][cep] Added option to
throw exception on pattern variable miss during SKIP_TO_FIRST/LAST
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new c1aca82 [FLINK-10417][cep] Added option to throw exception on pattern variable miss during SKIP_TO_FIRST/LAST
c1aca82 is described below
commit c1aca8247075e948e7f9dc649008e97be523d717
Author: Dawid Wysakowicz <wy...@gmail.com>
AuthorDate: Mon Oct 1 09:46:37 2018 +0200
[FLINK-10417][cep] Added option to throw exception on pattern variable miss during SKIP_TO_FIRST/LAST
---
docs/dev/libs/cep.md | 17 ++++++
.../cep/nfa/aftermatch/AfterMatchSkipStrategy.java | 12 ++--
...rstStrategy.java => SkipToElementStrategy.java} | 45 +++++++++-----
.../cep/nfa/aftermatch/SkipToFirstStrategy.java | 45 +++-----------
.../cep/nfa/aftermatch/SkipToLastStrategy.java | 45 +++-----------
.../apache/flink/cep/nfa/AfterMatchSkipITCase.java | 69 ++++++++++++++++++++++
6 files changed, 137 insertions(+), 96 deletions(-)
diff --git a/docs/dev/libs/cep.md b/docs/dev/libs/cep.md
index ad321bf..d7f915f 100644
--- a/docs/dev/libs/cep.md
+++ b/docs/dev/libs/cep.md
@@ -1385,6 +1385,23 @@ Pattern.begin("patternName", skipStrategy)
</div>
</div>
+{% warn Attention %} For SKIP_TO_FIRST/LAST there are two options how to handle cases when there are no elements mapped to
+the specified variable. By default a NO_SKIP strategy will be used in this case. The other option is to throw exception in such situation.
+One can enable this option by:
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+AfterMatchSkipStrategy.skipToFirst(patternName).throwExceptionOnMiss()
+{% endhighlight %}
+</div>
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+AfterMatchSkipStrategy.skipToFirst(patternName).throwExceptionOnMiss()
+{% endhighlight %}
+</div>
+</div>
+
## Detecting Patterns
After specifying the pattern sequence you are looking for, it is time to apply it to your input stream to detect
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/aftermatch/AfterMatchSkipStrategy.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/aftermatch/AfterMatchSkipStrategy.java
index 8151a12..f4448a3 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/aftermatch/AfterMatchSkipStrategy.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/aftermatch/AfterMatchSkipStrategy.java
@@ -43,8 +43,8 @@ public abstract class AfterMatchSkipStrategy implements Serializable {
* @param patternName the pattern name to skip to
* @return the created AfterMatchSkipStrategy
*/
- public static AfterMatchSkipStrategy skipToFirst(String patternName) {
- return new SkipToFirstStrategy(patternName);
+ public static SkipToFirstStrategy skipToFirst(String patternName) {
+ return new SkipToFirstStrategy(patternName, false);
}
/**
@@ -53,8 +53,8 @@ public abstract class AfterMatchSkipStrategy implements Serializable {
* @param patternName the pattern name to skip to
* @return the created AfterMatchSkipStrategy
*/
- public static AfterMatchSkipStrategy skipToLast(String patternName) {
- return new SkipToLastStrategy(patternName);
+ public static SkipToLastStrategy skipToLast(String patternName) {
+ return new SkipToLastStrategy(patternName, false);
}
/**
@@ -62,7 +62,7 @@ public abstract class AfterMatchSkipStrategy implements Serializable {
*
* @return the created AfterMatchSkipStrategy
*/
- public static AfterMatchSkipStrategy skipPastLastEvent() {
+ public static SkipPastLastStrategy skipPastLastEvent() {
return SkipPastLastStrategy.INSTANCE;
}
@@ -71,7 +71,7 @@ public abstract class AfterMatchSkipStrategy implements Serializable {
*
* @return the created AfterMatchSkipStrategy
*/
- public static AfterMatchSkipStrategy noSkip() {
+ public static NoSkipStrategy noSkip() {
return NoSkipStrategy.INSTANCE;
}
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/aftermatch/SkipToFirstStrategy.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/aftermatch/SkipToElementStrategy.java
similarity index 59%
copy from flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/aftermatch/SkipToFirstStrategy.java
copy to flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/aftermatch/SkipToElementStrategy.java
index 7d7be4a..5554151 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/aftermatch/SkipToFirstStrategy.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/aftermatch/SkipToElementStrategy.java
@@ -19,6 +19,7 @@
package org.apache.flink.cep.nfa.aftermatch;
import org.apache.flink.cep.nfa.sharedbuffer.EventId;
+import org.apache.flink.util.FlinkRuntimeException;
import java.util.Collection;
import java.util.List;
@@ -27,16 +28,14 @@ import java.util.Optional;
import static org.apache.flink.util.Preconditions.checkNotNull;
-/**
- * Discards every partial match that contains event of the match preceding the first of *PatternName*.
- */
-public class SkipToFirstStrategy extends AfterMatchSkipStrategy {
-
+abstract class SkipToElementStrategy extends AfterMatchSkipStrategy {
private static final long serialVersionUID = 7127107527654629026L;
private final String patternName;
+ private final boolean shouldThrowException;
- SkipToFirstStrategy(String patternName) {
+ SkipToElementStrategy(String patternName, boolean shouldThrowException) {
this.patternName = checkNotNull(patternName);
+ this.shouldThrowException = shouldThrowException;
}
@Override
@@ -51,15 +50,22 @@ public class SkipToFirstStrategy extends AfterMatchSkipStrategy {
@Override
protected EventId getPruningId(Collection<Map<String, List<EventId>>> match) {
- EventId pruniningId = null;
+ EventId pruningId = null;
for (Map<String, List<EventId>> resultMap : match) {
List<EventId> pruningPattern = resultMap.get(patternName);
- if (pruningPattern != null && !pruningPattern.isEmpty()) {
- pruniningId = max(pruniningId, pruningPattern.get(0));
+ if (pruningPattern == null || pruningPattern.isEmpty()) {
+ if (shouldThrowException) {
+ throw new FlinkRuntimeException(String.format(
+ "Could not skip to %s. No such element in the found match %s",
+ patternName,
+ resultMap));
+ }
+ } else {
+ pruningId = max(pruningId, pruningPattern.get(getIndex(pruningPattern.size())));
}
}
- return pruniningId;
+ return pruningId;
}
@Override
@@ -67,10 +73,17 @@ public class SkipToFirstStrategy extends AfterMatchSkipStrategy {
return Optional.of(patternName);
}
- @Override
- public String toString() {
- return "SkipToFirstStrategy{" +
- "patternName='" + patternName + '\'' +
- '}';
- }
+ /**
+ * Tells which element from the list of events mapped to *PatternName* to use.
+ *
+ * @param size number of elements mapped to the *PatternName*
+ * @return index of event mapped to *PatternName* to use for pruning
+ */
+ abstract int getIndex(int size);
+
+ /**
+ * Enables throwing exception if no events mapped to the *PatternName*. If not enabled and no events were mapped,
+ * {@link NoSkipStrategy} will be used
+ */
+ public abstract SkipToElementStrategy throwExceptionOnMiss();
}
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/aftermatch/SkipToFirstStrategy.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/aftermatch/SkipToFirstStrategy.java
index 7d7be4a..e8befeb 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/aftermatch/SkipToFirstStrategy.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/aftermatch/SkipToFirstStrategy.java
@@ -18,59 +18,30 @@
package org.apache.flink.cep.nfa.aftermatch;
-import org.apache.flink.cep.nfa.sharedbuffer.EventId;
-
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
/**
* Discards every partial match that contains event of the match preceding the first of *PatternName*.
*/
-public class SkipToFirstStrategy extends AfterMatchSkipStrategy {
-
+public final class SkipToFirstStrategy extends SkipToElementStrategy {
private static final long serialVersionUID = 7127107527654629026L;
- private final String patternName;
- SkipToFirstStrategy(String patternName) {
- this.patternName = checkNotNull(patternName);
+ SkipToFirstStrategy(String patternName, boolean shouldThrowException) {
+ super(patternName, shouldThrowException);
}
@Override
- public boolean isSkipStrategy() {
- return true;
- }
-
- @Override
- protected boolean shouldPrune(EventId startEventID, EventId pruningId) {
- return startEventID != null && startEventID.compareTo(pruningId) < 0;
- }
-
- @Override
- protected EventId getPruningId(Collection<Map<String, List<EventId>>> match) {
- EventId pruniningId = null;
- for (Map<String, List<EventId>> resultMap : match) {
- List<EventId> pruningPattern = resultMap.get(patternName);
- if (pruningPattern != null && !pruningPattern.isEmpty()) {
- pruniningId = max(pruniningId, pruningPattern.get(0));
- }
- }
-
- return pruniningId;
+ public SkipToElementStrategy throwExceptionOnMiss() {
+ return new SkipToFirstStrategy(getPatternName().get(), true);
}
@Override
- public Optional<String> getPatternName() {
- return Optional.of(patternName);
+ int getIndex(int size) {
+ return 0;
}
@Override
public String toString() {
return "SkipToFirstStrategy{" +
- "patternName='" + patternName + '\'' +
+ "patternName='" + getPatternName().get() + '\'' +
'}';
}
}
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/aftermatch/SkipToLastStrategy.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/aftermatch/SkipToLastStrategy.java
index 0f6c3ed..3c33bd1 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/aftermatch/SkipToLastStrategy.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/aftermatch/SkipToLastStrategy.java
@@ -18,59 +18,30 @@
package org.apache.flink.cep.nfa.aftermatch;
-import org.apache.flink.cep.nfa.sharedbuffer.EventId;
-
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
/**
* Discards every partial match that contains event of the match preceding the last of *PatternName*.
*/
-public class SkipToLastStrategy extends AfterMatchSkipStrategy {
+public final class SkipToLastStrategy extends SkipToElementStrategy {
private static final long serialVersionUID = 7585116990619594531L;
- private final String patternName;
- SkipToLastStrategy(String patternName) {
- this.patternName = checkNotNull(patternName);
- }
-
- @Override
- public boolean isSkipStrategy() {
- return true;
+ SkipToLastStrategy(String patternName, boolean shouldThrowException) {
+ super(patternName, shouldThrowException);
}
@Override
- protected boolean shouldPrune(EventId startEventID, EventId pruningId) {
- return startEventID != null && startEventID.compareTo(pruningId) < 0;
- }
-
- @Override
- protected EventId getPruningId(Collection<Map<String, List<EventId>>> match) {
- EventId pruningId = null;
- for (Map<String, List<EventId>> resultMap : match) {
- List<EventId> pruningPattern = resultMap.get(patternName);
-
- if (pruningPattern != null && !pruningPattern.isEmpty()) {
- pruningId = max(pruningId, pruningPattern.get(pruningPattern.size() - 1));
- }
- }
-
- return pruningId;
+ public SkipToElementStrategy throwExceptionOnMiss() {
+ return new SkipToLastStrategy(getPatternName().get(), true);
}
@Override
- public Optional<String> getPatternName() {
- return Optional.of(patternName);
+ int getIndex(int size) {
+ return size - 1;
}
@Override
public String toString() {
return "SkipToLastStrategy{" +
- "patternName='" + patternName + '\'' +
+ "patternName='" + getPatternName().get() + '\'' +
'}';
}
}
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/AfterMatchSkipITCase.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/AfterMatchSkipITCase.java
index 4462d10..6681090 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/AfterMatchSkipITCase.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/AfterMatchSkipITCase.java
@@ -24,6 +24,7 @@ import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.TestLogger;
import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
@@ -432,6 +433,74 @@ public class AfterMatchSkipITCase extends TestLogger{
));
}
+ @Test(expected = FlinkRuntimeException.class)
+ public void testSkipToFirstNonExistentPosition() throws Exception {
+ MissedSkipTo.compute(AfterMatchSkipStrategy.skipToFirst("b").throwExceptionOnMiss());
+
+ //exception should be thrown
+ }
+
+ @Test
+ public void testSkipToFirstNonExistentPositionWithoutException() throws Exception {
+ List<List<Event>> resultingPatterns = MissedSkipTo.compute(AfterMatchSkipStrategy.skipToFirst("b"));
+
+ compareMaps(resultingPatterns, Collections.singletonList(
+ Lists.newArrayList(MissedSkipTo.a, MissedSkipTo.c)
+ ));
+ }
+
+ @Test(expected = FlinkRuntimeException.class)
+ public void testSkipToLastNonExistentPosition() throws Exception {
+ MissedSkipTo.compute(AfterMatchSkipStrategy.skipToLast("b").throwExceptionOnMiss());
+
+ //exception should be thrown
+ }
+
+ @Test
+ public void testSkipToLastNonExistentPositionWithoutException() throws Exception {
+ List<List<Event>> resultingPatterns = MissedSkipTo.compute(AfterMatchSkipStrategy.skipToFirst("b"));
+
+ compareMaps(resultingPatterns, Collections.singletonList(
+ Lists.newArrayList(MissedSkipTo.a, MissedSkipTo.c)
+ ));
+ }
+
+ static class MissedSkipTo {
+ static Event a = new Event(1, "a", 0.0);
+ static Event c = new Event(4, "c", 0.0);
+
+ static List<List<Event>> compute(AfterMatchSkipStrategy skipStrategy) throws Exception {
+ List<StreamRecord<Event>> streamEvents = new ArrayList<>();
+
+ streamEvents.add(new StreamRecord<>(a));
+ streamEvents.add(new StreamRecord<>(c));
+
+ Pattern<Event, ?> pattern = Pattern.<Event>begin("a").where(
+ new SimpleCondition<Event>() {
+
+ @Override
+ public boolean filter(Event value) throws Exception {
+ return value.getName().contains("a");
+ }
+ }
+ ).next("b").where(new SimpleCondition<Event>() {
+ @Override
+ public boolean filter(Event value) throws Exception {
+ return value.getName().contains("b");
+ }
+ }).oneOrMore().optional().consecutive()
+ .next("c").where(new SimpleCondition<Event>() {
+ @Override
+ public boolean filter(Event value) throws Exception {
+ return value.getName().contains("c");
+ }
+ });
+ NFA<Event> nfa = compile(pattern, false);
+
+ return feedNFA(streamEvents, nfa, skipStrategy);
+ }
+ }
+
@Test
public void testSkipToLastWithOneOrMore() throws Exception {
List<StreamRecord<Event>> streamEvents = new ArrayList<>();