flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ches...@apache.org
Subject [2/3] flink git commit: [FLINK-6137] Activate strict checkstyle for flink-cep
Date Sun, 28 May 2017 08:51:58 GMT
http://git-wip-us.apache.org/repos/asf/flink/blob/c9e574bf/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java
----------------------------------------------------------------------
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 d00bbb7..92b49d3 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
@@ -18,34 +18,37 @@
 
 package org.apache.flink.cep.nfa;
 
-import com.google.common.collect.Lists;
-import com.google.common.primitives.Doubles;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.cep.Event;
 import org.apache.flink.cep.SubEvent;
 import org.apache.flink.cep.nfa.compiler.NFACompiler;
 import org.apache.flink.cep.pattern.Pattern;
 import org.apache.flink.cep.pattern.Quantifier;
-import org.apache.flink.cep.pattern.conditions.IterativeCondition;
 import org.apache.flink.cep.pattern.conditions.SimpleCondition;
 import org.apache.flink.streaming.api.windowing.time.Time;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 import org.apache.flink.util.TestLogger;
-import org.junit.Assert;
+
+import com.google.common.collect.Lists;
 import org.junit.Test;
 
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.Comparator;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
+import static org.apache.flink.cep.nfa.NFATestUtilities.compareMaps;
+import static org.apache.flink.cep.nfa.NFATestUtilities.feedNFA;
 import static org.junit.Assert.assertEquals;
 
+/**
+ * General tests for {@link NFA} features. See also {@link IterativeConditionsITCase}, {@link NotPatternITCase},
+ * {@link SameElementITCase} for more specific tests.
+ */
 @SuppressWarnings("unchecked")
 public class NFAITCase extends TestLogger {
 
@@ -234,7 +237,7 @@ public class NFAITCase extends TestLogger {
 
 	/**
 	 * Tests that the NFA successfully filters out expired elements with respect to the window
-	 * length
+	 * length.
 	 */
 	@Test
 	public void testSimplePatternWithTimeWindowNFA() {
@@ -251,7 +254,6 @@ public class NFAITCase extends TestLogger {
 		events.add(new StreamRecord<>(endEvent = new Event(5, "end", 1.0), 11));
 		events.add(new StreamRecord<>(new Event(6, "end", 1.0), 13));
 
-
 		Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
 			private static final long serialVersionUID = 7907391379273505897L;
 
@@ -373,7 +375,7 @@ public class NFAITCase extends TestLogger {
 		SubEvent middleEvent3 = new SubEvent(43, "foo3", 1.0, 10.0);
 		SubEvent nextOne1 = new SubEvent(44, "next-one", 1.0, 2.0);
 		SubEvent nextOne2 = new SubEvent(45, "next-one", 1.0, 2.0);
-		Event endEvent=  new Event(46, "end", 1.0);
+		Event endEvent = new Event(46, "end", 1.0);
 
 		inputEvents.add(new StreamRecord<>(startEvent, 1));
 		inputEvents.add(new StreamRecord<Event>(middleEvent1, 3));
@@ -1418,7 +1420,7 @@ public class NFAITCase extends TestLogger {
 		final List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
 
 		compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
-				Lists.newArrayList(startEvent,  end1),
+				Lists.newArrayList(startEvent, end1),
 				Lists.newArrayList(end1)
 		));
 	}
@@ -1458,9 +1460,9 @@ public class NFAITCase extends TestLogger {
 		final List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
 
 		compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
-				Lists.newArrayList(startEvent,  middleEvent1, middleEvent2, middleEvent3),
-				Lists.newArrayList(startEvent,  middleEvent1, middleEvent2),
-				Lists.newArrayList(startEvent,  middleEvent1),
+				Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3),
+				Lists.newArrayList(startEvent, middleEvent1, middleEvent2),
+				Lists.newArrayList(startEvent, middleEvent1),
 				Lists.newArrayList(startEvent)
 		));
 	}
@@ -1499,10 +1501,10 @@ public class NFAITCase extends TestLogger {
 		final List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
 
 		compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
-			Lists.newArrayList(middleEvent1,  middleEvent2, middleEvent3),
-			Lists.newArrayList(middleEvent1,  middleEvent2),
+			Lists.newArrayList(middleEvent1, middleEvent2, middleEvent3),
+			Lists.newArrayList(middleEvent1, middleEvent2),
 			Lists.newArrayList(middleEvent1),
-			Lists.newArrayList(middleEvent2,  middleEvent3),
+			Lists.newArrayList(middleEvent2, middleEvent3),
 			Lists.newArrayList(middleEvent2),
 			Lists.newArrayList(middleEvent3)
 		));
@@ -1539,7 +1541,7 @@ public class NFAITCase extends TestLogger {
 		final List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
 
 		compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
-				Lists.newArrayList(startEvent,  middleEvent1),
+				Lists.newArrayList(startEvent, middleEvent1),
 				Lists.newArrayList(startEvent)
 		));
 	}
@@ -1579,9 +1581,9 @@ public class NFAITCase extends TestLogger {
 		final List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
 
 		compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
-			Lists.newArrayList(startEvent,  middleEvent1, middleEvent2, middleEvent3),
-			Lists.newArrayList(startEvent,  middleEvent1, middleEvent2),
-			Lists.newArrayList(startEvent,  middleEvent1)
+			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3),
+			Lists.newArrayList(startEvent, middleEvent1, middleEvent2),
+			Lists.newArrayList(startEvent, middleEvent1)
 		));
 	}
 
