You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by di...@apache.org on 2023/02/17 08:13:22 UTC
[flink] branch release-1.17 updated: [FLINK-31017][FLINK-31042][cep] Fix no after-match processing when pattern timeout
This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.17 by this push:
new 257be14bb47 [FLINK-31017][FLINK-31042][cep] Fix no after-match processing when pattern timeout
257be14bb47 is described below
commit 257be14bb47f7481927ea17bc82b4b6ff63e8b30
Author: Juntao Hu <ma...@gmail.com>
AuthorDate: Mon Feb 13 21:29:25 2023 +0800
[FLINK-31017][FLINK-31042][cep] Fix no after-match processing when pattern timeout
This closes #21920.
---
.../main/java/org/apache/flink/cep/nfa/NFA.java | 115 +++++++++++----------
.../cep/nfa/aftermatch/AfterMatchSkipStrategy.java | 3 +
.../org/apache/flink/cep/operator/CepOperator.java | 6 +-
.../test/java/org/apache/flink/cep/CEPITCase.java | 71 +++++++++++++
.../java/org/apache/flink/cep/nfa/NFAITCase.java | 36 +++++--
.../flink/cep/nfa/NFAStatusChangeITCase.java | 8 +-
.../org/apache/flink/cep/nfa/NotPatternITCase.java | 41 ++++++++
.../org/apache/flink/cep/utils/NFATestHarness.java | 7 +-
8 files changed, 214 insertions(+), 73 deletions(-)
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java
index 537a28eac04..7b62ed5f458 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java
@@ -270,13 +270,16 @@ public class NFA<T> {
advanceTime(
final SharedBufferAccessor<T> sharedBufferAccessor,
final NFAState nfaState,
- final long timestamp)
+ final long timestamp,
+ final AfterMatchSkipStrategy afterMatchSkipStrategy)
throws Exception {
- final Collection<Map<String, List<T>>> pendingMatches = new ArrayList<>();
+ final List<Map<String, List<T>>> result = new ArrayList<>();
final Collection<Tuple2<Map<String, List<T>>, Long>> timeoutResult = new ArrayList<>();
final PriorityQueue<ComputationState> newPartialMatches =
new PriorityQueue<>(NFAState.COMPUTATION_STATE_COMPARATOR);
+ final PriorityQueue<ComputationState> potentialMatches =
+ new PriorityQueue<>(NFAState.COMPUTATION_STATE_COMPARATOR);
for (ComputationState computationState : nfaState.getPartialMatches()) {
String currentStateName = computationState.getCurrentStateName();
@@ -294,13 +297,16 @@ public class NFA<T> {
computationState.getStartTimestamp(),
windowTime);
if (isTimeoutForPreviousEvent || isTimeoutForFirstEvent) {
+ nfaState.setStateChanged();
+
if (getState(computationState).isPending()) {
- // extract the Pending State
- Map<String, List<T>> pendingPattern =
- sharedBufferAccessor.materializeMatch(
- extractCurrentMatches(sharedBufferAccessor, computationState));
- pendingMatches.add(pendingPattern);
- } else if (handleTimeout) {
+ // save pending states for after-match pruning, where those states will be
+ // released
+ potentialMatches.add(computationState);
+ continue;
+ }
+
+ if (handleTimeout) {
// extract the timed out event pattern
Map<String, List<T>> timedOutPattern =
sharedBufferAccessor.materializeMatch(
@@ -315,20 +321,29 @@ public class NFA<T> {
: computationState.getStartTimestamp() + windowTime));
}
+ // release timeout states
sharedBufferAccessor.releaseNode(
computationState.getPreviousBufferEntry(), computationState.getVersion());
-
- nfaState.setStateChanged();
} else {
newPartialMatches.add(computationState);
}
}
+ // If a timeout partial match "frees" some completed matches
+ // Or if completed not-followed-by matches need pruning
+ processMatchesAccordingToSkipStrategy(
+ sharedBufferAccessor,
+ nfaState,
+ afterMatchSkipStrategy,
+ potentialMatches,
+ newPartialMatches,
+ result);
+
nfaState.setNewPartialMatches(newPartialMatches);
sharedBufferAccessor.advanceTime(timestamp);
- return Tuple2.of(pendingMatches, timeoutResult);
+ return Tuple2.of(result, timeoutResult);
}
private boolean isStateTimedOut(
@@ -349,7 +364,7 @@ public class NFA<T> {
final PriorityQueue<ComputationState> newPartialMatches =
new PriorityQueue<>(NFAState.COMPUTATION_STATE_COMPARATOR);
- final PriorityQueue<ComputationState> potentialMatches =
+ PriorityQueue<ComputationState> potentialMatches =
new PriorityQueue<>(NFAState.COMPUTATION_STATE_COMPARATOR);
// iterate over all current computations
@@ -406,28 +421,13 @@ public class NFA<T> {
}
List<Map<String, List<T>>> result = new ArrayList<>();
- if (afterMatchSkipStrategy.isSkipStrategy()) {
- processMatchesAccordingToSkipStrategy(
- sharedBufferAccessor,
- nfaState,
- afterMatchSkipStrategy,
- potentialMatches,
- newPartialMatches,
- result);
- } else {
- for (ComputationState match : potentialMatches) {
- Map<String, List<T>> materializedMatch =
- sharedBufferAccessor.materializeMatch(
- sharedBufferAccessor
- .extractPatterns(
- match.getPreviousBufferEntry(), match.getVersion())
- .get(0));
-
- result.add(materializedMatch);
- sharedBufferAccessor.releaseNode(
- match.getPreviousBufferEntry(), match.getVersion());
- }
- }
+ processMatchesAccordingToSkipStrategy(
+ sharedBufferAccessor,
+ nfaState,
+ afterMatchSkipStrategy,
+ potentialMatches,
+ newPartialMatches,
+ result);
nfaState.setNewPartialMatches(newPartialMatches);
@@ -445,35 +445,36 @@ public class NFA<T> {
nfaState.getCompletedMatches().addAll(potentialMatches);
- ComputationState earliestMatch = nfaState.getCompletedMatches().peek();
+ ComputationState earliestMatch;
+ while ((earliestMatch = nfaState.getCompletedMatches().peek()) != null) {
- if (earliestMatch != null) {
-
- ComputationState earliestPartialMatch;
- while (earliestMatch != null
- && ((earliestPartialMatch = partialMatches.peek()) == null
- || isEarlier(earliestMatch, earliestPartialMatch))) {
-
- nfaState.setStateChanged();
- nfaState.getCompletedMatches().poll();
- List<Map<String, List<EventId>>> matchedResult =
- sharedBufferAccessor.extractPatterns(
- earliestMatch.getPreviousBufferEntry(), earliestMatch.getVersion());
+ // Care for ordering when it's not NO_SKIP
+ if (afterMatchSkipStrategy.isSkipStrategy()) {
+ ComputationState earliestPartialMatch = partialMatches.peek();
+ if (earliestPartialMatch != null
+ && !isEarlier(earliestMatch, earliestPartialMatch)) {
+ break;
+ }
+ }
- afterMatchSkipStrategy.prune(partialMatches, matchedResult, sharedBufferAccessor);
+ nfaState.setStateChanged();
+ nfaState.getCompletedMatches().poll();
+ List<Map<String, List<EventId>>> matchedResult =
+ sharedBufferAccessor.extractPatterns(
+ earliestMatch.getPreviousBufferEntry(), earliestMatch.getVersion());
- afterMatchSkipStrategy.prune(
- nfaState.getCompletedMatches(), matchedResult, sharedBufferAccessor);
+ afterMatchSkipStrategy.prune(partialMatches, matchedResult, sharedBufferAccessor);
- result.add(sharedBufferAccessor.materializeMatch(matchedResult.get(0)));
- sharedBufferAccessor.releaseNode(
- earliestMatch.getPreviousBufferEntry(), earliestMatch.getVersion());
- earliestMatch = nfaState.getCompletedMatches().peek();
- }
+ afterMatchSkipStrategy.prune(
+ nfaState.getCompletedMatches(), matchedResult, sharedBufferAccessor);
- nfaState.getPartialMatches()
- .removeIf(pm -> pm.getStartEventID() != null && !partialMatches.contains(pm));
+ result.add(sharedBufferAccessor.materializeMatch(matchedResult.get(0)));
+ sharedBufferAccessor.releaseNode(
+ earliestMatch.getPreviousBufferEntry(), earliestMatch.getVersion());
}
+
+ nfaState.getPartialMatches()
+ .removeIf(pm -> pm.getStartEventID() != null && !partialMatches.contains(pm));
}
private boolean isEarlier(
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 346a8f9c49e..a3491a5f804 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
@@ -103,6 +103,9 @@ public abstract class AfterMatchSkipStrategy implements Serializable {
Collection<Map<String, List<EventId>>> matchedResult,
SharedBufferAccessor<?> sharedBufferAccessor)
throws Exception {
+ if (!isSkipStrategy()) {
+ return;
+ }
EventId pruningId = getPruningId(matchedResult);
if (pruningId != null) {
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepOperator.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepOperator.java
index b7ea2373a59..202402e8ee3 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepOperator.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepOperator.java
@@ -426,7 +426,11 @@ public class CepOperator<IN, KEY, OUT>
Collection<Map<String, List<IN>>>,
Collection<Tuple2<Map<String, List<IN>>, Long>>>
pendingMatchesAndTimeout =
- nfa.advanceTime(sharedBufferAccessor, nfaState, timestamp);
+ nfa.advanceTime(
+ sharedBufferAccessor,
+ nfaState,
+ timestamp,
+ afterMatchSkipStrategy);
Collection<Map<String, List<IN>>> pendingMatches = pendingMatchesAndTimeout.f0;
Collection<Tuple2<Map<String, List<IN>>, Long>> timedOut = pendingMatchesAndTimeout.f1;
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 aeb40f7101e..da96b9416a9 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
@@ -18,6 +18,9 @@
package org.apache.flink.cep;
+import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
+import org.apache.flink.api.common.eventtime.TimestampAssignerSupplier;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
@@ -29,6 +32,7 @@ import org.apache.flink.cep.nfa.NFA;
import org.apache.flink.cep.nfa.aftermatch.AfterMatchSkipStrategy;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.WithinType;
+import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.cep.pattern.conditions.RichIterativeCondition;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
@@ -41,6 +45,7 @@ import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.test.util.AbstractTestBase;
import org.apache.flink.types.Either;
+import org.apache.flink.util.CloseableIterator;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
@@ -1028,4 +1033,70 @@ public class CEPITCase extends AbstractTestBase {
env.execute();
}
+
+ @Test
+ public void testPartialMatchTimeoutOutputCompletedMatch() throws Exception {
+ StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment(envConfiguration);
+
+ // (Event, timestamp)
+ DataStream<Event> input =
+ env.fromElements(
+ Tuple2.of(new Event(1, "start", 1.0), 0L),
+ Tuple2.of(new Event(2, "start", 2.0), 1L),
+ Tuple2.of(new Event(3, "start", 3.0), 2L),
+ Tuple2.of(new Event(4, "start", 4.0), 3L),
+ Tuple2.of(new Event(5, "end", 5.0), 4L))
+ .assignTimestampsAndWatermarks(
+ WatermarkStrategy.<Tuple2<Event, Long>>forBoundedOutOfOrderness(
+ Duration.ofMillis(5))
+ .withTimestampAssigner(
+ TimestampAssignerSupplier.of(
+ (SerializableTimestampAssigner<
+ Tuple2<Event, Long>>)
+ (element, recordTimestamp) ->
+ element.f1)))
+ .map((MapFunction<Tuple2<Event, Long>, Event>) value -> value.f0);
+
+ Pattern<Event, ?> pattern =
+ Pattern.<Event>begin("start", AfterMatchSkipStrategy.skipPastLastEvent())
+ .where(SimpleCondition.of(value -> value.getName().equals("start")))
+ .oneOrMore()
+ .consecutive()
+ .greedy()
+ .followedBy("middle")
+ .where(
+ new IterativeCondition<Event>() {
+ @Override
+ public boolean filter(Event value, Context<Event> ctx)
+ throws Exception {
+ int count = 0;
+ for (Event ignored : ctx.getEventsForPattern("start")) {
+ count++;
+ }
+ if (count > 2) {
+ return value.getName().equals("middle");
+ } else {
+ return value.getName().equals("end");
+ }
+ }
+ })
+ .within(Time.milliseconds(100L));
+
+ DataStream<String> result =
+ CEP.pattern(input, pattern)
+ .select(
+ (PatternSelectFunction<Event, String>)
+ pattern1 ->
+ pattern1.get("start").get(0).getId()
+ + ","
+ + pattern1.get("middle").get(0).getId());
+
+ List<String> resultList = new ArrayList<>();
+ try (CloseableIterator<String> iterator = result.executeAndCollect()) {
+ iterator.forEachRemaining(resultList::add);
+ }
+ resultList.sort(String::compareTo);
+ assertEquals(Arrays.asList("3,5"), resultList);
+ }
}
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java
index fd70328c1c3..2047961e140 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java
@@ -402,7 +402,12 @@ public class NFAITCase extends TestLogger {
for (StreamRecord<Event> event : events) {
Collection<Tuple2<Map<String, List<Event>>, Long>> timeoutPatterns =
- nfa.advanceTime(sharedBufferAccessor, nfaState, event.getTimestamp()).f1;
+ nfa.advanceTime(
+ sharedBufferAccessor,
+ nfaState,
+ event.getTimestamp(),
+ AfterMatchSkipStrategy.noSkip())
+ .f1;
Collection<Map<String, List<Event>>> matchedPatterns =
nfa.process(
sharedBufferAccessor,
@@ -467,7 +472,12 @@ public class NFAITCase extends TestLogger {
for (StreamRecord<Event> event : events) {
Collection<Tuple2<Map<String, List<Event>>, Long>> timeoutPatterns =
- nfa.advanceTime(sharedBufferAccessor, nfaState, event.getTimestamp()).f1;
+ nfa.advanceTime(
+ sharedBufferAccessor,
+ nfaState,
+ event.getTimestamp(),
+ AfterMatchSkipStrategy.noSkip())
+ .f1;
Collection<Map<String, List<Event>>> matchedPatterns =
nfa.process(
sharedBufferAccessor,
@@ -530,7 +540,12 @@ public class NFAITCase extends TestLogger {
for (StreamRecord<Event> event : events) {
Collection<Map<String, List<Event>>> pendingMatches =
- nfa.advanceTime(sharedBufferAccessor, nfaState, event.getTimestamp()).f0;
+ nfa.advanceTime(
+ sharedBufferAccessor,
+ nfaState,
+ event.getTimestamp(),
+ AfterMatchSkipStrategy.noSkip())
+ .f0;
resultingPendingMatches.addAll(pendingMatches);
nfa.process(
sharedBufferAccessor,
@@ -2304,7 +2319,7 @@ public class NFAITCase extends TestLogger {
nfaTestHarness.feedRecord(new StreamRecord<>(end1, 6));
// pruning element
- nfa.advanceTime(sharedBufferAccessor, nfaState, 10);
+ nfa.advanceTime(sharedBufferAccessor, nfaState, 10, AfterMatchSkipStrategy.noSkip());
assertEquals(1, nfaState.getPartialMatches().size());
assertEquals("start", nfaState.getPartialMatches().peek().getCurrentStateName());
@@ -2345,7 +2360,7 @@ public class NFAITCase extends TestLogger {
nfaTestHarness.feedRecord(new StreamRecord<>(end1, 6));
// pruning element
- nfa.advanceTime(sharedBufferAccessor, nfaState, 10);
+ nfa.advanceTime(sharedBufferAccessor, nfaState, 10, AfterMatchSkipStrategy.noSkip());
assertEquals(1, nfaState.getPartialMatches().size());
assertEquals("start", nfaState.getPartialMatches().peek().getCurrentStateName());
@@ -2389,7 +2404,7 @@ public class NFAITCase extends TestLogger {
nfaTestHarness.consumeRecord(new StreamRecord<>(end1, 6));
// pruning element
- nfa.advanceTime(sharedBufferAccessor, nfaState, 10);
+ nfa.advanceTime(sharedBufferAccessor, nfaState, 10, AfterMatchSkipStrategy.noSkip());
assertEquals(1, nfaState.getPartialMatches().size());
assertEquals("start", nfaState.getPartialMatches().peek().getCurrentStateName());
@@ -2425,7 +2440,7 @@ public class NFAITCase extends TestLogger {
nfaTestHarness.consumeRecord(new StreamRecord<>(end1, 6));
// pruning element
- nfa.advanceTime(sharedBufferAccessor, nfaState, 10);
+ nfa.advanceTime(sharedBufferAccessor, nfaState, 10, AfterMatchSkipStrategy.noSkip());
assertEquals(1, nfaState.getPartialMatches().size());
assertEquals("start", nfaState.getPartialMatches().peek().getCurrentStateName());
@@ -2461,7 +2476,7 @@ public class NFAITCase extends TestLogger {
nfaTestHarness.consumeRecord(new StreamRecord<>(end1, 6));
// pruning element
- nfa.advanceTime(sharedBufferAccessor, nfaState, 10);
+ nfa.advanceTime(sharedBufferAccessor, nfaState, 10, AfterMatchSkipStrategy.noSkip());
assertEquals(3, nfaState.getPartialMatches().size());
assertEquals(
@@ -2826,7 +2841,8 @@ public class NFAITCase extends TestLogger {
AfterMatchSkipStrategy.noSkip(),
timerService);
Mockito.verify(accessor, Mockito.never()).advanceTime(anyLong());
- nfa.advanceTime(accessor, nfa.createInitialNFAState(), 2);
+ nfa.advanceTime(
+ accessor, nfa.createInitialNFAState(), 2, AfterMatchSkipStrategy.noSkip());
Mockito.verify(accessor, Mockito.times(1)).advanceTime(2);
}
}
@@ -2857,7 +2873,7 @@ public class NFAITCase extends TestLogger {
try (SharedBufferAccessor<Event> accessor = sharedBuffer.getAccessor()) {
nfa.process(accessor, nfaState, a1, 1, AfterMatchSkipStrategy.noSkip(), timerService);
nfa.process(accessor, nfaState, a2, 2, AfterMatchSkipStrategy.noSkip(), timerService);
- nfa.advanceTime(accessor, nfaState, 4);
+ nfa.advanceTime(accessor, nfaState, 4, AfterMatchSkipStrategy.noSkip());
}
assertThat(
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAStatusChangeITCase.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAStatusChangeITCase.java
index 94cbc9001f2..12ecee55548 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAStatusChangeITCase.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAStatusChangeITCase.java
@@ -191,7 +191,7 @@ public class NFAStatusChangeITCase {
// both the queue of ComputationStatus and eventSharedBuffer have not changed
// as the timestamp is within the window
nfaState.resetStateChanged();
- nfa.advanceTime(sharedBufferAccessor, nfaState, 8L);
+ nfa.advanceTime(sharedBufferAccessor, nfaState, 8L, skipStrategy);
assertFalse(
"NFA status should not change as the timestamp is within the window",
nfaState.isStateChanged());
@@ -201,7 +201,7 @@ public class NFAStatusChangeITCase {
// be removed from eventSharedBuffer as the timeout happens
nfaState.resetStateChanged();
Collection<Tuple2<Map<String, List<Event>>, Long>> timeoutResults =
- nfa.advanceTime(sharedBufferAccessor, nfaState, 12L).f1;
+ nfa.advanceTime(sharedBufferAccessor, nfaState, 12L, skipStrategy).f1;
assertTrue(
"NFA status should change as timeout happens",
nfaState.isStateChanged() && !timeoutResults.isEmpty());
@@ -285,7 +285,7 @@ public class NFAStatusChangeITCase {
NFAState nfaState = nfa.createInitialNFAState();
nfaState.resetStateChanged();
- nfa.advanceTime(sharedBufferAccessor, nfaState, 6L);
+ nfa.advanceTime(sharedBufferAccessor, nfaState, 6L, skipStrategy);
nfa.process(
sharedBufferAccessor,
nfaState,
@@ -295,7 +295,7 @@ public class NFAStatusChangeITCase {
timerService);
nfaState.resetStateChanged();
- nfa.advanceTime(sharedBufferAccessor, nfaState, 17L);
+ nfa.advanceTime(sharedBufferAccessor, nfaState, 17L, skipStrategy);
assertTrue(nfaState.isStateChanged());
}
}
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NotPatternITCase.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NotPatternITCase.java
index 75288d7ef8c..dc1dbe3dc90 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NotPatternITCase.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NotPatternITCase.java
@@ -19,9 +19,11 @@
package org.apache.flink.cep.nfa;
import org.apache.flink.cep.Event;
+import org.apache.flink.cep.nfa.aftermatch.AfterMatchSkipStrategy;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.WithinType;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
+import org.apache.flink.cep.utils.NFATestHarness;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.util.TestLogger;
@@ -31,6 +33,7 @@ import org.apache.flink.shaded.guava30.com.google.common.collect.Lists;
import org.junit.Test;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import static org.apache.flink.cep.utils.NFATestUtilities.comparePatterns;
@@ -1070,4 +1073,42 @@ public class NotPatternITCase extends TestLogger {
Lists.newArrayList(a3),
Lists.newArrayList(a3, c3)));
}
+
+ @Test
+ public void testNotFollowedByWithinAtEndAfterMatch() throws Exception {
+ List<StreamRecord<Event>> inputEvents = new ArrayList<>();
+
+ Event a1 = new Event(40, "a", 1.0);
+ Event a2 = new Event(41, "a", 2.0);
+ Event a3 = new Event(42, "a", 3.0);
+ Event c1 = new Event(43, "c", 4.0);
+ Event c2 = new Event(44, "c", 5.0);
+
+ inputEvents.add(new StreamRecord<>(a1, 1));
+ inputEvents.add(new StreamRecord<>(a2, 2));
+ inputEvents.add(new StreamRecord<>(a3, 3));
+ inputEvents.add(new StreamRecord<>(c1, 4));
+ inputEvents.add(new StreamRecord<>(c2, 10));
+
+ Pattern<Event, ?> pattern =
+ Pattern.<Event>begin("a", AfterMatchSkipStrategy.skipPastLastEvent())
+ .where(SimpleCondition.of(value -> value.getName().equals("a")))
+ .oneOrMore()
+ .allowCombinations()
+ .followedBy("c")
+ .where(SimpleCondition.of(value -> value.getName().equals("c")))
+ .notFollowedBy("b")
+ .where(SimpleCondition.of(value -> value.getName().equals("b")))
+ .within(Time.milliseconds(5));
+
+ NFA<Event> nfa = compile(pattern, false);
+
+ NFATestHarness harness =
+ NFATestHarness.forNFA(nfa)
+ .withAfterMatchSkipStrategy(AfterMatchSkipStrategy.skipPastLastEvent())
+ .build();
+ final List<List<Event>> matches = harness.feedRecords(inputEvents);
+
+ comparePatterns(matches, Collections.singletonList(Lists.newArrayList(a1, a2, a3, c1)));
+ }
}
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/utils/NFATestHarness.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/utils/NFATestHarness.java
index bc1a51cc5e2..1343b178237 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/utils/NFATestHarness.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/utils/NFATestHarness.java
@@ -100,7 +100,12 @@ public final class NFATestHarness {
throws Exception {
try (SharedBufferAccessor<Event> sharedBufferAccessor = sharedBuffer.getAccessor()) {
Collection<Map<String, List<Event>>> pendingMatches =
- nfa.advanceTime(sharedBufferAccessor, nfaState, inputEvent.getTimestamp()).f0;
+ nfa.advanceTime(
+ sharedBufferAccessor,
+ nfaState,
+ inputEvent.getTimestamp(),
+ afterMatchSkipStrategy)
+ .f0;
Collection<Map<String, List<Event>>> matchedPatterns =
nfa.process(
sharedBufferAccessor,