@@ -1905,12 +1907,12 @@ public class NFAITCase extends TestLogger {
 	///////////////////////////////         Consecutive           ////////////////////////////////////////
 
 	private static class ConsecutiveData {
-		static final Event startEvent = new Event(40, "c", 1.0);
-		static final Event middleEvent1 = new Event(41, "a", 2.0);
-		static final Event middleEvent2 = new Event(42, "a", 3.0);
-		static final Event middleEvent3 = new Event(43, "a", 4.0);
-		static final Event middleEvent4 = new Event(43, "a", 5.0);
-		static final Event end = new Event(44, "b", 5.0);
+		private static final Event startEvent = new Event(40, "c", 1.0);
+		private static final Event middleEvent1 = new Event(41, "a", 2.0);
+		private static final Event middleEvent2 = new Event(42, "a", 3.0);
+		private static final Event middleEvent3 = new Event(43, "a", 4.0);
+		private static final Event middleEvent4 = new Event(43, "a", 5.0);
+		private static final Event end = new Event(44, "b", 5.0);
 
 		private ConsecutiveData() {
 		}
@@ -2374,7 +2376,6 @@ public class NFAITCase extends TestLogger {
 		assertEquals(true, nfa.isEmpty());
 	}
 
-
 	@Test
 	public void testZeroOrMoreClearingBuffer() {
 		Event startEvent = new Event(40, "c", 1.0);
@@ -2418,385 +2419,6 @@ public class NFAITCase extends TestLogger {
 		assertEquals(true, nfa.isEmpty());
 	}
 
-
-	//////////////////////			Iterative BooleanConditions			/////////////////////////
-
-	private final Event startEvent1 = new Event(40, "start", 1.0);
-	private final Event startEvent2 = new Event(40, "start", 2.0);
-	private final Event startEvent3 = new Event(40, "start", 3.0);
-	private final Event startEvent4 = new Event(40, "start", 4.0);
-	private final SubEvent middleEvent1 = new SubEvent(41, "foo1", 1.0, 10);
-	private final SubEvent middleEvent2 = new SubEvent(42, "foo2", 2.0, 10);
-	private final SubEvent middleEvent3 = new SubEvent(43, "foo3", 3.0, 10);
-	private final SubEvent middleEvent4 = new SubEvent(43, "foo4", 1.0, 10);
-	private final Event nextOne = new Event(44, "next-one", 1.0);
-	private final Event endEvent = new Event(46, "end", 1.0);
-
-	@Test
-	public void testIterativeWithBranchingPatternEager() {
-		List<List<Event>> actual = testIterativeWithBranchingPattern(true);
-
-		compareMaps(actual,
-				Lists.<List<Event>>newArrayList(
-						Lists.newArrayList(startEvent1, endEvent, middleEvent1, middleEvent2, middleEvent4),
-						Lists.newArrayList(startEvent1, endEvent, middleEvent2, middleEvent1),
-						Lists.newArrayList(startEvent1, endEvent, middleEvent1),
-						Lists.newArrayList(startEvent2, endEvent, middleEvent3, middleEvent4),
-						Lists.newArrayList(startEvent2, endEvent, middleEvent3)
-				)
-		);
-	}
-
-	@Test
-	public void testIterativeWithBranchingPatternCombinations() {
-		List<List<Event>> actual = testIterativeWithBranchingPattern(false);
-
-		compareMaps(actual,
-				Lists.<List<Event>>newArrayList(
-						Lists.newArrayList(startEvent1, endEvent, middleEvent1, middleEvent2, middleEvent4),
-						Lists.newArrayList(startEvent1, endEvent, middleEvent2, middleEvent1),
-						Lists.newArrayList(startEvent1, endEvent, middleEvent3, middleEvent1),
-						Lists.newArrayList(startEvent2, endEvent, middleEvent3, middleEvent4),
-						Lists.newArrayList(startEvent1, endEvent, middleEvent4, middleEvent1),
-						Lists.newArrayList(startEvent1, endEvent, middleEvent1),
-						Lists.newArrayList(startEvent2, endEvent, middleEvent3)
-				)
-		);
-	}
-
-	private List<List<Event>> testIterativeWithBranchingPattern(boolean eager) {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		inputEvents.add(new StreamRecord<>(startEvent1, 1));
-		inputEvents.add(new StreamRecord<Event>(middleEvent1, 2));
-		inputEvents.add(new StreamRecord<Event>(middleEvent2, 3));
-		inputEvents.add(new StreamRecord<>(startEvent2, 4));
-		inputEvents.add(new StreamRecord<Event>(middleEvent3, 5));
-		inputEvents.add(new StreamRecord<Event>(middleEvent4, 5));
-		inputEvents.add(new StreamRecord<>(nextOne, 6));
-		inputEvents.add(new StreamRecord<>(endEvent, 8));
-
-		Pattern<Event, ?> pattern = eager
-				? Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
-					private static final long serialVersionUID = 5726188262756267490L;
-
-					@Override
-					public boolean filter(Event value) throws Exception {
-						return value.getName().equals("start");
-					}
-				})
-				.followedBy("middle").subtype(SubEvent.class).where(new MySubeventIterCondition()).oneOrMore()
-				.followedBy("end").where(new SimpleCondition<Event>() {
-					private static final long serialVersionUID = 7056763917392056548L;
-
-					@Override
-					public boolean filter(Event value) throws Exception {
-						return value.getName().equals("end");
-					}
-				})
-				: Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
-					private static final long serialVersionUID = 5726188262756267490L;
-
-					@Override
-					public boolean filter(Event value) throws Exception {
-						return value.getName().equals("start");
-					}
-				})
-				.followedBy("middle").subtype(SubEvent.class).where(new MySubeventIterCondition()).oneOrMore().allowCombinations()
-				.followedBy("end").where(new SimpleCondition<Event>() {
-					private static final long serialVersionUID = 7056763917392056548L;
-
-					@Override
-					public boolean filter(Event value) throws Exception {
-						return value.getName().equals("end");
-					}
-				});
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		return feedNFA(inputEvents, nfa);
-	}
-
-	private static class MySubeventIterCondition extends IterativeCondition<SubEvent> {
-
-		private static final long serialVersionUID = 6215754202506583964L;
-
-		@Override
-		public boolean filter (SubEvent value, Context < SubEvent > ctx) throws Exception {
-			if (!value.getName().startsWith("foo")) {
-				return false;
-			}
-
-			double sum = 0.0;
-			for (Event event : ctx.getEventsForPattern("middle")) {
-				sum += event.getPrice();
-			}
-			sum += value.getPrice();
-			return Double.compare(sum, 5.0) < 0;
-		}
-	}
-
-	@Test
-	public void testIterativeWithLoopingStartingEager() {
-		List<List<Event>> actual = testIterativeWithLoopingStarting(true);
-
-		compareMaps(actual,
-				Lists.<List<Event>>newArrayList(
-						Lists.newArrayList(startEvent1, startEvent2, endEvent),
-						Lists.newArrayList(startEvent1, endEvent),
-						Lists.newArrayList(startEvent2, endEvent),
-						Lists.newArrayList(startEvent3, endEvent),
-						Lists.newArrayList(endEvent)
-				)
-		);
-	}
-
-	@Test
-	public void testIterativeWithLoopingStartingCombination() {
-		List<List<Event>> actual = testIterativeWithLoopingStarting(false);
-
-		compareMaps(actual,
-				Lists.<List<Event>>newArrayList(
-						Lists.newArrayList(startEvent1, startEvent2, endEvent),
-						Lists.newArrayList(startEvent1, startEvent3, endEvent),
-						Lists.newArrayList(startEvent1, endEvent),
-						Lists.newArrayList(startEvent2, endEvent),
-						Lists.newArrayList(startEvent3, endEvent),
-						Lists.newArrayList(endEvent)
-				)
-		);
-	}
-
-	private List<List<Event>> testIterativeWithLoopingStarting(boolean eager) {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		inputEvents.add(new StreamRecord<>(startEvent1, 1L));
-		inputEvents.add(new StreamRecord<>(startEvent2, 2L));
-		inputEvents.add(new StreamRecord<>(startEvent3, 3L));
-		inputEvents.add(new StreamRecord<>(endEvent, 4L));
-
-		// for now, a pattern inherits its continuity property from the followedBy() or next(), and the default
-		// behavior (which is the one applied in the case that the pattern graph starts with such a pattern)
-		// of a looping pattern is with relaxed continuity (as in followedBy).
-
-		Pattern<Event, ?> pattern = eager
-				? Pattern.<Event>begin("start").where(new MyEventIterCondition()).oneOrMore().optional()
-					.followedBy("end").where(new SimpleCondition<Event>() {
-						private static final long serialVersionUID = 7056763917392056548L;
-
-						@Override
-						public boolean filter(Event value) throws Exception {
-							return value.getName().equals("end");
-						}
-					})
-				: Pattern.<Event>begin("start").where(new MyEventIterCondition()).oneOrMore().allowCombinations().optional()
-					.followedBy("end").where(new SimpleCondition<Event>() {
-						private static final long serialVersionUID = 7056763917392056548L;
-
-						@Override
-						public boolean filter(Event value) throws Exception {
-							return value.getName().equals("end");
-						}
-					});
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		return feedNFA(inputEvents, nfa);
-	}
-
-	private static class MyEventIterCondition extends IterativeCondition<Event> {
-
-		private static final long serialVersionUID = 6215754202506583964L;
-
-		@Override
-		public boolean filter(Event value, Context<Event> ctx) throws Exception {
-			if (!value.getName().equals("start")) {
-				return false;
-			}
-
-			double sum = 0.0;
-			for (Event event : ctx.getEventsForPattern("start")) {
-				sum += event.getPrice();
-			}
-			sum += value.getPrice();
-			return Double.compare(sum, 5.0) < 0;
-		}
-	}
-
-	@Test
-	public void testIterativeWithPrevPatternDependency() {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		inputEvents.add(new StreamRecord<>(startEvent1, 1L));
-		inputEvents.add(new StreamRecord<>(startEvent2, 2L));
-		inputEvents.add(new StreamRecord<>(endEvent, 4L));
-
-		Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 6215754202506583964L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("start");
-			}
-		}).oneOrMore().followedBy("end").where(new IterativeCondition<Event>() {
-			private static final long serialVersionUID = 7056763917392056548L;
-
-			@Override
-			public boolean filter(Event value, Context<Event> ctx) throws Exception {
-				if (!value.getName().equals("end")) {
-					return false;
-				}
-
-				double sum = 0.0;
-				for (Event event : ctx.getEventsForPattern("start")) {
-					sum += event.getPrice();
-				}
-				return Double.compare(sum, 2.0) >= 0;
-			}
-		});
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
-
-		compareMaps(resultingPatterns,
-				Lists.<List<Event>>newArrayList(
-						Lists.newArrayList(startEvent1, startEvent2, endEvent),
-						Lists.newArrayList(startEvent2, endEvent)
-				)
-		);
-	}
-
-	@Test
-	public void testIterativeWithABACPattern() {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		inputEvents.add(new StreamRecord<>(startEvent1, 1L)); //1
-		inputEvents.add(new StreamRecord<Event>(middleEvent1, 2L)); //1
-
-		inputEvents.add(new StreamRecord<>(startEvent2, 2L)); //2
-		inputEvents.add(new StreamRecord<>(startEvent3, 2L)); //3
-		inputEvents.add(new StreamRecord<Event>(middleEvent2, 2L)); //2
-
-		inputEvents.add(new StreamRecord<>(startEvent4, 2L)); //4
-		inputEvents.add(new StreamRecord<Event>(middleEvent3, 2L)); //3
-		inputEvents.add(new StreamRecord<Event>(middleEvent4, 2L)); //1
-		inputEvents.add(new StreamRecord<>(endEvent, 4L));
-
-		Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 6215754202506583964L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("start");
-			}
-		}).followedByAny("middle1").subtype(SubEvent.class).where(new SimpleCondition<SubEvent>() {
-			private static final long serialVersionUID = 2178338526904474690L;
-
-			@Override
-			public boolean filter(SubEvent value) throws Exception {
-				return value.getName().startsWith("foo");
-			}
-		}).followedBy("middle2").where(new IterativeCondition<Event>() {
-			private static final long serialVersionUID = -1223388426808292695L;
-
-			@Override
-			public boolean filter(Event value, Context<Event> ctx) throws Exception {
-				if (!value.getName().equals("start")) {
-					return false;
-				}
-
-				double sum = 0.0;
-				for (Event e: ctx.getEventsForPattern("middle2")) {
-					sum += e.getPrice();
-				}
-				sum += value.getPrice();
-				return Double.compare(sum, 5.0) <= 0;
-			}
-		}).oneOrMore().followedBy("end").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 562590474115118323L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("end");
-			}
-		});
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
-
-		compareMaps(resultingPatterns,
-				Lists.<List<Event>>newArrayList(
-						Lists.newArrayList(startEvent1, startEvent2, startEvent3, middleEvent1, endEvent),
-						Lists.newArrayList(startEvent1, middleEvent1, startEvent2, endEvent),
-						Lists.newArrayList(startEvent1, middleEvent2, startEvent4, endEvent),
-						Lists.newArrayList(startEvent2, middleEvent2, startEvent4, endEvent),
-						Lists.newArrayList(startEvent3, middleEvent2, startEvent4, endEvent)
-				)
-		);
-	}
-
-	@Test
-	public void testIterativeWithPrevPatternDependencyAfterBranching() {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		inputEvents.add(new StreamRecord<>(startEvent1, 1L));
-		inputEvents.add(new StreamRecord<>(startEvent2, 2L));
-		inputEvents.add(new StreamRecord<Event>(middleEvent1, 4L));
-		inputEvents.add(new StreamRecord<>(startEvent3, 5L));
-		inputEvents.add(new StreamRecord<Event>(middleEvent2, 6L));
-		inputEvents.add(new StreamRecord<>(endEvent, 7L));
-
-		Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 6215754202506583964L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("start");
-			}
-		}).oneOrMore().followedByAny("middle1").subtype(SubEvent.class).where(new SimpleCondition<SubEvent>() {
-			private static final long serialVersionUID = 2178338526904474690L;
-
-			@Override
-			public boolean filter(SubEvent value) throws Exception {
-				return value.getName().startsWith("foo");
-			}
-		}).followedByAny("end").where(new IterativeCondition<Event>() {
-			private static final long serialVersionUID = 7056763917392056548L;
-
-			@Override
-			public boolean filter(Event value, Context<Event> ctx) throws Exception {
-				if (!value.getName().equals("end")) {
-					return false;
-				}
-
-				double sum = 0.0;
-				for (Event event : ctx.getEventsForPattern("start")) {
-					sum += event.getPrice();
-				}
-				return Double.compare(sum, 2.0) >= 0;
-			}
-		});
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
-
-		compareMaps(resultingPatterns,
-				Lists.<List<Event>>newArrayList(
-						Lists.newArrayList(startEvent1, startEvent2, middleEvent1, endEvent),
-						Lists.newArrayList(startEvent2, middleEvent1, endEvent),
-						Lists.newArrayList(startEvent1, startEvent2, middleEvent2, endEvent),
-						Lists.newArrayList(startEvent1, startEvent2, startEvent3, middleEvent2, endEvent),
-						Lists.newArrayList(startEvent2, startEvent3, middleEvent2, endEvent),
-						Lists.newArrayList(startEvent2, middleEvent2, endEvent),
-						Lists.newArrayList(startEvent3, middleEvent2, endEvent)
-				)
-		);
-	}
-
-
 	///////////////////////////////////////   Skip till next     /////////////////////////////
 
 	@Test
@@ -2809,7 +2431,7 @@ public class NFAITCase extends TestLogger {
 		SubEvent middleEvent3 = new SubEvent(43, "foo3", 1.0, 10.0);
 		SubEvent nextOne1 = new SubEvent(44, "next-one", 1.0, 2.0);
 		SubEvent nextOne2 = new SubEvent(45, "next-one", 1.0, 2.0);
-		Event endEvent=  new Event(46, "end", 1.0);
+		Event endEvent = new Event(46, "end", 1.0);
 
 		inputEvents.add(new StreamRecord<>(startEvent, 1));
 		inputEvents.add(new StreamRecord<Event>(middleEvent1, 3));
@@ -2868,7 +2490,7 @@ public class NFAITCase extends TestLogger {
 		SubEvent middleEvent3 = new SubEvent(43, "foo3", 1.0, 10.0);
 		SubEvent nextOne1 = new SubEvent(44, "next-one", 1.0, 2.0);
 		SubEvent nextOne2 = new SubEvent(45, "next-one", 1.0, 2.0);
-		Event endEvent=  new Event(46, "end", 1.0);
+		Event endEvent = new Event(46, "end", 1.0);
 
 		inputEvents.add(new StreamRecord<>(startEvent, 1));
 		inputEvents.add(new StreamRecord<Event>(middleEvent1, 3));
@@ -2919,1130 +2541,76 @@ public class NFAITCase extends TestLogger {
 		));
 	}
 
-
-	/////////////////////////////////////////       Not pattern    /////////////////////////////////////////////////
-
 	@Test
-	public void testNotNext() {
+	public void testMultipleTakesVersionCollision() {
 		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
 
-		Event a1 = new Event(40, "a", 1.0);
-		Event c1 = new Event(41, "c", 2.0);
-		Event b1 = new Event(42, "b", 3.0);
-		Event c2 = new Event(43, "c", 4.0);
-		Event d = new Event(43, "d", 4.0);
+		Event startEvent = new Event(40, "c", 1.0);
+		Event middleEvent1 = new Event(41, "a", 2.0);
+		Event middleEvent2 = new Event(41, "a", 3.0);
+		Event middleEvent3 = new Event(41, "a", 4.0);
+		Event middleEvent4 = new Event(41, "a", 5.0);
+		Event middleEvent5 = new Event(41, "a", 6.0);
+		Event end = new Event(44, "b", 5.0);
 
-		inputEvents.add(new StreamRecord<>(a1, 1));
-		inputEvents.add(new StreamRecord<>(c1, 2));
-		inputEvents.add(new StreamRecord<>(b1, 3));
-		inputEvents.add(new StreamRecord<>(c2, 4));
-		inputEvents.add(new StreamRecord<>(d, 5));
+		inputEvents.add(new StreamRecord<>(startEvent, 1));
+		inputEvents.add(new StreamRecord<>(middleEvent1, 3));
+		inputEvents.add(new StreamRecord<>(middleEvent2, 4));
+		inputEvents.add(new StreamRecord<>(middleEvent3, 5));
+		inputEvents.add(new StreamRecord<>(middleEvent4, 6));
+		inputEvents.add(new StreamRecord<>(middleEvent5, 7));
+		inputEvents.add(new StreamRecord<>(end, 10));
 
 		Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5167288560432018992L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("a");
-			}
-		}).notNext("notPattern").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 2242479288129905510L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("b");
-			}
-		}).followedByAny("middle").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 1404509325548220892L;
+			private static final long serialVersionUID = 5726188262756267490L;
 
 			@Override
 			public boolean filter(Event value) throws Exception {
 				return value.getName().equals("c");
 			}
-		}).followedBy("end").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = -8907427230007830915L;
+		}).followedBy("middle1").where(new SimpleCondition<Event>() {
+			private static final long serialVersionUID = 5726188262756267490L;
 
 			@Override
 			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("d");
+				return value.getName().equals("a");
 			}
-		});
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		final List<List<Event>> matches = feedNFA(inputEvents, nfa);
-
-		compareMaps(matches, Lists.<List<Event>>newArrayList(
-			Lists.newArrayList(a1, c1, d),
-			Lists.newArrayList(a1, c2, d)
-		));
-	}
-
-	@Test
-	public void testNotNextNoMatches() {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		Event a1 = new Event(40, "a", 1.0);
-		Event b1 = new Event(42, "b", 3.0);
-		Event c1 = new Event(41, "c", 2.0);
-		Event c2 = new Event(43, "c", 4.0);
-		Event d = new Event(43, "d", 4.0);
-
-		inputEvents.add(new StreamRecord<>(a1, 1));
-		inputEvents.add(new StreamRecord<>(b1, 2));
-		inputEvents.add(new StreamRecord<>(c1, 3));
-		inputEvents.add(new StreamRecord<>(c2, 4));
-		inputEvents.add(new StreamRecord<>(d, 5));
-
-		Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = -339500190577666439L;
+		}).oneOrMore().allowCombinations().followedBy("middle2").where(new SimpleCondition<Event>() {
+			private static final long serialVersionUID = 5726188262756267490L;
 
 			@Override
 			public boolean filter(Event value) throws Exception {
 				return value.getName().equals("a");
 			}
-		}).notNext("notPattern").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = -6913980632538046451L;
+		}).oneOrMore().allowCombinations().followedBy("end").where(new SimpleCondition<Event>() {
+			private static final long serialVersionUID = 5726188262756267490L;
 
 			@Override
 			public boolean filter(Event value) throws Exception {
 				return value.getName().equals("b");
 			}
-		}).followedBy("middle").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 3332196998905139891L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("c");
-			}
-		}).followedBy("end").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 2086563479959018387L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("d");
-			}
 		});
 
 		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
 
-		final List<List<Event>> matches = feedNFA(inputEvents, nfa);
-
-		assertEquals(0, matches.size());
-	}
+		final List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
 
-	@Test
-	public void testNotNextNoMatchesAtTheEnd() {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
+		compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
 
-		Event a1 = new Event(40, "a", 1.0);
-		Event c1 = new Event(41, "c", 2.0);
-		Event c2 = new Event(43, "c", 4.0);
-		Event d = new Event(43, "d", 4.0);
-		Event b1 = new Event(42, "b", 3.0);
+			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent4, middleEvent5, end),
+			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent4, middleEvent5, end),
+			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent4, middleEvent5, end),
+			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent4, middleEvent5, end),
 
-		inputEvents.add(new StreamRecord<>(a1, 1));
-		inputEvents.add(new StreamRecord<>(c1, 2));
-		inputEvents.add(new StreamRecord<>(c2, 3));
-		inputEvents.add(new StreamRecord<>(d, 4));
-		inputEvents.add(new StreamRecord<>(b1, 5));
-
-		Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 1672995058886176627L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("a");
-			}
-		}).followedByAny("middle").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 6003621617520261554L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("c");
-			}
-		}).followedByAny("end").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 887700237024758417L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("d");
-			}
-		}).notNext("notPattern").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5239529076086933032L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("b");
-			}
-		});
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		final List<List<Event>> matches = feedNFA(inputEvents, nfa);
-
-		assertEquals(0, matches.size());
-	}
-
-	@Test
-	public void testNotFollowedBy() {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		Event a1 = new Event(40, "a", 1.0);
-		Event c1 = new Event(41, "c", 2.0);
-		Event b1 = new Event(42, "b", 3.0);
-		Event c2 = new Event(43, "c", 4.0);
-		Event d = new Event(43, "d", 4.0);
-
-		inputEvents.add(new StreamRecord<>(a1, 1));
-		inputEvents.add(new StreamRecord<>(c1, 2));
-		inputEvents.add(new StreamRecord<>(b1, 3));
-		inputEvents.add(new StreamRecord<>(c2, 4));
-		inputEvents.add(new StreamRecord<>(d, 5));
-
-		Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = -2641662468313191976L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("a");
-			}
-		}).notFollowedBy("notPattern").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = -3632144132379494778L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("b");
-			}
-		}).followedByAny("middle").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 3818766882138348167L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("c");
-			}
-		}).followedBy("end").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 2033204730795451288L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("d");
-			}
-		});
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		final List<List<Event>> matches = feedNFA(inputEvents, nfa);
-
-		compareMaps(matches,Lists.<List<Event>>newArrayList(
-			Lists.newArrayList(a1, c1, d)
-		));
-	}
-
-	@Test
-	public void testNotFollowedByBeforeOptional() {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		Event a1 = new Event(40, "a", 1.0);
-		Event c1 = new Event(41, "c", 2.0);
-		Event b1 = new Event(42, "b", 3.0);
-		Event c2 = new Event(43, "c", 4.0);
-		Event d = new Event(43, "d", 4.0);
-
-		inputEvents.add(new StreamRecord<>(a1, 1));
-		inputEvents.add(new StreamRecord<>(c1, 2));
-		inputEvents.add(new StreamRecord<>(b1, 3));
-		inputEvents.add(new StreamRecord<>(c2, 4));
-		inputEvents.add(new StreamRecord<>(d, 5));
-
-		Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = -2454396370205097543L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("a");
-			}
-		}).notFollowedBy("notPattern").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 2749547391611263290L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("b");
-			}
-		}).followedByAny("middle").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = -4989511337298217255L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("c");
-			}
-		}).optional().followedBy("end").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = -8466223836652936608L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("d");
-			}
-		});
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		final List<List<Event>> matches = feedNFA(inputEvents, nfa);
-
-		compareMaps(matches,Lists.<List<Event>>newArrayList(
-			Lists.newArrayList(a1, c1, d)
-		));
-	}
-
-	@Test
-	public void testTimesWithNotFollowedBy() {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		Event a1 = new Event(40, "a", 1.0);
-		Event b1 = new Event(41, "b", 2.0);
-		Event c = new Event(42, "c", 3.0);
-		Event b2 = new Event(43, "b", 4.0);
-		Event d = new Event(43, "d", 4.0);
-
-		inputEvents.add(new StreamRecord<>(a1, 1));
-		inputEvents.add(new StreamRecord<>(b1, 2));
-		inputEvents.add(new StreamRecord<>(c, 3));
-		inputEvents.add(new StreamRecord<>(b2, 4));
-		inputEvents.add(new StreamRecord<>(d, 5));
-
-		Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = -2568839911852184515L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("a");
-			}
-		}).followedByAny("middle").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = -3632232424064269636L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("b");
-			}
-		}).times(2).notFollowedBy("notPattern").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 3685596793523534611L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("c");
-			}
-		}).followedBy("end").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 1960758663575587243L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("d");
-			}
-		});
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		final List<List<Event>> matches = feedNFA(inputEvents, nfa);
-
-		compareMaps(matches,Lists.<List<Event>>newArrayList());
-	}
-
-	@Test
-	public void testIgnoreStateOfTimesWithNotFollowedBy() {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		Event a1 = new Event(40, "a", 1.0);
-		Event e = new Event(41, "e", 2.0);
-		Event c1 = new Event(42, "c", 3.0);
-		Event b1 = new Event(43, "b", 4.0);
-		Event c2 = new Event(44, "c", 5.0);
-		Event d1 = new Event(45, "d", 6.0);
-		Event d2 = new Event(46, "d", 7.0);
-
-		inputEvents.add(new StreamRecord<>(a1, 1));
-		inputEvents.add(new StreamRecord<>(d1, 2));
-		inputEvents.add(new StreamRecord<>(e, 1));
-		inputEvents.add(new StreamRecord<>(b1, 3));
-		inputEvents.add(new StreamRecord<>(c1, 2));
-		inputEvents.add(new StreamRecord<>(c2, 4));
-		inputEvents.add(new StreamRecord<>(d2, 5));
-
-		Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 2814850350025111940L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("a");
-			}
-		}).notFollowedBy("notPattern").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 4988756153568853834L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("b");
-			}
-		}).followedByAny("middle").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = -225909103322018778L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("c");
-			}
-		}).times(2).optional().followedBy("end").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = -924294627956373696L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("d");
-			}
-		});
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		final List<List<Event>> matches = feedNFA(inputEvents, nfa);
-
-		compareMaps(matches, Lists.<List<Event>>newArrayList(
-			Lists.newArrayList(a1, d1)
-		));
-	}
-
-	@Test
-	public void testTimesWithNotFollowedByAfter() {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		Event a1 = new Event(40, "a", 1.0);
-		Event e = new Event(41, "e", 2.0);
-		Event c1 = new Event(42, "c", 3.0);
-		Event b1 = new Event(43, "b", 4.0);
-		Event b2 = new Event(44, "b", 5.0);
-		Event d1 = new Event(46, "d", 7.0);
-		Event d2 = new Event(47, "d", 8.0);
-
-		inputEvents.add(new StreamRecord<>(a1, 1));
-		inputEvents.add(new StreamRecord<>(d1, 2));
-		inputEvents.add(new StreamRecord<>(e, 1));
-		inputEvents.add(new StreamRecord<>(b1, 3));
-		inputEvents.add(new StreamRecord<>(b2, 3));
-		inputEvents.add(new StreamRecord<>(c1, 2));
-		inputEvents.add(new StreamRecord<>(d2, 5));
-
-		Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 6193105689601702341L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("a");
-			}
-		}).followedByAny("middle").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5195859580923169111L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("b");
-			}
-		}).times(2).notFollowedBy("notPattern").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 4973027956103783831L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("c");
-			}
-		}).followedBy("end").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 2724622546678984894L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("d");
-			}
-		});
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		final List<List<Event>> matches = feedNFA(inputEvents, nfa);
-
-		compareMaps(matches, Lists.<List<Event>>newArrayList());
-	}
-
-	@Test
-	public void testNotFollowedByBeforeOptionalAtTheEnd() {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		Event a1 = new Event(40, "a", 1.0);
-		Event c1 = new Event(41, "c", 2.0);
-		Event b1 = new Event(42, "b", 3.0);
-		Event c2 = new Event(43, "c", 4.0);
-
-		inputEvents.add(new StreamRecord<>(a1, 1));
-		inputEvents.add(new StreamRecord<>(c1, 2));
-		inputEvents.add(new StreamRecord<>(b1, 3));
-		inputEvents.add(new StreamRecord<>(c2, 4));
-
-		Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = -4289351792573443294L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("a");
-			}
-		}).notFollowedBy("notPattern").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = -4989574608417523507L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("b");
-			}
-		}).followedByAny("end").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = -5940131818629290579L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("c");
-			}
-		}).optional();
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		final List<List<Event>> matches = feedNFA(inputEvents, nfa);
-
-		compareMaps(matches,Lists.<List<Event>>newArrayList(
-			Lists.newArrayList(a1, c1),
-			Lists.newArrayList(a1)
-		));
-	}
-
-	@Test
-	public void testNotFollowedByBeforeOptionalTimes() {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		Event a1 = new Event(40, "a", 1.0);
-		Event c1 = new Event(41, "c", 2.0);
-		Event b1 = new Event(42, "b", 3.0);
-		Event c2 = new Event(43, "c", 4.0);
-		Event d = new Event(43, "d", 4.0);
-
-		inputEvents.add(new StreamRecord<>(a1, 1));
-		inputEvents.add(new StreamRecord<>(c1, 2));
-		inputEvents.add(new StreamRecord<>(b1, 3));
-		inputEvents.add(new StreamRecord<>(c2, 4));
-		inputEvents.add(new StreamRecord<>(d, 5));
-
-		Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = -7885381452276160322L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("a");
-			}
-		}).notFollowedBy("notPattern").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 3471511260235826653L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("b");
-			}
-		}).followedByAny("middle").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 9073793782452363833L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("c");
-			}
-		}).times(2).optional().followedBy("end").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 7972902718259767076L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("d");
-			}
-		});
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		final List<List<Event>> matches = feedNFA(inputEvents, nfa);
-
-		compareMaps(matches,Lists.<List<Event>>newArrayList(
-			Lists.newArrayList(a1, c1, c2, d)
-		));
-	}
-
-	@Test
-	public void testNotFollowedByWithBranchingAtStart() {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		Event a1 = new Event(40, "a", 1.0);
-		Event b1 = new Event(42, "b", 3.0);
-		Event c1 = new Event(41, "c", 2.0);
-		Event a2 = new Event(41, "a", 4.0);
-		Event c2 = new Event(43, "c", 5.0);
-		Event d = new Event(43, "d", 6.0);
-
-		inputEvents.add(new StreamRecord<>(a1, 1));
-		inputEvents.add(new StreamRecord<>(b1, 2));
-		inputEvents.add(new StreamRecord<>(c1, 3));
-		inputEvents.add(new StreamRecord<>(a2, 4));
-		inputEvents.add(new StreamRecord<>(c2, 5));
-		inputEvents.add(new StreamRecord<>(d, 6));
-
-		Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = -7866220136345465444L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("a");
-			}
-		}).notFollowedBy("notPattern").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 4957837489028234932L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("b");
-			}
-		}).followedBy("middle").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5569569968862808007L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("c");
-			}
-		}).followedBy("end").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = -8579678167937416269L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("d");
-			}
-		});
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		final List<List<Event>> matches = feedNFA(inputEvents, nfa);
-
-		compareMaps(matches, Lists.<List<Event>>newArrayList(
-			Lists.newArrayList(a2, c2, d)
-		));
-	}
-
-	private static class NotFollowByData {
-		static final Event a1 = new Event(40, "a", 1.0);
-		static final Event b1 = new Event(41, "b", 2.0);
-		static final Event b2 = new Event(42, "b", 3.0);
-		static final Event b3 = new Event(42, "b", 4.0);
-		static final Event c1 = new Event(43, "c", 5.0);
-		static final Event b4 = new Event(42, "b", 6.0);
-		static final Event b5 = new Event(42, "b", 7.0);
-		static final Event b6 = new Event(42, "b", 8.0);
-		static final Event d1 = new Event(43, "d", 9.0);
-
-		private NotFollowByData() {
-		}
-	}
-
-	@Test
-	public void testNotNextAfterOneOrMoreSkipTillNext() {
-		final List<List<Event>> matches = testNotNextAfterOneOrMore(false);
-		assertEquals(0, matches.size());
-	}
-
-	@Test
-	public void testNotNextAfterOneOrMoreSkipTillAny() {
-		final List<List<Event>> matches = testNotNextAfterOneOrMore(true);
-		compareMaps(matches, Lists.<List<Event>>newArrayList(
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b2, NotFollowByData.d1)
-		));
-	}
-
-	private List<List<Event>> testNotNextAfterOneOrMore(boolean allMatches) {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		int i = 0;
-		inputEvents.add(new StreamRecord<>(NotFollowByData.a1, i++));
-		inputEvents.add(new StreamRecord<>(NotFollowByData.b1, i++));
-		inputEvents.add(new StreamRecord<>(NotFollowByData.c1, i++));
-		inputEvents.add(new StreamRecord<>(NotFollowByData.b2, i++));
-		inputEvents.add(new StreamRecord<>(NotFollowByData.d1, i++));
-
-		Pattern<Event, ?> pattern = Pattern
-			.<Event>begin("a").where(new SimpleCondition<Event>() {
-				private static final long serialVersionUID = 5726188262756267490L;
-
-				@Override
-				public boolean filter(Event value) throws Exception {
-					return value.getName().equals("a");
-				}
-			});
-
-		pattern = (allMatches ? pattern.followedByAny("b*") : pattern.followedBy("b*")).where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5726188262756267490L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("b");
-			}
-		}).oneOrMore()
-			.notNext("not c").where(new SimpleCondition<Event>() {
-				private static final long serialVersionUID = 5726188262756267490L;
-
-				@Override
-				public boolean filter(Event value) throws Exception {
-					return value.getName().equals("c");
-				}
-			})
-			.followedBy("d").where(new SimpleCondition<Event>() {
-				private static final long serialVersionUID = 5726188262756267490L;
-
-				@Override
-				public boolean filter(Event value) throws Exception {
-					return value.getName().equals("d");
-				}
-			});
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		return feedNFA(inputEvents, nfa);
-	}
-
-	@Test
-	public void testNotFollowedByNextAfterOneOrMoreEager() {
-		final List<List<Event>> matches = testNotFollowedByAfterOneOrMore(true, false);
-		assertEquals(0, matches.size());
-	}
-
-	@Test
-	public void testNotFollowedByAnyAfterOneOrMoreEager() {
-		final List<List<Event>> matches = testNotFollowedByAfterOneOrMore(true, true);
-		compareMaps(matches, Lists.<List<Event>>newArrayList(
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b4, NotFollowByData.b5, NotFollowByData.b6, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b4, NotFollowByData.b5, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b4, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b5, NotFollowByData.b6, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b5, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b6, NotFollowByData.d1)
-		));
-	}
-
-	@Test
-	public void testNotFollowedByNextAfterOneOrMoreCombinations() {
-		final List<List<Event>> matches = testNotFollowedByAfterOneOrMore(false, false);
-		assertEquals(0, matches.size());
-	}
-
-	@Test
-	public void testNotFollowedByAnyAfterOneOrMoreCombinations() {
-		final List<List<Event>> matches = testNotFollowedByAfterOneOrMore(false, true);
-		compareMaps(matches, Lists.<List<Event>>newArrayList(
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b4, NotFollowByData.b5, NotFollowByData.b6, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b4, NotFollowByData.b5, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b4, NotFollowByData.b6, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b4, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b5, NotFollowByData.b6, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b5, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b6, NotFollowByData.d1)
-		));
-	}
-
-	private List<List<Event>> testNotFollowedByAfterOneOrMore(boolean eager, boolean allMatches) {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		int i = 0;
-		inputEvents.add(new StreamRecord<>(NotFollowByData.a1, i++));
-		inputEvents.add(new StreamRecord<>(NotFollowByData.b1, i++));
-		inputEvents.add(new StreamRecord<>(NotFollowByData.b2, i++));
-		inputEvents.add(new StreamRecord<>(NotFollowByData.b3, i++));
-		inputEvents.add(new StreamRecord<>(NotFollowByData.c1, i++));
-		inputEvents.add(new StreamRecord<>(NotFollowByData.b4, i++));
-		inputEvents.add(new StreamRecord<>(NotFollowByData.b5, i++));
-		inputEvents.add(new StreamRecord<>(NotFollowByData.b6, i++));
-		inputEvents.add(new StreamRecord<>(NotFollowByData.d1, i));
-
-		Pattern<Event, ?> pattern = Pattern
-			.<Event>begin("a").where(new SimpleCondition<Event>() {
-				private static final long serialVersionUID = 5726188262756267490L;
-
-				@Override
-				public boolean filter(Event value) throws Exception {
-					return value.getName().equals("a");
-				}
-			});
-
-		pattern = (allMatches ? pattern.followedByAny("b*") : pattern.followedBy("b*"))
-			.where(new SimpleCondition<Event>() {
-				private static final long serialVersionUID = 5726188262756267490L;
-
-				@Override
-				public boolean filter(Event value) throws Exception {
-					return value.getName().equals("b");
-				}
-			});
-
-		pattern = (eager ? pattern.oneOrMore() : pattern.oneOrMore().allowCombinations())
-			.notFollowedBy("not c").where(new SimpleCondition<Event>() {
-				private static final long serialVersionUID = 5726188262756267490L;
-
-				@Override
-				public boolean filter(Event value) throws Exception {
-					return value.getName().equals("c");
-				}
-			})
-			.followedBy("d").where(new SimpleCondition<Event>() {
-				private static final long serialVersionUID = 5726188262756267490L;
-
-				@Override
-				public boolean filter(Event value) throws Exception {
-					return value.getName().equals("d");
-				}
-			});
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		return feedNFA(inputEvents, nfa);
-	}
-
-	@Test
-	public void testNotFollowedByAnyBeforeOneOrMoreEager() {
-		final List<List<Event>> matches = testNotFollowedByBeforeOneOrMore(true, true);
-
-		compareMaps(matches, Lists.<List<Event>>newArrayList(
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.b5, NotFollowByData.b6, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.b5, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.d1)
-		));
-	}
-
-	@Test
-	public void testNotFollowedByAnyBeforeOneOrMoreCombinations() {
-		final List<List<Event>> matches = testNotFollowedByBeforeOneOrMore(false, true);
-
-		compareMaps(matches, Lists.<List<Event>>newArrayList(
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.b5, NotFollowByData.b6, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.b6, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.b5, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b5, NotFollowByData.b6, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b5, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b6, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.d1)
-		));
-	}
-
-	@Test
-	public void testNotFollowedByBeforeOneOrMoreEager() {
-		final List<List<Event>> matches = testNotFollowedByBeforeOneOrMore(true, false);
-
-		compareMaps(matches, Lists.<List<Event>>newArrayList(
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.b5, NotFollowByData.b6, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.b5, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.d1)
-		));
-	}
-
-	@Test
-	public void testNotFollowedByBeforeOneOrMoreCombinations() {
-		final List<List<Event>> matches = testNotFollowedByBeforeOneOrMore(false, false);
-
-		compareMaps(matches, Lists.<List<Event>>newArrayList(
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.b5, NotFollowByData.b6, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.b6, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.b5, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b5, NotFollowByData.b6, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b5, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b6, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.d1)
-		));
-	}
-
-	private List<List<Event>> testNotFollowedByBeforeOneOrMore(boolean eager, boolean allMatches) {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		int i = 0;
-		inputEvents.add(new StreamRecord<>(NotFollowByData.a1, i++));
-		inputEvents.add(new StreamRecord<>(NotFollowByData.b1, i++));
-		inputEvents.add(new StreamRecord<>(NotFollowByData.c1, i++));
-		inputEvents.add(new StreamRecord<>(NotFollowByData.b4, i++));
-		inputEvents.add(new StreamRecord<>(NotFollowByData.b5, i++));
-		inputEvents.add(new StreamRecord<>(NotFollowByData.b6, i++));
-		inputEvents.add(new StreamRecord<>(NotFollowByData.d1, i));
-
-		Pattern<Event, ?> pattern = Pattern
-			.<Event>begin("a").where(new SimpleCondition<Event>() {
-				private static final long serialVersionUID = 5726188262756267490L;
-
-				@Override
-				public boolean filter(Event value) throws Exception {
-					return value.getName().equals("a");
-				}
-			})
-			.notFollowedBy("not c").where(new SimpleCondition<Event>() {
-				private static final long serialVersionUID = 5726188262756267490L;
-
-				@Override
-				public boolean filter(Event value) throws Exception {
-					return value.getName().equals("c");
-				}
-			});
-
-		pattern = (allMatches ? pattern.followedByAny("b*") : pattern.followedBy("b*"))
-			.where(new SimpleCondition<Event>() {
-				private static final long serialVersionUID = 5726188262756267490L;
-
-				@Override
-				public boolean filter(Event value) throws Exception {
-					return value.getName().equals("b");
-				}
-			}).oneOrMore();
-
-		pattern = (eager ? pattern : pattern.allowCombinations())
-			.followedBy("d").where(new SimpleCondition<Event>() {
-				private static final long serialVersionUID = 5726188262756267490L;
-
-				@Override
-				public boolean filter(Event value) throws Exception {
-					return value.getName().equals("d");
-				}
-			});
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		return feedNFA(inputEvents, nfa);
-	}
-
-	@Test
-	public void testNotFollowedByBeforeZeroOrMoreEagerSkipTillNext() {
-		final List<List<Event>> matches = testNotFollowedByBeforeZeroOrMore(true, false);
-		compareMaps(matches, Lists.<List<Event>>newArrayList(
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.b5, NotFollowByData.b6, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.b5, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.d1)
-		));
-	}
-
-	@Test
-	public void testNotFollowedByBeforeZeroOrMoreCombinationsSkipTillNext() {
-		final List<List<Event>> matches = testNotFollowedByBeforeZeroOrMore(false, false);
-		compareMaps(matches, Lists.<List<Event>>newArrayList(
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.b5, NotFollowByData.b6, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.b5, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.b6, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b5, NotFollowByData.b6, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b5, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b6, NotFollowByData.d1)
-		));
-	}
-
-	@Test
-	public void testNotFollowedByBeforeZeroOrMoreEagerSkipTillAny() {
-		final List<List<Event>> matches = testNotFollowedByBeforeZeroOrMore(true, true);
-		compareMaps(matches, Lists.<List<Event>>newArrayList(
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.b5, NotFollowByData.b6, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.b5, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.d1)
-		));
-	}
-
-	@Test
-	public void testNotFollowedByBeforeZeroOrMoreCombinationsSkipTillAny() {
-		final List<List<Event>> matches = testNotFollowedByBeforeZeroOrMore(false, true);
-		compareMaps(matches, Lists.<List<Event>>newArrayList(
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.b5, NotFollowByData.b6, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.b5, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.b6, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b4, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b5, NotFollowByData.b6, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b5, NotFollowByData.d1),
-			Lists.newArrayList(NotFollowByData.a1, NotFollowByData.b1, NotFollowByData.b6, NotFollowByData.d1)
-		));
-	}
-
-	private List<List<Event>> testNotFollowedByBeforeZeroOrMore(boolean eager, boolean allMatches) {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		int i = 0;
-		inputEvents.add(new StreamRecord<>(NotFollowByData.a1, i++));
-		inputEvents.add(new StreamRecord<>(NotFollowByData.b1, i++));
-		inputEvents.add(new StreamRecord<>(NotFollowByData.c1, i++));
-		inputEvents.add(new StreamRecord<>(NotFollowByData.b4, i++));
-		inputEvents.add(new StreamRecord<>(NotFollowByData.b5, i++));
-		inputEvents.add(new StreamRecord<>(NotFollowByData.b6, i++));
-		inputEvents.add(new StreamRecord<>(NotFollowByData.d1, i));
-
-		Pattern<Event, ?> pattern = Pattern
-			.<Event>begin("a").where(new SimpleCondition<Event>() {
-				private static final long serialVersionUID = 5726188262756267490L;
-
-				@Override
-				public boolean filter(Event value) throws Exception {
-					return value.getName().equals("a");
-				}
-			})
-			.notFollowedBy("not c").where(new SimpleCondition<Event>() {
-				private static final long serialVersionUID = 5726188262756267490L;
-
-				@Override
-				public boolean filter(Event value) throws Exception {
-					return value.getName().equals("c");
-				}
-			});
-
-		pattern = (allMatches ? pattern.followedByAny("b*") : pattern.followedBy("b*"))
-			.where(new SimpleCondition<Event>() {
-				private static final long serialVersionUID = 5726188262756267490L;
-
-				@Override
-				public boolean filter(Event value) throws Exception {
-					return value.getName().equals("b");
-				}
-			}).oneOrMore().optional();
-
-		pattern = (eager ? pattern : pattern.allowCombinations())
-			.followedBy("d").where(new SimpleCondition<Event>() {
-				private static final long serialVersionUID = 5726188262756267490L;
-
-				@Override
-				public boolean filter(Event value) throws Exception {
-					return value.getName().equals("d");
-				}
-			});
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		return feedNFA(inputEvents, nfa);
-	}
-
-	@Test
-	public void testEagerZeroOrMoreSameElement() {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		Event startEvent = new Event(40, "c", 1.0);
-		Event middleEvent1 = new Event(41, "a", 2.0);
-		Event middleEvent2 = new Event(42, "a", 3.0);
-		Event middleEvent3 = new Event(43, "a", 4.0);
-		Event end1 = new Event(44, "b", 5.0);
-
-		inputEvents.add(new StreamRecord<>(startEvent, 1));
-		inputEvents.add(new StreamRecord<>(middleEvent1, 3));
-		inputEvents.add(new StreamRecord<>(middleEvent1, 3));
-		inputEvents.add(new StreamRecord<>(middleEvent1, 3));
-		inputEvents.add(new StreamRecord<>(middleEvent2, 4));
-		inputEvents.add(new StreamRecord<>(new Event(50, "d", 6.0), 5));
-		inputEvents.add(new StreamRecord<>(middleEvent3, 6));
-		inputEvents.add(new StreamRecord<>(middleEvent3, 6));
-		inputEvents.add(new StreamRecord<>(end1, 7));
-
-		Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5726188262756267490L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("c");
-			}
-		}).followedBy("middle").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5726188262756267490L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("a");
-			}
-		}).oneOrMore().optional().followedBy("end1").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5726188262756267490L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("b");
-			}
-		});
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		final List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
-
-		compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
-				Lists.newArrayList(startEvent, middleEvent1, middleEvent1, middleEvent1, middleEvent2, middleEvent3, middleEvent3, end1),
-				Lists.newArrayList(startEvent, middleEvent1, middleEvent1, middleEvent1, middleEvent2, middleEvent3, end1),
-				Lists.newArrayList(startEvent, middleEvent1, middleEvent1, middleEvent1, middleEvent2, end1),
-				Lists.newArrayList(startEvent, middleEvent1, middleEvent1, middleEvent1, end1),
-				Lists.newArrayList(startEvent, middleEvent1, middleEvent1, end1),
-				Lists.newArrayList(startEvent, middleEvent1, end1),
-				Lists.newArrayList(startEvent, end1)
-		));
-	}
-
-	@Test
-	public void testMultipleTakesVersionCollision() {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		Event startEvent = new Event(40, "c", 1.0);
-		Event middleEvent1 = new Event(41, "a", 2.0);
-		Event middleEvent2 = new Event(41, "a", 3.0);
-		Event middleEvent3 = new Event(41, "a", 4.0);
-		Event middleEvent4 = new Event(41, "a", 5.0);
-		Event middleEvent5 = new Event(41, "a", 6.0);
-		Event end = new Event(44, "b", 5.0);
-
-		inputEvents.add(new StreamRecord<>(startEvent, 1));
-		inputEvents.add(new StreamRecord<>(middleEvent1, 3));
-		inputEvents.add(new StreamRecord<>(middleEvent2, 4));
-		inputEvents.add(new StreamRecord<>(middleEvent3, 5));
-		inputEvents.add(new StreamRecord<>(middleEvent4, 6));
-		inputEvents.add(new StreamRecord<>(middleEvent5, 7));
-		inputEvents.add(new StreamRecord<>(end, 10));
-
-		Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5726188262756267490L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("c");
-			}
-		}).followedBy("middle1").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5726188262756267490L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("a");
-			}
-		}).oneOrMore().allowCombinations().followedBy("middle2").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5726188262756267490L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("a");
-			}
-		}).oneOrMore().allowCombinations().followedBy("end").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5726188262756267490L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("b");
-			}
-		});
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		final List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
-
-		compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
-
-			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent4, middleEvent5, end),
-			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent4, middleEvent5, end),
-			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent4, middleEvent5, end),
-			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent4, middleEvent5, end),
-
-			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent4, end),
-			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent4, middleEvent5, end),
-			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent4, end),
-			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent5, end),
-			Lists.newArrayList(startEvent, middleEvent1, middleEvent3, middleEvent4, middleEvent5, end),
-			Lists.newArrayList(startEvent, middleEvent1, middleEvent3, middleEvent4, middleEvent5, end),
-			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent4, end),
-			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent5, end),
-			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent4, middleEvent5, end),
+			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent4, end),
+			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent4, middleEvent5, end),
+			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent4, end),
+			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent5, end),
+			Lists.newArrayList(startEvent, middleEvent1, middleEvent3, middleEvent4, middleEvent5, end),
+			Lists.newArrayList(startEvent, middleEvent1, middleEvent3, middleEvent4, middleEvent5, end),
+			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent4, end),
+			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent5, end),
+			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent4, middleEvent5, end),
 
 			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, end),
 			Lists.newArrayList(startEvent, middleEvent1, middleEvent3, middleEvent4, end),
@@ -4054,386 +2622,4 @@ public class NFAITCase extends TestLogger {
 			Lists.newArrayList(startEvent, middleEvent1, middleEvent2, end)
 			));
 	}
-
-	@Test
-	public void testZeroOrMoreSameElement() {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		Event startEvent = new Event(40, "c", 1.0);
-		Event middleEvent1 = new Event(41, "a", 2.0);
-		Event middleEvent1a = new Event(41, "a", 2.0);
-		Event middleEvent2 = new Event(42, "a", 3.0);
-		Event middleEvent3 = new Event(43, "a", 4.0);
-		Event middleEvent3a = new Event(43, "a", 4.0);
-		Event end1 = new Event(44, "b", 5.0);
-
-		inputEvents.add(new StreamRecord<>(startEvent, 1));
-		inputEvents.add(new StreamRecord<>(middleEvent1, 3));
-		inputEvents.add(new StreamRecord<>(middleEvent1a, 3));
-		inputEvents.add(new StreamRecord<>(middleEvent2, 4));
-		inputEvents.add(new StreamRecord<>(new Event(50, "d", 6.0), 5));
-		inputEvents.add(new StreamRecord<>(middleEvent3, 6));
-		inputEvents.add(new StreamRecord<>(middleEvent3a, 6));
-		inputEvents.add(new StreamRecord<>(end1, 7));
-
-		Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5726188262756267490L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("c");
-			}
-		}).followedByAny("middle").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5726188262756267490L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("a");
-			}
-		}).oneOrMore().optional().allowCombinations().followedByAny("end1").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5726188262756267490L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("b");
-			}
-		});
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		final List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
-
-		compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
-				Lists.newArrayList(startEvent, middleEvent1, middleEvent1a, middleEvent2, middleEvent3, middleEvent3a, end1),
-
-				Lists.newArrayList(startEvent, middleEvent1, middleEvent1a, middleEvent2, middleEvent3, end1),
-				Lists.newArrayList(startEvent, middleEvent1, middleEvent1a, middleEvent2, middleEvent3a, end1),
-				Lists.newArrayList(startEvent, middleEvent1, middleEvent1a, middleEvent3, middleEvent3a, end1),
-				Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, middleEvent3a, end1),
-				Lists.newArrayList(startEvent, middleEvent1a, middleEvent2, middleEvent3, middleEvent3a, end1),
-
-				Lists.newArrayList(startEvent, middleEvent1, middleEvent1a, middleEvent2, end1),
-				Lists.newArrayList(startEvent, middleEvent1, middleEvent1a, middleEvent3, end1),
-				Lists.newArrayList(startEvent, middleEvent1, middleEvent1a, middleEvent3a, end1),
-				Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, end1),
-				Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3a, end1),
-				Lists.newArrayList(startEvent, middleEvent1, middleEvent3, middleEvent3a, end1),
-				Lists.newArrayList(startEvent, middleEvent2, middleEvent3, middleEvent3a, end1),
-				Lists.newArrayList(startEvent, middleEvent1a, middleEvent2, middleEvent3, end1),
-				Lists.newArrayList(startEvent, middleEvent1a, middleEvent2, middleEvent3a, end1),
-				Lists.newArrayList(startEvent, middleEvent1a, middleEvent3, middleEvent3a, end1),
-
-				Lists.newArrayList(startEvent, middleEvent1, middleEvent1, end1),
-				Lists.newArrayList(startEvent, middleEvent1, middleEvent2, end1),
-				Lists.newArrayList(startEvent, middleEvent1, middleEvent3, end1),
-				Lists.newArrayList(startEvent, middleEvent1, middleEvent3a, end1),
-				Lists.newArrayList(startEvent, middleEvent1a, middleEvent2, end1),
-				Lists.newArrayList(startEvent, middleEvent1a, middleEvent3, end1),
-				Lists.newArrayList(startEvent, middleEvent1a, middleEvent3a, end1),
-				Lists.newArrayList(startEvent, middleEvent2, middleEvent3, end1),
-				Lists.newArrayList(startEvent, middleEvent2, middleEvent3a, end1),
-				Lists.newArrayList(startEvent, middleEvent3, middleEvent3a, end1),
-
-				Lists.newArrayList(startEvent, middleEvent1, end1),
-				Lists.newArrayList(startEvent, middleEvent1a, end1),
-				Lists.newArrayList(startEvent, middleEvent2, end1),
-				Lists.newArrayList(startEvent, middleEvent3, end1),
-				Lists.newArrayList(startEvent, middleEvent3a, end1),
-
-				Lists.newArrayList(startEvent, end1)
-		));
-	}
-
-	@Test
-	public void testSimplePatternWSameElement() throws Exception {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		Event startEvent = new Event(40, "c", 1.0);
-		Event middleEvent1 = new Event(41, "a", 2.0);
-		Event end1 = new Event(44, "b", 5.0);
-
-		inputEvents.add(new StreamRecord<>(startEvent, 1));
-		inputEvents.add(new StreamRecord<>(middleEvent1, 3));
-		inputEvents.add(new StreamRecord<>(middleEvent1, 3));
-		inputEvents.add(new StreamRecord<>(end1, 7));
-
-		Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5726188262756267490L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("c");
-			}
-		}).followedByAny("middle").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5726188262756267490L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("a");
-			}
-		}).followedBy("end1").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5726188262756267490L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("b");
-			}
-		});
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		final List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
-
-		compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
-				Lists.newArrayList(startEvent, middleEvent1, end1),
-				Lists.newArrayList(startEvent, middleEvent1, end1)
-		));
-	}
-
-	@Test
-	public void testIterativeConditionWSameElement() throws Exception {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		Event startEvent = new Event(40, "c", 1.0);
-		Event middleEvent1 = new Event(41, "a", 2.0);
-		Event middleEvent1a = new Event(41, "a", 2.0);
-		Event middleEvent1b = new Event(41, "a", 2.0);
-		final Event end = new Event(44, "b", 5.0);
-
-		inputEvents.add(new StreamRecord<>(startEvent, 1));
-		inputEvents.add(new StreamRecord<>(middleEvent1, 3));
-		inputEvents.add(new StreamRecord<>(middleEvent1a, 3));
-		inputEvents.add(new StreamRecord<>(middleEvent1b, 3));
-		inputEvents.add(new StreamRecord<>(end, 7));
-
-		Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5726188262756267490L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("c");
-			}
-		}).followedByAny("middle").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5726188262756267490L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("a");
-			}
-		}).oneOrMore().optional().allowCombinations().followedBy("end").where(new IterativeCondition<Event>() {
-
-			private static final long serialVersionUID = -5566639743229703237L;
-
-			@Override
-			public boolean filter(Event value, Context<Event> ctx) throws Exception {
-				double sum = 0.0;
-				for (Event event: ctx.getEventsForPattern("middle")) {
-					sum += event.getPrice();
-				}
-				return Double.compare(sum, 4.0) == 0;
-			}
-
-		});
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		final List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
-
-		compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
-				Lists.newArrayList(startEvent, middleEvent1, middleEvent1a, end),
-				Lists.newArrayList(startEvent, middleEvent1, middleEvent1a, middleEvent1b),
-				Lists.newArrayList(startEvent, middleEvent1a, middleEvent1b, end)
-		));
-	}
-
-	@Test
-	public void testEndWLoopingWSameElement() throws Exception {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		Event startEvent = new Event(40, "c", 1.0);
-		Event middleEvent1 = new Event(41, "a", 2.0);
-		Event middleEvent1a = new Event(41, "a", 2.0);
-		Event middleEvent1b = new Event(41, "a", 2.0);
-		final Event end = new Event(44, "b", 5.0);
-
-		inputEvents.add(new StreamRecord<>(startEvent, 1));
-		inputEvents.add(new StreamRecord<>(middleEvent1, 3));
-		inputEvents.add(new StreamRecord<>(middleEvent1a, 3));
-		inputEvents.add(new StreamRecord<>(middleEvent1b, 3));
-		inputEvents.add(new StreamRecord<>(end, 7));
-
-		Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5726188262756267490L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("c");
-			}
-		}).followedByAny("middle").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5726188262756267490L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("a");
-			}
-		}).oneOrMore().optional();
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		final List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
-
-		compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
-				Lists.newArrayList(startEvent),
-				Lists.newArrayList(startEvent, middleEvent1),
-				Lists.newArrayList(startEvent, middleEvent1a),
-				Lists.newArrayList(startEvent, middleEvent1b),
-				Lists.newArrayList(startEvent, middleEvent1, middleEvent1a),
-				Lists.newArrayList(startEvent, middleEvent1a, middleEvent1b),
-				Lists.newArrayList(startEvent, middleEvent1, middleEvent1a, middleEvent1b)
-		));
-	}
-
-	@Test
-	public void testRepeatingPatternWSameElement() throws Exception {
-		List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
-		Event startEvent = new Event(40, "c", 1.0);
-		Event middle1Event1 = new Event(40, "a", 2.0);
-		Event middle1Event2 = new Event(40, "a", 3.0);
-		Event middle1Event3 = new Event(40, "a", 4.0);
-		Event middle2Event1 = new Event(40, "b", 5.0);
-
-		inputEvents.add(new StreamRecord<>(startEvent, 1));
-		inputEvents.add(new StreamRecord<>(middle1Event1, 3));
-		inputEvents.add(new StreamRecord<>(middle1Event1, 3));
-		inputEvents.add(new StreamRecord<>(middle1Event2, 3));
-		inputEvents.add(new StreamRecord<>(new Event(40, "d", 6.0), 5));
-		inputEvents.add(new StreamRecord<>(middle2Event1, 6));
-		inputEvents.add(new StreamRecord<>(middle1Event3, 7));
-
-		Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5726188262756267490L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("c");
-			}
-		}).followedBy("middle1").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5726188262756267490L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("a");
-			}
-		}).oneOrMore().optional().followedBy("middle2").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5726188262756267490L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("b");
-			}
-		}).optional().followedBy("end").where(new SimpleCondition<Event>() {
-			private static final long serialVersionUID = 5726188262756267490L;
-
-			@Override
-			public boolean filter(Event value) throws Exception {
-				return value.getName().equals("a");
-			}
-		});
-
-		NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
-
-		final List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
-
-		compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
-				Lists.newArrayList(startEvent, middle1Event1),
-
-				Lists.newArrayList(startEvent, middle1Event1, middle1Event1),
-				Lists.newArrayList(startEvent, middle2Event1, middle1Event3),
-
-				Lists.newArrayList(startEvent, middle1Event1, middle1Event1, middle1Event2),
-				Lists.newArrayList(startEvent, middle1Event1, middle2Event1, middle1Event3),
-
-				Lists.newArrayList(startEvent, middle1Event1, middle1Event1, middle1Event2, middle1Event3),
-				Lists.newArrayList(startEvent, middle1Event1, middle1Event1, middle2Event1, middle1Event3),
-
-				Lists.newArrayList(startEvent, middle1Event1, middle1Event1, middle1Event2, middle2Event1, middle1Event3)
-		));
-	}
-
-	/////////////////////////////////////////       Utility        /////////////////////////////////////////////////
-
-	private List<List<Event>> feedNFA(List<StreamRecord<Event>> inputEvents, NFA<Event> nfa) {
-		List<List<Event>> resultingPatterns = new ArrayList<>();
-
-		for (StreamRecord<Event> inputEvent : inputEvents) {
-			Collection<Map<String, List<Event>>> patterns = nfa.process(
-				inputEvent.getValue(),
-				inputEvent.getTimestamp()).f0;
-
-			for (Map<String, List<Event>> p: patterns) {
-				List<Event> res = new ArrayList<>();
-				for (List<Event> le: p.values()) {
-					res.addAll(le);
-				}
-				resultingPatterns.add(res);
-			}
-		}
-		return resultingPatterns;
-	}
-
-	private void compareMaps(List<List<Event>> actual, List<List<Event>> expected) {
-		Assert.assertEquals(expected.size(), actual.size());
-
-		for (List<Event> p: actual) {
-			Collections.sort(p, new EventComparator());
-		}
-
-		for (List<Event> p: expected) {
-			Collections.sort(p, new EventComparator());
-		}
-
-		Collections.sort(actual, new ListEventComparator());
-		Collections.sort(expected, new ListEventComparator());
-		Assert.assertArrayEquals(expected.toArray(), actual.toArray());
-	}
-
-	private class ListEventComparator implements Comparator<List<Event>> {
-
-		@Override
-		public int compare(List<Event> o1, List<Event> o2) {
-			int sizeComp = Integer.compare(o1.size(), o2.size());
-			if (sizeComp == 0) {
-				EventComparator comp = new EventComparator();
-				for (int i = 0; i < o1.size(); i++) {
-					int eventComp = comp.compare(o1.get(i), o2.get(i));
-					if (eventComp != 0) {
-						return eventComp;
-					}
-				}
-				return 0;
-			} else {
-				return sizeComp;
-			}
-		}
-	}
-
-	private class EventComparator implements Comparator<Event> {
-
-		@Override
-		public int compare(Event o1, Event o2) {
-			int nameComp = o1.getName().compareTo(o2.getName());
-			int priceComp = Doubles.compare(o1.getPrice(), o2.getPrice());
-			int idComp = Integer.compare(o1.getId(), o2.getId());
-			if (nameComp == 0) {
-				if (priceComp == 0) {
-					return idComp;
-				} else {
-					return priceComp;
-				}
-			} else {
-				return nameComp;
-			}
-		}
-	}
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/c9e574bf/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFATest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFATest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFATest.java
index 2619764..2586342 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFATest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFATest.java
@@ -18,7 +18,6 @@
 
 package org.apache.flink.cep.nfa;
 
-import org.apache.commons.io.output.ByteArrayOutputStream;
 import org.apache.flink.cep.Event;
 import org.apache.flink.cep.nfa.compiler.NFACompiler;
 import org.apache.flink.cep.pattern.Pattern;
@@ -29,6 +28,8 @@ import org.apache.flink.core.memory.DataInputViewStreamWrapper;
 import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 import org.apache.flink.util.TestLogger;
+
+import org.apache.commons.io.output.ByteArrayOutputStream;
 import org.junit.Test;
 
 import java.io.ByteArrayInputStream;
@@ -44,6 +45,9 @@ import java.util.Set;
 
 import static org.junit.Assert.assertEquals;
 
+/**
+ * Tests for {@link NFA}.
+ */
 public class NFATest extends TestLogger {
 	@Test
 	public void testSimpleNFA() {
@@ -147,7 +151,7 @@ public class NFATest extends TestLogger {
 
 	/**
 	 * Tests that pruning shared buffer elements and computations state use the same window border
-	 * semantics (left side inclusive and right side exclusive)
+	 * semantics (left side inclusive and right side exclusive).
 	 */
 	@Test
 	public void testTimeoutWindowPruningWindowBorders() {


Mime
View raw message