You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2016/07/18 10:39:25 UTC
[1/7] flink git commit: Replace StreamEvent by StreamRecord in CEP
Tests
Repository: flink
Updated Branches:
refs/heads/master e9f660d1f -> a710fc3d5
Replace StreamEvent by StreamRecord in CEP Tests
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a4c9d665
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a4c9d665
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a4c9d665
Branch: refs/heads/master
Commit: a4c9d66599378d4ec0be466f806d1603f8960438
Parents: e9f660d
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Tue Jul 5 17:56:49 2016 +0200
Committer: Aljoscha Krettek <al...@gmail.com>
Committed: Mon Jul 18 11:15:18 2016 +0200
----------------------------------------------------------------------
.../java/org/apache/flink/cep/StreamEvent.java | 41 --------------
.../org/apache/flink/cep/nfa/NFAITCase.java | 57 ++++++++++----------
.../java/org/apache/flink/cep/nfa/NFATest.java | 44 +++++++--------
3 files changed, 51 insertions(+), 91 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/a4c9d665/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/StreamEvent.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/StreamEvent.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/StreamEvent.java
deleted file mode 100644
index adb9446..0000000
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/StreamEvent.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.cep;
-
-public class StreamEvent<T> {
- private final T event;
- private final long timestamp;
-
- public StreamEvent(T event, long timestamp) {
- this.event = event;
- this.timestamp = timestamp;
- }
-
- public long getTimestamp() {
- return timestamp;
- }
-
- public T getEvent() {
- return event;
- }
-
- public static <V> StreamEvent<V> of(V event, long timestamp) {
- return new StreamEvent<>(event, timestamp);
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/a4c9d665/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 6152191..4a33c1e 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
@@ -21,11 +21,11 @@ package org.apache.flink.cep.nfa;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.cep.Event;
-import org.apache.flink.cep.StreamEvent;
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.streaming.api.windowing.time.Time;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.util.TestLogger;
import org.junit.Test;
@@ -43,18 +43,18 @@ public class NFAITCase extends TestLogger {
@Test
public void testSimplePatternNFA() {
- List<StreamEvent<Event>> inputEvents = new ArrayList<StreamEvent<Event>>();
+ List<StreamRecord<Event>> inputEvents = new ArrayList<>();
Event startEvent = new Event(42, "start", 1.0);
SubEvent middleEvent = new SubEvent(42, "foo", 1.0, 10.0);
Event endEvent= new Event(43, "end", 1.0);
- inputEvents.add(new StreamEvent<Event>(startEvent, 1));
- inputEvents.add(new StreamEvent<Event>(new Event(43, "foobar", 1.0), 2));
- inputEvents.add(new StreamEvent<Event>(new SubEvent(41, "barfoo", 1.0, 5.0), 3));
- inputEvents.add(new StreamEvent<Event>(middleEvent, 3));
- inputEvents.add(new StreamEvent<Event>(new Event(43, "start", 1.0), 4));
- inputEvents.add(new StreamEvent<Event>(endEvent, 5));
+ inputEvents.add(new StreamRecord<Event>(startEvent, 1));
+ inputEvents.add(new StreamRecord<Event>(new Event(43, "foobar", 1.0), 2));
+ inputEvents.add(new StreamRecord<Event>(new SubEvent(41, "barfoo", 1.0, 5.0), 3));
+ inputEvents.add(new StreamRecord<Event>(middleEvent, 3));
+ inputEvents.add(new StreamRecord<Event>(new Event(43, "start", 1.0), 4));
+ inputEvents.add(new StreamRecord<Event>(endEvent, 5));
Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new FilterFunction<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@@ -82,11 +82,12 @@ public class NFAITCase extends TestLogger {
});
NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
+
List<Map<String, Event>> resultingPatterns = new ArrayList<>();
- for (StreamEvent<Event> inputEvent: inputEvents) {
+ for (StreamRecord<Event> inputEvent: inputEvents) {
Collection<Map<String, Event>> patterns = nfa.process(
- inputEvent.getEvent(),
+ inputEvent.getValue(),
inputEvent.getTimestamp()).f0;
resultingPatterns.addAll(patterns);
@@ -106,19 +107,19 @@ public class NFAITCase extends TestLogger {
*/
@Test
public void testSimplePatternWithTimeWindowNFA() {
- List<StreamEvent<Event>> events = new ArrayList<>();
+ List<StreamRecord<Event>> events = new ArrayList<>();
List<Map<String, Event>> resultingPatterns = new ArrayList<>();
final Event startEvent;
final Event middleEvent;
final Event endEvent;
- events.add(new StreamEvent<Event>(new Event(1, "start", 1.0), 1));
- events.add(new StreamEvent<Event>(startEvent = new Event(2, "start", 1.0), 2));
- events.add(new StreamEvent<Event>(middleEvent = new Event(3, "middle", 1.0), 3));
- events.add(new StreamEvent<Event>(new Event(4, "foobar", 1.0), 4));
- events.add(new StreamEvent<Event>(endEvent = new Event(5, "end", 1.0), 11));
- events.add(new StreamEvent<Event>(new Event(6, "end", 1.0), 13));
+ events.add(new StreamRecord<Event>(new Event(1, "start", 1.0), 1));
+ events.add(new StreamRecord<Event>(startEvent = new Event(2, "start", 1.0), 2));
+ events.add(new StreamRecord<Event>(middleEvent = new Event(3, "middle", 1.0), 3));
+ events.add(new StreamRecord<Event>(new Event(4, "foobar", 1.0), 4));
+ events.add(new StreamRecord<Event>(endEvent = new Event(5, "end", 1.0), 11));
+ events.add(new StreamRecord<Event>(new Event(6, "end", 1.0), 13));
Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new FilterFunction<Event>() {
@@ -147,8 +148,8 @@ public class NFAITCase extends TestLogger {
NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
- for (StreamEvent<Event> event: events) {
- Collection<Map<String, Event>> patterns = nfa.process(event.getEvent(), event.getTimestamp()).f0;
+ for (StreamRecord<Event> event: events) {
+ Collection<Map<String, Event>> patterns = nfa.process(event.getValue(), event.getTimestamp()).f0;
resultingPatterns.addAll(patterns);
}
@@ -168,18 +169,18 @@ public class NFAITCase extends TestLogger {
*/
@Test
public void testSimplePatternWithTimeoutHandling() {
- List<StreamEvent<Event>> events = new ArrayList<>();
+ List<StreamRecord<Event>> events = new ArrayList<>();
List<Map<String, Event>> resultingPatterns = new ArrayList<>();
Set<Tuple2<Map<String, Event>, Long>> resultingTimeoutPatterns = new HashSet<>();
Set<Tuple2<Map<String, Event>, Long>> expectedTimeoutPatterns = new HashSet<>();
- events.add(new StreamEvent<Event>(new Event(1, "start", 1.0), 1));
- events.add(new StreamEvent<Event>(new Event(2, "start", 1.0), 2));
- events.add(new StreamEvent<Event>(new Event(3, "middle", 1.0), 3));
- events.add(new StreamEvent<Event>(new Event(4, "foobar", 1.0), 4));
- events.add(new StreamEvent<Event>(new Event(5, "end", 1.0), 11));
- events.add(new StreamEvent<Event>(new Event(6, "end", 1.0), 13));
+ events.add(new StreamRecord<Event>(new Event(1, "start", 1.0), 1));
+ events.add(new StreamRecord<Event>(new Event(2, "start", 1.0), 2));
+ events.add(new StreamRecord<Event>(new Event(3, "middle", 1.0), 3));
+ events.add(new StreamRecord<Event>(new Event(4, "foobar", 1.0), 4));
+ events.add(new StreamRecord<Event>(new Event(5, "end", 1.0), 11));
+ events.add(new StreamRecord<Event>(new Event(6, "end", 1.0), 13));
Map<String, Event> timeoutPattern1 = new HashMap<>();
timeoutPattern1.put("start", new Event(1, "start", 1.0));
@@ -226,9 +227,9 @@ public class NFAITCase extends TestLogger {
NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), true);
- for (StreamEvent<Event> event: events) {
+ for (StreamRecord<Event> event: events) {
Tuple2<Collection<Map<String, Event>>, Collection<Tuple2<Map<String, Event>, Long>>> patterns =
- nfa.process(event.getEvent(), event.getTimestamp());
+ nfa.process(event.getValue(), event.getTimestamp());
Collection<Map<String, Event>> matchedPatterns = patterns.f0;
Collection<Tuple2<Map<String, Event>, Long>> timeoutPatterns = patterns.f1;
http://git-wip-us.apache.org/repos/asf/flink/blob/a4c9d665/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 f48dab3..9f65132 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
@@ -21,7 +21,7 @@ package org.apache.flink.cep.nfa;
import org.apache.commons.io.output.ByteArrayOutputStream;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.cep.Event;
-import org.apache.flink.cep.StreamEvent;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.util.TestLogger;
import org.junit.Test;
@@ -44,12 +44,12 @@ public class NFATest extends TestLogger {
@Test
public void testSimpleNFA() {
NFA<Event> nfa = new NFA<>(Event.createTypeSerializer(), 0, false);
- List<StreamEvent<Event>> streamEvents = new ArrayList<>();
+ List<StreamRecord<Event>> streamEvents = new ArrayList<>();
- streamEvents.add(StreamEvent.of(new Event(1, "start", 1.0), 1L));
- streamEvents.add(StreamEvent.of(new Event(2, "bar", 2.0), 2L));
- streamEvents.add(StreamEvent.of(new Event(3, "start", 3.0), 3L));
- streamEvents.add(StreamEvent.of(new Event(4, "end", 4.0), 4L));
+ streamEvents.add(new StreamRecord<>(new Event(1, "start", 1.0), 1L));
+ streamEvents.add(new StreamRecord<>(new Event(2, "bar", 2.0), 2L));
+ streamEvents.add(new StreamRecord<>(new Event(3, "start", 3.0), 3L));
+ streamEvents.add(new StreamRecord<>(new Event(4, "end", 4.0), 4L));
State<Event> startingState = new State<>("", State.StateType.Start);
State<Event> startState = new State<>("start", State.StateType.Normal);
@@ -111,12 +111,12 @@ public class NFATest extends TestLogger {
@Test
public void testTimeoutWindowPruning() {
NFA<Event> nfa = createStartEndNFA(2);
- List<StreamEvent<Event>> streamEvents = new ArrayList<>();
+ List<StreamRecord<Event>> streamEvents = new ArrayList<>();
- streamEvents.add(StreamEvent.of(new Event(1, "start", 1.0), 1L));
- streamEvents.add(StreamEvent.of(new Event(2, "bar", 2.0), 2L));
- streamEvents.add(StreamEvent.of(new Event(3, "start", 3.0), 3L));
- streamEvents.add(StreamEvent.of(new Event(4, "end", 4.0), 4L));
+ streamEvents.add(new StreamRecord<>(new Event(1, "start", 1.0), 1L));
+ streamEvents.add(new StreamRecord<>(new Event(2, "bar", 2.0), 2L));
+ streamEvents.add(new StreamRecord<>(new Event(3, "start", 3.0), 3L));
+ streamEvents.add(new StreamRecord<>(new Event(4, "end", 4.0), 4L));
Set<Map<String, Event>> expectedPatterns = new HashSet<>();
@@ -138,10 +138,10 @@ public class NFATest extends TestLogger {
@Test
public void testWindowBorders() {
NFA<Event> nfa = createStartEndNFA(2);
- List<StreamEvent<Event>> streamEvents = new ArrayList<>();
+ List<StreamRecord<Event>> streamEvents = new ArrayList<>();
- streamEvents.add(StreamEvent.of(new Event(1, "start", 1.0), 1L));
- streamEvents.add(StreamEvent.of(new Event(2, "end", 2.0), 3L));
+ streamEvents.add(new StreamRecord<>(new Event(1, "start", 1.0), 1L));
+ streamEvents.add(new StreamRecord<>(new Event(2, "end", 2.0), 3L));
Set<Map<String, Event>> expectedPatterns = Collections.emptySet();
@@ -157,12 +157,12 @@ public class NFATest extends TestLogger {
@Test
public void testTimeoutWindowPruningWindowBorders() {
NFA<Event> nfa = createStartEndNFA(2);
- List<StreamEvent<Event>> streamEvents = new ArrayList<>();
+ List<StreamRecord<Event>> streamEvents = new ArrayList<>();
- streamEvents.add(StreamEvent.of(new Event(1, "start", 1.0), 1L));
- streamEvents.add(StreamEvent.of(new Event(2, "start", 2.0), 2L));
- streamEvents.add(StreamEvent.of(new Event(3, "foobar", 3.0), 3L));
- streamEvents.add(StreamEvent.of(new Event(4, "end", 4.0), 3L));
+ streamEvents.add(new StreamRecord<>(new Event(1, "start", 1.0), 1L));
+ streamEvents.add(new StreamRecord<>(new Event(2, "start", 2.0), 2L));
+ streamEvents.add(new StreamRecord<>(new Event(3, "foobar", 3.0), 3L));
+ streamEvents.add(new StreamRecord<>(new Event(4, "end", 4.0), 3L));
Set<Map<String, Event>> expectedPatterns = new HashSet<>();
@@ -193,11 +193,11 @@ public class NFATest extends TestLogger {
assertEquals(expectedName3, generatedName3);
}
- public <T> Collection<Map<String, T>> runNFA(NFA<T> nfa, List<StreamEvent<T>> inputs) {
+ public <T> Collection<Map<String, T>> runNFA(NFA<T> nfa, List<StreamRecord<T>> inputs) {
Set<Map<String, T>> actualPatterns = new HashSet<>();
- for (StreamEvent<T> streamEvent: inputs) {
- Collection<Map<String, T>> matchedPatterns = nfa.process(streamEvent.getEvent(), streamEvent.getTimestamp()).f0;
+ for (StreamRecord<T> streamEvent: inputs) {
+ Collection<Map<String, T>> matchedPatterns = nfa.process(streamEvent.getValue(), streamEvent.getTimestamp()).f0;
actualPatterns.addAll(matchedPatterns);
}
[3/7] flink git commit: [FLINK-4149] Fix Serialization of NFA in
AbstractKeyedCEPPatternOperator
Posted by al...@apache.org.
[FLINK-4149] Fix Serialization of NFA in AbstractKeyedCEPPatternOperator
NFA is Serializable and has readObject()/writeObject() methods. In
AbstractKeyedCEPPatternOperator a KryoSerializer was used as the
TypeSerializer for the ValueState that holds NFA instances. Kryo does
not call readObject()/writeObject() therefore the state of the NFA was
invalid after deserialization.
This change adds a new TypeSerializer for NFA that uses
Java Serialization. In the long run it will be better to get rid of the
readObject()/writeObject() methods and instead efficiently serialize
using a specialized TypeSerializer.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ac061467
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ac061467
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ac061467
Branch: refs/heads/master
Commit: ac06146719e1061aecdc7f18c95913ac9009a711
Parents: a4c9d66
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Tue Jul 5 17:58:45 2016 +0200
Committer: Aljoscha Krettek <al...@gmail.com>
Committed: Mon Jul 18 11:16:03 2016 +0200
----------------------------------------------------------------------
.../main/java/org/apache/flink/cep/nfa/NFA.java | 110 +++++++++++++++
.../AbstractKeyedCEPPatternOperator.java | 3 +-
.../flink/cep/operator/CEPOperatorTest.java | 140 ++++++++++++++++---
3 files changed, 231 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ac061467/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java
----------------------------------------------------------------------
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 f769a2b..624db0d 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
@@ -21,10 +21,16 @@ package org.apache.flink.cep.nfa;
import com.google.common.collect.LinkedHashMultimap;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.runtime.DataInputViewStream;
+import org.apache.flink.api.java.typeutils.runtime.DataOutputViewStream;
import org.apache.flink.cep.NonDuplicatingTypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
@@ -454,4 +460,108 @@ public class NFA<T> implements Serializable {
return name + "_" + index;
}
}
+
+ /**
+ * {@link TypeSerializer} for {@link NFA} that uses Java Serialization.
+ */
+ public static class Serializer<T> extends TypeSerializer<NFA<T>> {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public boolean isImmutableType() {
+ return false;
+ }
+
+ @Override
+ public TypeSerializer<NFA<T>> duplicate() {
+ return this;
+ }
+
+ @Override
+ public NFA<T> createInstance() {
+ return null;
+ }
+
+ @Override
+ public NFA<T> copy(NFA<T> from) {
+ try {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ ObjectOutputStream oos = new ObjectOutputStream(baos);
+
+ oos.writeObject(from);
+
+ oos.close();
+ baos.close();
+
+ byte[] data = baos.toByteArray();
+
+ ByteArrayInputStream bais = new ByteArrayInputStream(data);
+ ObjectInputStream ois = new ObjectInputStream(bais);
+
+ @SuppressWarnings("unchecked")
+ NFA<T> copy = (NFA<T>) ois.readObject();
+ return copy;
+ } catch (IOException|ClassNotFoundException e) {
+ throw new RuntimeException("Could not copy NFA.", e);
+ }
+ }
+
+ @Override
+ public NFA<T> copy(NFA<T> from, NFA<T> reuse) {
+ return copy(from);
+ }
+
+ @Override
+ public int getLength() {
+ return 0;
+ }
+
+ @Override
+ public void serialize(NFA<T> record, DataOutputView target) throws IOException {
+ ObjectOutputStream oos = new ObjectOutputStream(new DataOutputViewStream(target));
+ oos.writeObject(record);
+ oos.close();
+ }
+
+ @Override
+ public NFA<T> deserialize(DataInputView source) throws IOException {
+ ObjectInputStream ois = new ObjectInputStream(new DataInputViewStream(source));
+
+ try {
+ @SuppressWarnings("unchecked")
+ NFA<T> nfa = null;
+ nfa = (NFA<T>) ois.readObject();
+ return nfa;
+ } catch (ClassNotFoundException e) {
+ throw new RuntimeException("Could not deserialize NFA.", e);
+ }
+ }
+
+ @Override
+ public NFA<T> deserialize(NFA<T> reuse, DataInputView source) throws IOException {
+ return deserialize(source);
+ }
+
+ @Override
+ public void copy(DataInputView source, DataOutputView target) throws IOException {
+ int size = source.readInt();
+ target.writeInt(size);
+ target.write(source, size);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return obj instanceof Serializer && ((Serializer) obj).canEqual(this);
+ }
+
+ @Override
+ public boolean canEqual(Object obj) {
+ return obj instanceof Serializer;
+ }
+
+ @Override
+ public int hashCode() {
+ return getClass().hashCode();
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ac061467/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
index 9ffe9b6..c384bb8 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
@@ -22,7 +22,6 @@ import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.functions.KeySelector;
-import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
import org.apache.flink.cep.nfa.NFA;
import org.apache.flink.cep.nfa.compiler.NFACompiler;
import org.apache.flink.core.memory.DataInputView;
@@ -100,7 +99,7 @@ abstract public class AbstractKeyedCEPPatternOperator<IN, KEY, OUT> extends Abst
nfaOperatorState = getPartitionedState(
new ValueStateDescriptor<NFA<IN>>(
NFA_OPERATOR_STATE_NAME,
- new KryoSerializer<NFA<IN>>((Class<NFA<IN>>) (Class<?>) NFA.class, getExecutionConfig()),
+ new NFA.Serializer<IN>(),
null));
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ac061467/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
index d5ef5be..0105715 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
@@ -18,29 +18,35 @@
package org.apache.flink.cep.operator;
+import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.cep.Event;
+import org.apache.flink.cep.SubEvent;
import org.apache.flink.cep.nfa.NFA;
import org.apache.flink.cep.nfa.compiler.NFACompiler;
+import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskState;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.util.TestLogger;
import org.junit.Test;
import static org.junit.Assert.*;
import java.util.Map;
+import java.util.concurrent.ConcurrentLinkedQueue;
public class CEPOperatorTest extends TestLogger {
@Test
public void testCEPOperatorWatermarkForwarding() throws Exception {
- OneInputStreamOperatorTestHarness<Integer, Map<String, Integer>> harness = new OneInputStreamOperatorTestHarness<>(
- new CEPPatternOperator<Integer>(
- IntSerializer.INSTANCE,
+ OneInputStreamOperatorTestHarness<Event, Map<String, Event>> harness = new OneInputStreamOperatorTestHarness<>(
+ new CEPPatternOperator<>(
+ Event.createTypeSerializer(),
false,
- new DummyNFAFactory<>(IntSerializer.INSTANCE))
+ new NFAFactory())
);
harness.open();
@@ -59,22 +65,22 @@ public class CEPOperatorTest extends TestLogger {
@Test
public void testKeyedCEPOperatorWatermarkForwarding() throws Exception {
- KeySelector<Integer, Integer> keySelector = new KeySelector<Integer, Integer>() {
+ KeySelector<Event, Integer> keySelector = new KeySelector<Event, Integer>() {
private static final long serialVersionUID = -4873366487571254798L;
@Override
- public Integer getKey(Integer value) throws Exception {
- return value;
+ public Integer getKey(Event value) throws Exception {
+ return value.getId();
}
};
- OneInputStreamOperatorTestHarness<Integer, Map<String, Integer>> harness = new OneInputStreamOperatorTestHarness<>(
- new KeyedCEPPatternOperator<Integer, Integer>(
- IntSerializer.INSTANCE,
+ OneInputStreamOperatorTestHarness<Event, Map<String, Event>> harness = new OneInputStreamOperatorTestHarness<>(
+ new KeyedCEPPatternOperator<>(
+ Event.createTypeSerializer(),
false,
keySelector,
IntSerializer.INSTANCE,
- new DummyNFAFactory<>(IntSerializer.INSTANCE))
+ new NFAFactory())
);
harness.configureForKeyedStream(keySelector, BasicTypeInfo.INT_TYPE_INFO);
@@ -93,19 +99,113 @@ public class CEPOperatorTest extends TestLogger {
harness.close();
}
- public static class DummyNFAFactory<T> implements NFACompiler.NFAFactory<T> {
+ @Test
+ public void testKeyedCEPOperatorCheckpointing() throws Exception {
+ KeySelector<Event, Integer> keySelector = new KeySelector<Event, Integer>() {
+ private static final long serialVersionUID = -4873366487571254798L;
- private static final long serialVersionUID = 1173020762472766713L;
+ @Override
+ public Integer getKey(Event value) throws Exception {
+ return value.getId();
+ }
+ };
- private final TypeSerializer<T> inputTypeSerializer;
+ OneInputStreamOperatorTestHarness<Event, Map<String, Event>> harness = new OneInputStreamOperatorTestHarness<>(
+ new KeyedCEPPatternOperator<>(
+ Event.createTypeSerializer(),
+ false,
+ keySelector,
+ IntSerializer.INSTANCE,
+ new NFAFactory()));
- public DummyNFAFactory(TypeSerializer<T> inputTypeSerializer) {
- this.inputTypeSerializer = inputTypeSerializer;
- }
+ harness.configureForKeyedStream(keySelector, BasicTypeInfo.INT_TYPE_INFO);
+
+ harness.open();
+
+ Event startEvent = new Event(42, "start", 1.0);
+ SubEvent middleEvent = new SubEvent(42, "foo", 1.0, 10.0);
+ Event endEvent= new Event(42, "end", 1.0);
+
+ harness.processElement(new StreamRecord<Event>(startEvent, 1));
+ harness.processElement(new StreamRecord<Event>(new Event(42, "foobar", 1.0), 2));
+ harness.processElement(new StreamRecord<Event>(new SubEvent(42, "barfoo", 1.0, 5.0), 3));
+
+ harness.processWatermark(new Watermark(2));
+
+ // simulate snapshot/restore
+ StreamTaskState snapshot = harness.snapshot(0, 0);
+
+ harness = new OneInputStreamOperatorTestHarness<>(
+ new KeyedCEPPatternOperator<>(
+ Event.createTypeSerializer(),
+ false,
+ keySelector,
+ IntSerializer.INSTANCE,
+ new NFAFactory()));
+
+ harness.configureForKeyedStream(keySelector, BasicTypeInfo.INT_TYPE_INFO);
+ harness.setup();
+ harness.restore(snapshot, 1);
+ harness.open();
+
+ harness.processElement(new StreamRecord<Event>(middleEvent, 3));
+ harness.processElement(new StreamRecord<Event>(new Event(42, "start", 1.0), 4));
+ harness.processElement(new StreamRecord<Event>(endEvent, 5));
+
+ harness.processWatermark(new Watermark(Long.MAX_VALUE));
+
+ ConcurrentLinkedQueue<Object> result = harness.getOutput();
+
+ assertEquals(2, result.size());
+
+ Object resultObject = result.poll();
+ assertTrue(resultObject instanceof StreamRecord);
+ StreamRecord<?> resultRecord = (StreamRecord<?>) resultObject;
+ assertTrue(resultRecord.getValue() instanceof Map);
+
+ @SuppressWarnings("unchecked")
+ Map<String, Event> patternMap = (Map<String, Event>) resultRecord.getValue();
+
+ assertEquals(startEvent, patternMap.get("start"));
+ assertEquals(middleEvent, patternMap.get("middle"));
+ assertEquals(endEvent, patternMap.get("end"));
+
+ harness.close();
+ }
+
+ private static class NFAFactory implements NFACompiler.NFAFactory<Event> {
+
+ private static final long serialVersionUID = 1173020762472766713L;
@Override
- public NFA<T> createNFA() {
- return new NFA<>(inputTypeSerializer.duplicate(), 0, false);
+ public NFA<Event> createNFA() {
+
+ Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new FilterFunction<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 FilterFunction<SubEvent>() {
+ private static final long serialVersionUID = 6215754202506583964L;
+
+ @Override
+ public boolean filter(SubEvent value) throws Exception {
+ return value.getVolume() > 5.0;
+ }
+ })
+ .followedBy("end").where(new FilterFunction<Event>() {
+ private static final long serialVersionUID = 7056763917392056548L;
+
+ @Override
+ public boolean filter(Event value) throws Exception {
+ return value.getName().equals("end");
+ }
+ });
+
+ return NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
}
}
}
[2/7] flink git commit: Allow Setting StateBackend in
OneInputStreamOperatorTestHarness
Posted by al...@apache.org.
Allow Setting StateBackend in OneInputStreamOperatorTestHarness
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/4e6d89ca
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/4e6d89ca
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/4e6d89ca
Branch: refs/heads/master
Commit: 4e6d89ca20f6888cdda911e8d90baec19245b6cc
Parents: 7329074
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Thu Jul 7 16:45:46 2016 +0200
Committer: Aljoscha Krettek <al...@gmail.com>
Committed: Mon Jul 18 11:16:03 2016 +0200
----------------------------------------------------------------------
.../util/OneInputStreamOperatorTestHarness.java | 54 +++++++++++++++++---
1 file changed, 47 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/4e6d89ca/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/OneInputStreamOperatorTestHarness.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/OneInputStreamOperatorTestHarness.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/OneInputStreamOperatorTestHarness.java
index 7d0fc57..40086c5 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/OneInputStreamOperatorTestHarness.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/OneInputStreamOperatorTestHarness.java
@@ -27,6 +27,9 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.operators.testutils.MockEnvironment;
import org.apache.flink.runtime.operators.testutils.MockInputSplitProvider;
+import org.apache.flink.runtime.state.AsynchronousKvStateSnapshot;
+import org.apache.flink.runtime.state.AsynchronousStateHandle;
+import org.apache.flink.runtime.state.KvStateSnapshot;
import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.Output;
@@ -42,7 +45,10 @@ import org.apache.flink.streaming.runtime.tasks.TimeServiceProvider;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
+import java.io.Serializable;
import java.util.Collection;
+import java.util.HashMap;
+import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;
@@ -76,6 +82,9 @@ public class OneInputStreamOperatorTestHarness<IN, OUT> {
StreamTask<?, ?> mockTask;
+ // use this as default for tests
+ private AbstractStateBackend stateBackend = new MemoryStateBackend();
+
/**
* Whether setup() was called on the operator. This is reset when calling close().
*/
@@ -86,12 +95,16 @@ public class OneInputStreamOperatorTestHarness<IN, OUT> {
this(operator, new ExecutionConfig());
}
- public OneInputStreamOperatorTestHarness(OneInputStreamOperator<IN, OUT> operator, ExecutionConfig executionConfig) {
+ public OneInputStreamOperatorTestHarness(
+ OneInputStreamOperator<IN, OUT> operator,
+ ExecutionConfig executionConfig) {
this(operator, executionConfig, DefaultTimeServiceProvider.create(Executors.newSingleThreadScheduledExecutor()));
}
- public OneInputStreamOperatorTestHarness(OneInputStreamOperator<IN, OUT> operator, ExecutionConfig executionConfig,
- TimeServiceProvider testTimeProvider) {
+ public OneInputStreamOperatorTestHarness(
+ OneInputStreamOperator<IN, OUT> operator,
+ ExecutionConfig executionConfig,
+ TimeServiceProvider testTimeProvider) {
this.operator = operator;
this.outputList = new ConcurrentLinkedQueue<Object>();
this.config = new StreamConfig(new Configuration());
@@ -114,9 +127,9 @@ public class OneInputStreamOperatorTestHarness<IN, OUT> {
public AbstractStateBackend answer(InvocationOnMock invocationOnMock) throws Throwable {
final String operatorIdentifier = (String) invocationOnMock.getArguments()[0];
final TypeSerializer<?> keySerializer = (TypeSerializer<?>) invocationOnMock.getArguments()[1];
- MemoryStateBackend backend = MemoryStateBackend.create();
- backend.initializeForJob(env, operatorIdentifier, keySerializer);
- return backend;
+ OneInputStreamOperatorTestHarness.this.stateBackend.disposeAllStateForCurrentJob();
+ OneInputStreamOperatorTestHarness.this.stateBackend.initializeForJob(env, operatorIdentifier, keySerializer);
+ return OneInputStreamOperatorTestHarness.this.stateBackend;
}
}).when(mockTask).createStateBackend(any(String.class), any(TypeSerializer.class));
} catch (Exception e) {
@@ -143,6 +156,10 @@ public class OneInputStreamOperatorTestHarness<IN, OUT> {
}).when(mockTask).getCurrentProcessingTime();
}
+ public void setStateBackend(AbstractStateBackend stateBackend) {
+ this.stateBackend = stateBackend;
+ }
+
public Object getCheckpointLock() {
return mockTask.getCheckpointLock();
}
@@ -191,7 +208,30 @@ public class OneInputStreamOperatorTestHarness<IN, OUT> {
* Calls {@link org.apache.flink.streaming.api.operators.StreamOperator#snapshotOperatorState(long, long)} ()}
*/
public StreamTaskState snapshot(long checkpointId, long timestamp) throws Exception {
- return operator.snapshotOperatorState(checkpointId, timestamp);
+ StreamTaskState snapshot = operator.snapshotOperatorState(checkpointId, timestamp);
+ // materialize asynchronous state handles
+ if (snapshot != null) {
+ if (snapshot.getFunctionState() instanceof AsynchronousStateHandle) {
+ AsynchronousStateHandle<Serializable> asyncState = (AsynchronousStateHandle<Serializable>) snapshot.getFunctionState();
+ snapshot.setFunctionState(asyncState.materialize());
+ }
+ if (snapshot.getOperatorState() instanceof AsynchronousStateHandle) {
+ AsynchronousStateHandle<?> asyncState = (AsynchronousStateHandle<?>) snapshot.getOperatorState();
+ snapshot.setOperatorState(asyncState.materialize());
+ }
+ if (snapshot.getKvStates() != null) {
+ Set<String> keys = snapshot.getKvStates().keySet();
+ HashMap<String, KvStateSnapshot<?, ?, ?, ?, ?>> kvStates = snapshot.getKvStates();
+ for (String key: keys) {
+ if (kvStates.get(key) instanceof AsynchronousKvStateSnapshot) {
+ AsynchronousKvStateSnapshot<?, ?, ?, ?, ?> asyncHandle = (AsynchronousKvStateSnapshot<?, ?, ?, ?, ?>) kvStates.get(key);
+ kvStates.put(key, asyncHandle.materialize());
+ }
+ }
+ }
+
+ }
+ return snapshot;
}
/**
[5/7] flink git commit: [FLINK-4162] Fix Event Queue Serialization in
Abstract(Keyed)CEPPatternOperator
Posted by al...@apache.org.
[FLINK-4162] Fix Event Queue Serialization in Abstract(Keyed)CEPPatternOperator
Before, these were using StreamRecordSerializer, which does not serialize
timestamps. Now it uses MultiplexingStreamRecordSerializer.
This also extends the tests in CEPOperatorTest to test that timestamps
are correctly checkpointed/restored.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/254379b7
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/254379b7
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/254379b7
Branch: refs/heads/master
Commit: 254379b7809447cd566bd1d1e9bf3b51aa3e9d99
Parents: ac06146
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Wed Jul 6 16:46:32 2016 +0200
Committer: Aljoscha Krettek <al...@gmail.com>
Committed: Mon Jul 18 11:16:03 2016 +0200
----------------------------------------------------------------------
.../operator/AbstractCEPPatternOperator.java | 10 +-
.../AbstractKeyedCEPPatternOperator.java | 14 ++-
.../flink/cep/operator/CEPOperatorTest.java | 122 ++++++++++++++++++-
3 files changed, 133 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/254379b7/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractCEPPatternOperator.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractCEPPatternOperator.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractCEPPatternOperator.java
index 8150eae..2e21f67 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractCEPPatternOperator.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractCEPPatternOperator.java
@@ -25,8 +25,9 @@ import org.apache.flink.core.memory.DataInputViewStreamWrapper;
import org.apache.flink.runtime.state.AbstractStateBackend;
import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.MultiplexingStreamRecordSerializer;
+import org.apache.flink.streaming.runtime.streamrecord.StreamElement;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.streaming.runtime.streamrecord.StreamRecordSerializer;
import org.apache.flink.streaming.runtime.tasks.StreamTaskState;
import java.io.IOException;
@@ -46,7 +47,7 @@ import java.util.PriorityQueue;
abstract public class AbstractCEPPatternOperator<IN, OUT> extends AbstractCEPBasePatternOperator<IN, OUT> {
private static final long serialVersionUID = 7487334510746595640L;
- private final StreamRecordSerializer<IN> streamRecordSerializer;
+ private final MultiplexingStreamRecordSerializer<IN> streamRecordSerializer;
// global nfa for all elements
private NFA<IN> nfa;
@@ -60,7 +61,7 @@ abstract public class AbstractCEPPatternOperator<IN, OUT> extends AbstractCEPBas
NFACompiler.NFAFactory<IN> nfaFactory) {
super(inputSerializer, isProcessingTime);
- this.streamRecordSerializer = new StreamRecordSerializer<>(inputSerializer);
+ this.streamRecordSerializer = new MultiplexingStreamRecordSerializer<>(inputSerializer);
this.nfa = nfaFactory.createNFA();
}
@@ -134,7 +135,8 @@ abstract public class AbstractCEPPatternOperator<IN, OUT> extends AbstractCEPBas
priorityQueue = new PriorityQueue<StreamRecord<IN>>(numberPriorityQueueEntries, new StreamRecordComparator<IN>());
for (int i = 0; i <numberPriorityQueueEntries; i++) {
- priorityQueue.offer(streamRecordSerializer.deserialize(div));
+ StreamElement streamElement = streamRecordSerializer.deserialize(div);
+ priorityQueue.offer(streamElement.<IN>asRecord());
}
div.close();
http://git-wip-us.apache.org/repos/asf/flink/blob/254379b7/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
index c384bb8..79ffb2b 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
@@ -29,8 +29,8 @@ import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.runtime.state.AbstractStateBackend;
import org.apache.flink.runtime.state.StateHandle;
import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.MultiplexingStreamRecordSerializer;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.streaming.runtime.streamrecord.StreamRecordSerializer;
import org.apache.flink.streaming.runtime.tasks.StreamTaskState;
import java.io.IOException;
@@ -103,13 +103,17 @@ abstract public class AbstractKeyedCEPPatternOperator<IN, KEY, OUT> extends Abst
null));
}
+ @SuppressWarnings("unchecked,rawtypes")
+ TypeSerializer<StreamRecord<IN>> streamRecordSerializer =
+ (TypeSerializer) new MultiplexingStreamRecordSerializer<>(getInputSerializer());
+
if (priorityQueueOperatorState == null) {
priorityQueueOperatorState = getPartitionedState(
- new ValueStateDescriptor<PriorityQueue<StreamRecord<IN>>>(
+ new ValueStateDescriptor<>(
PRIORIRY_QUEUE_STATE_NAME,
- new PriorityQueueSerializer<StreamRecord<IN>>(
- new StreamRecordSerializer<IN>(getInputSerializer()),
- new PriorityQueueStreamRecordFactory<IN>()),
+ new PriorityQueueSerializer<>(
+ streamRecordSerializer,
+ new PriorityQueueStreamRecordFactory<IN>()),
null));
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/254379b7/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
index 0105715..1f53d97 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
@@ -28,6 +28,7 @@ import org.apache.flink.cep.nfa.NFA;
import org.apache.flink.cep.nfa.compiler.NFACompiler;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.tasks.StreamTaskState;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
@@ -100,6 +101,94 @@ public class CEPOperatorTest extends TestLogger {
}
@Test
+ public void testCEPOperatorCheckpointing() throws Exception {
+ KeySelector<Event, Integer> keySelector = new KeySelector<Event, Integer>() {
+ private static final long serialVersionUID = -4873366487571254798L;
+
+ @Override
+ public Integer getKey(Event value) throws Exception {
+ return value.getId();
+ }
+ };
+
+ OneInputStreamOperatorTestHarness<Event, Map<String, Event>> harness = new OneInputStreamOperatorTestHarness<>(
+ new CEPPatternOperator<>(
+ Event.createTypeSerializer(),
+ false,
+ new NFAFactory()));
+
+ harness.open();
+
+ Event startEvent = new Event(42, "start", 1.0);
+ SubEvent middleEvent = new SubEvent(42, "foo", 1.0, 10.0);
+ Event endEvent= new Event(42, "end", 1.0);
+
+ harness.processElement(new StreamRecord<Event>(startEvent, 1));
+ harness.processElement(new StreamRecord<Event>(new Event(42, "foobar", 1.0), 2));
+
+ // simulate snapshot/restore with some elements in internal sorting queue
+ StreamTaskState snapshot = harness.snapshot(0, 0);
+
+ harness = new OneInputStreamOperatorTestHarness<>(
+ new CEPPatternOperator<>(
+ Event.createTypeSerializer(),
+ false,
+ new NFAFactory()));
+
+ harness.setup();
+ harness.restore(snapshot, 1);
+ harness.open();
+
+ harness.processWatermark(new Watermark(Long.MIN_VALUE));
+
+ harness.processElement(new StreamRecord<Event>(new SubEvent(42, "barfoo", 1.0, 5.0), 3));
+
+ // if element timestamps are not correctly checkpointed/restored this will lead to
+ // a pruning time underflow exception in NFA
+ harness.processWatermark(new Watermark(2));
+
+ // simulate snapshot/restore with empty element queue but NFA state
+ StreamTaskState snapshot2 = harness.snapshot(1, 1);
+
+ harness = new OneInputStreamOperatorTestHarness<>(
+ new CEPPatternOperator<>(
+ Event.createTypeSerializer(),
+ false,
+ new NFAFactory()));
+
+ harness.setup();
+ harness.restore(snapshot2, 2);
+ harness.open();
+
+ harness.processElement(new StreamRecord<Event>(middleEvent, 3));
+ harness.processElement(new StreamRecord<Event>(new Event(42, "start", 1.0), 4));
+ harness.processElement(new StreamRecord<Event>(endEvent, 5));
+
+ harness.processWatermark(new Watermark(Long.MAX_VALUE));
+
+ ConcurrentLinkedQueue<Object> result = harness.getOutput();
+
+ // watermark and the result
+ assertEquals(2, result.size());
+
+ Object resultObject = result.poll();
+ assertTrue(resultObject instanceof StreamRecord);
+ StreamRecord<?> resultRecord = (StreamRecord<?>) resultObject;
+ assertTrue(resultRecord.getValue() instanceof Map);
+
+ @SuppressWarnings("unchecked")
+ Map<String, Event> patternMap = (Map<String, Event>) resultRecord.getValue();
+
+ assertEquals(startEvent, patternMap.get("start"));
+ assertEquals(middleEvent, patternMap.get("middle"));
+ assertEquals(endEvent, patternMap.get("end"));
+
+ harness.close();
+ }
+
+
+
+ @Test
public void testKeyedCEPOperatorCheckpointing() throws Exception {
KeySelector<Event, Integer> keySelector = new KeySelector<Event, Integer>() {
private static final long serialVersionUID = -4873366487571254798L;
@@ -128,12 +217,33 @@ public class CEPOperatorTest extends TestLogger {
harness.processElement(new StreamRecord<Event>(startEvent, 1));
harness.processElement(new StreamRecord<Event>(new Event(42, "foobar", 1.0), 2));
+
+ // simulate snapshot/restore with some elements in internal sorting queue
+ StreamTaskState snapshot = harness.snapshot(0, 0);
+
+ harness = new OneInputStreamOperatorTestHarness<>(
+ new KeyedCEPPatternOperator<>(
+ Event.createTypeSerializer(),
+ false,
+ keySelector,
+ IntSerializer.INSTANCE,
+ new NFAFactory()));
+
+ harness.configureForKeyedStream(keySelector, BasicTypeInfo.INT_TYPE_INFO);
+ harness.setup();
+ harness.restore(snapshot, 1);
+ harness.open();
+
+ harness.processWatermark(new Watermark(Long.MIN_VALUE));
+
harness.processElement(new StreamRecord<Event>(new SubEvent(42, "barfoo", 1.0, 5.0), 3));
+ // if element timestamps are not correctly checkpointed/restored this will lead to
+ // a pruning time underflow exception in NFA
harness.processWatermark(new Watermark(2));
- // simulate snapshot/restore
- StreamTaskState snapshot = harness.snapshot(0, 0);
+ // simulate snapshot/restore with empty element queue but NFA state
+ StreamTaskState snapshot2 = harness.snapshot(1, 1);
harness = new OneInputStreamOperatorTestHarness<>(
new KeyedCEPPatternOperator<>(
@@ -145,7 +255,7 @@ public class CEPOperatorTest extends TestLogger {
harness.configureForKeyedStream(keySelector, BasicTypeInfo.INT_TYPE_INFO);
harness.setup();
- harness.restore(snapshot, 1);
+ harness.restore(snapshot2, 2);
harness.open();
harness.processElement(new StreamRecord<Event>(middleEvent, 3));
@@ -156,6 +266,7 @@ public class CEPOperatorTest extends TestLogger {
ConcurrentLinkedQueue<Object> result = harness.getOutput();
+ // watermark and the result
assertEquals(2, result.size());
Object resultObject = result.poll();
@@ -203,7 +314,10 @@ public class CEPOperatorTest extends TestLogger {
public boolean filter(Event value) throws Exception {
return value.getName().equals("end");
}
- });
+ })
+ // add a window timeout to test whether timestamps of elements in the
+ // priority queue in CEP operator are correctly checkpointed/restored
+ .within(Time.milliseconds(10));
return NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
}
[4/7] flink git commit: [FLINK-4165] Add warning about
equals/hashCode to CEP doc
Posted by al...@apache.org.
[FLINK-4165] Add warning about equals/hashCode to CEP doc
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/7329074d
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/7329074d
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/7329074d
Branch: refs/heads/master
Commit: 7329074d6d4458e6a5d4830ab12fc5dbf87df1ee
Parents: 3d8d192
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Thu Jul 7 15:12:44 2016 +0200
Committer: Aljoscha Krettek <al...@gmail.com>
Committed: Mon Jul 18 11:16:03 2016 +0200
----------------------------------------------------------------------
docs/apis/streaming/libs/cep.md | 11 +++++++++--
1 file changed, 9 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/7329074d/docs/apis/streaming/libs/cep.md
----------------------------------------------------------------------
diff --git a/docs/apis/streaming/libs/cep.md b/docs/apis/streaming/libs/cep.md
index d465519..57af916 100644
--- a/docs/apis/streaming/libs/cep.md
+++ b/docs/apis/streaming/libs/cep.md
@@ -35,6 +35,13 @@ It allows you to easily detect complex event patterns in a stream of endless dat
Complex events can then be constructed from matching sequences.
This gives you the opportunity to quickly get hold of what's really important in your data.
+<span class="label label-danger">Attention</span> The events in the `DataStream` to which
+you want to apply pattern matching have to implement proper `equals()` and `hashCode()` methods
+because these are used for comparing and matching events.
+
+* This will be replaced by the TOC
+{:toc}
+
## Getting Started
If you want to jump right in, you have to [set up a Flink program]({{ site.baseurl }}/apis/batch/index.html#linking-with-flink).
@@ -518,10 +525,10 @@ The collector can be used to emit an arbitrary number of events.
val patternStream: PatternStream[Event] = CEP.pattern(input, pattern)
DataStream[Either[TimeoutEvent, ComplexEvent]] result = patternStream.flatSelect{
- (pattern: mutable.Map[String, Event], timestamp: Long, out: Collector[TimeoutEvent]) =>
+ (pattern: mutable.Map[String, Event], timestamp: Long, out: Collector[TimeoutEvent]) =>
out.collect(TimeoutEvent())
} {
- (pattern: mutable.Map[String, Event], out: Collector[ComplexEvent]) =>
+ (pattern: mutable.Map[String, Event], out: Collector[ComplexEvent]) =>
out.collect(ComplexEvent())
}
{% endhighlight %}
[6/7] flink git commit: Fix Displayed CEP Operator Names
Posted by al...@apache.org.
Fix Displayed CEP Operator Names
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/3d8d1920
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/3d8d1920
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/3d8d1920
Branch: refs/heads/master
Commit: 3d8d1920f78f7c657fd933b2e83b45e111efa18b
Parents: 254379b
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Thu Jul 7 14:48:13 2016 +0200
Committer: Aljoscha Krettek <al...@gmail.com>
Committed: Mon Jul 18 11:16:03 2016 +0200
----------------------------------------------------------------------
.../org/apache/flink/cep/operator/CEPOperatorUtils.java | 9 ++++-----
1 file changed, 4 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/3d8d1920/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CEPOperatorUtils.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CEPOperatorUtils.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CEPOperatorUtils.java
index a3fffa5..548be16 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CEPOperatorUtils.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CEPOperatorUtils.java
@@ -36,7 +36,6 @@ import org.apache.flink.types.Either;
import java.util.Map;
public class CEPOperatorUtils {
- private static final String PATTERN_OPERATOR_NAME = "AbstractCEPPatternOperator";
/**
* Creates a data stream containing the fully matching event patterns of the NFA computation.
@@ -64,7 +63,7 @@ public class CEPOperatorUtils {
TypeSerializer<K> keySerializer = keyedStream.getKeyType().createSerializer(keyedStream.getExecutionConfig());
patternStream = keyedStream.transform(
- PATTERN_OPERATOR_NAME,
+ "KeyedCEPPatternOperator",
(TypeInformation<Map<String, T>>) (TypeInformation<?>) TypeExtractor.getForClass(Map.class),
new KeyedCEPPatternOperator<>(
inputSerializer,
@@ -74,7 +73,7 @@ public class CEPOperatorUtils {
nfaFactory));
} else {
patternStream = inputStream.transform(
- PATTERN_OPERATOR_NAME,
+ "CEPPatternOperator",
(TypeInformation<Map<String, T>>) (TypeInformation<?>) TypeExtractor.getForClass(Map.class),
new CEPPatternOperator<T>(
inputSerializer,
@@ -119,7 +118,7 @@ public class CEPOperatorUtils {
TypeSerializer<K> keySerializer = keyedStream.getKeyType().createSerializer(keyedStream.getExecutionConfig());
patternStream = keyedStream.transform(
- PATTERN_OPERATOR_NAME,
+ "TimeoutKeyedCEPPatternOperator",
eitherTypeInformation,
new TimeoutKeyedCEPPatternOperator<T, K>(
inputSerializer,
@@ -129,7 +128,7 @@ public class CEPOperatorUtils {
nfaFactory));
} else {
patternStream = inputStream.transform(
- PATTERN_OPERATOR_NAME,
+ "TimeoutCEPPatternOperator",
eitherTypeInformation,
new TimeoutCEPPatternOperator<T>(
inputSerializer,
[7/7] flink git commit: [FLINK-4169] Fix State Handling in CEP
Posted by al...@apache.org.
[FLINK-4169] Fix State Handling in CEP
Before, ValueState.update() was not called after changing the NFA or the
priority queue in CEP operators. This means that the operators don't
work with state backends such as RocksDB that strictly require that
update() be called when state changes.
This changes the operators to always call update() and also introduces a
test that verifies the changes.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a710fc3d
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a710fc3d
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a710fc3d
Branch: refs/heads/master
Commit: a710fc3d58014a7191de61fd5d6370de97c9e75e
Parents: 4e6d89c
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Thu Jul 7 16:46:29 2016 +0200
Committer: Aljoscha Krettek <al...@gmail.com>
Committed: Mon Jul 18 11:16:04 2016 +0200
----------------------------------------------------------------------
flink-libraries/flink-cep/pom.xml | 8 ++
.../AbstractCEPBasePatternOperator.java | 6 +
.../operator/AbstractCEPPatternOperator.java | 10 ++
.../AbstractKeyedCEPPatternOperator.java | 23 +++-
.../flink/cep/operator/CEPOperatorTest.java | 121 ++++++++++++++++++-
5 files changed, 166 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/a710fc3d/flink-libraries/flink-cep/pom.xml
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/pom.xml b/flink-libraries/flink-cep/pom.xml
index 32b26d9..5640eb8 100644
--- a/flink-libraries/flink-cep/pom.xml
+++ b/flink-libraries/flink-cep/pom.xml
@@ -83,6 +83,14 @@ under the License.
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-statebackend-rocksdb_2.10</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+
</dependencies>
<build>
http://git-wip-us.apache.org/repos/asf/flink/blob/a710fc3d/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractCEPBasePatternOperator.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractCEPBasePatternOperator.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractCEPBasePatternOperator.java
index 44649ac..aad408c 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractCEPBasePatternOperator.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractCEPBasePatternOperator.java
@@ -58,14 +58,19 @@ public abstract class AbstractCEPBasePatternOperator<IN, OUT>
protected abstract NFA<IN> getNFA() throws IOException;
+ protected abstract void updateNFA(NFA<IN> nfa) throws IOException;
+
protected abstract PriorityQueue<StreamRecord<IN>> getPriorityQueue() throws IOException;
+ protected abstract void updatePriorityQueue(PriorityQueue<StreamRecord<IN>> queue) throws IOException;
+
@Override
public void processElement(StreamRecord<IN> element) throws Exception {
if (isProcessingTime) {
// there can be no out of order elements in processing time
NFA<IN> nfa = getNFA();
processEvent(nfa, element.getValue(), System.currentTimeMillis());
+ updateNFA(nfa);
} else {
PriorityQueue<StreamRecord<IN>> priorityQueue = getPriorityQueue();
@@ -77,6 +82,7 @@ public abstract class AbstractCEPBasePatternOperator<IN, OUT>
} else {
priorityQueue.offer(element);
}
+ updatePriorityQueue(priorityQueue);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a710fc3d/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractCEPPatternOperator.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractCEPPatternOperator.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractCEPPatternOperator.java
index 2e21f67..c26eaeb 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractCEPPatternOperator.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractCEPPatternOperator.java
@@ -78,11 +78,21 @@ abstract public class AbstractCEPPatternOperator<IN, OUT> extends AbstractCEPBas
}
@Override
+ protected void updateNFA(NFA<IN> nfa) {
+ // a no-op, because we only have one NFA
+ }
+
+ @Override
protected PriorityQueue<StreamRecord<IN>> getPriorityQueue() throws IOException {
return priorityQueue;
}
@Override
+ protected void updatePriorityQueue(PriorityQueue<StreamRecord<IN>> queue) {
+ // a no-op, because we only have one priority queue
+ }
+
+ @Override
public void processWatermark(Watermark mark) throws Exception {
while(!priorityQueue.isEmpty() && priorityQueue.peek().getTimestamp() <= mark.getTimestamp()) {
StreamRecord<IN> streamRecord = priorityQueue.poll();
http://git-wip-us.apache.org/repos/asf/flink/blob/a710fc3d/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
index 79ffb2b..206be47 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
@@ -132,6 +132,11 @@ abstract public class AbstractKeyedCEPPatternOperator<IN, KEY, OUT> extends Abst
}
@Override
+ protected void updateNFA(NFA<IN> nfa) throws IOException {
+ nfaOperatorState.update(nfa);
+ }
+
+ @Override
protected PriorityQueue<StreamRecord<IN>> getPriorityQueue() throws IOException {
PriorityQueue<StreamRecord<IN>> priorityQueue = priorityQueueOperatorState.value();
@@ -145,6 +150,11 @@ abstract public class AbstractKeyedCEPPatternOperator<IN, KEY, OUT> extends Abst
}
@Override
+ protected void updatePriorityQueue(PriorityQueue<StreamRecord<IN>> queue) throws IOException {
+ priorityQueueOperatorState.update(queue);
+ }
+
+ @Override
public void processElement(StreamRecord<IN> element) throws Exception {
keys.add(keySelector.getKey(element.getValue()));
@@ -158,7 +168,6 @@ abstract public class AbstractKeyedCEPPatternOperator<IN, KEY, OUT> extends Abst
setKeyContext(key);
PriorityQueue<StreamRecord<IN>> priorityQueue = getPriorityQueue();
-
NFA<IN> nfa = getNFA();
while (!priorityQueue.isEmpty() && priorityQueue.peek().getTimestamp() <= mark.getTimestamp()) {
@@ -166,6 +175,8 @@ abstract public class AbstractKeyedCEPPatternOperator<IN, KEY, OUT> extends Abst
processEvent(nfa, streamRecord.getValue(), streamRecord.getTimestamp());
}
+ updateNFA(nfa);
+ updatePriorityQueue(priorityQueue);
}
output.emitWatermark(mark);
@@ -336,5 +347,15 @@ abstract public class AbstractKeyedCEPPatternOperator<IN, KEY, OUT> extends Abst
public PriorityQueue<StreamRecord<T>> createPriorityQueue() {
return new PriorityQueue<StreamRecord<T>>(INITIAL_PRIORITY_QUEUE_CAPACITY, new StreamRecordComparator<T>());
}
+
+ @Override
+ public boolean equals(Object obj) {
+ return obj instanceof PriorityQueueStreamRecordFactory;
+ }
+
+ @Override
+ public int hashCode() {
+ return getClass().hashCode();
+ }
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a710fc3d/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
index 1f53d97..56c4161 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
@@ -27,13 +27,18 @@ import org.apache.flink.cep.SubEvent;
import org.apache.flink.cep.nfa.NFA;
import org.apache.flink.cep.nfa.compiler.NFACompiler;
import org.apache.flink.cep.pattern.Pattern;
+import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
+import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.tasks.StreamTaskState;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.util.TestLogger;
+import org.junit.Rule;
import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
import static org.junit.Assert.*;
import java.util.Map;
@@ -41,6 +46,9 @@ import java.util.concurrent.ConcurrentLinkedQueue;
public class CEPOperatorTest extends TestLogger {
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
+
@Test
public void testCEPOperatorWatermarkForwarding() throws Exception {
OneInputStreamOperatorTestHarness<Event, Map<String, Event>> harness = new OneInputStreamOperatorTestHarness<>(
@@ -186,10 +194,112 @@ public class CEPOperatorTest extends TestLogger {
harness.close();
}
+ @Test
+ public void testKeyedCEPOperatorCheckpointing() throws Exception {
+
+ KeySelector<Event, Integer> keySelector = new KeySelector<Event, Integer>() {
+ private static final long serialVersionUID = -4873366487571254798L;
+
+ @Override
+ public Integer getKey(Event value) throws Exception {
+ return value.getId();
+ }
+ };
+
+ OneInputStreamOperatorTestHarness<Event, Map<String, Event>> harness = new OneInputStreamOperatorTestHarness<>(
+ new KeyedCEPPatternOperator<>(
+ Event.createTypeSerializer(),
+ false,
+ keySelector,
+ IntSerializer.INSTANCE,
+ new NFAFactory()));
+
+ harness.configureForKeyedStream(keySelector, BasicTypeInfo.INT_TYPE_INFO);
+
+ harness.open();
+
+ Event startEvent = new Event(42, "start", 1.0);
+ SubEvent middleEvent = new SubEvent(42, "foo", 1.0, 10.0);
+ Event endEvent= new Event(42, "end", 1.0);
+
+ harness.processElement(new StreamRecord<Event>(startEvent, 1));
+ harness.processElement(new StreamRecord<Event>(new Event(42, "foobar", 1.0), 2));
+
+ // simulate snapshot/restore with some elements in internal sorting queue
+ StreamTaskState snapshot = harness.snapshot(0, 0);
+
+ harness = new OneInputStreamOperatorTestHarness<>(
+ new KeyedCEPPatternOperator<>(
+ Event.createTypeSerializer(),
+ false,
+ keySelector,
+ IntSerializer.INSTANCE,
+ new NFAFactory()));
+
+ harness.configureForKeyedStream(keySelector, BasicTypeInfo.INT_TYPE_INFO);
+ harness.setup();
+ harness.restore(snapshot, 1);
+ harness.open();
+
+ harness.processWatermark(new Watermark(Long.MIN_VALUE));
+
+ harness.processElement(new StreamRecord<Event>(new SubEvent(42, "barfoo", 1.0, 5.0), 3));
+
+ // if element timestamps are not correctly checkpointed/restored this will lead to
+ // a pruning time underflow exception in NFA
+ harness.processWatermark(new Watermark(2));
+
+ // simulate snapshot/restore with empty element queue but NFA state
+ StreamTaskState snapshot2 = harness.snapshot(1, 1);
+
+ harness = new OneInputStreamOperatorTestHarness<>(
+ new KeyedCEPPatternOperator<>(
+ Event.createTypeSerializer(),
+ false,
+ keySelector,
+ IntSerializer.INSTANCE,
+ new NFAFactory()));
+
+ harness.configureForKeyedStream(keySelector, BasicTypeInfo.INT_TYPE_INFO);
+ harness.setup();
+ harness.restore(snapshot2, 2);
+ harness.open();
+
+ harness.processElement(new StreamRecord<Event>(middleEvent, 3));
+ harness.processElement(new StreamRecord<Event>(new Event(42, "start", 1.0), 4));
+ harness.processElement(new StreamRecord<Event>(endEvent, 5));
+
+ harness.processWatermark(new Watermark(Long.MAX_VALUE));
+ ConcurrentLinkedQueue<Object> result = harness.getOutput();
+
+ // watermark and the result
+ assertEquals(2, result.size());
+
+ Object resultObject = result.poll();
+ assertTrue(resultObject instanceof StreamRecord);
+ StreamRecord<?> resultRecord = (StreamRecord<?>) resultObject;
+ assertTrue(resultRecord.getValue() instanceof Map);
+
+ @SuppressWarnings("unchecked")
+ Map<String, Event> patternMap = (Map<String, Event>) resultRecord.getValue();
+
+ assertEquals(startEvent, patternMap.get("start"));
+ assertEquals(middleEvent, patternMap.get("middle"));
+ assertEquals(endEvent, patternMap.get("end"));
+
+ harness.close();
+ }
@Test
- public void testKeyedCEPOperatorCheckpointing() throws Exception {
+ public void testKeyedCEPOperatorCheckpointingWithRocksDB() throws Exception {
+
+ String rocksDbPath = tempFolder.newFolder().getAbsolutePath();
+ String rocksDbBackups = tempFolder.newFolder().toURI().toString();
+ RocksDBStateBackend rocksDBStateBackend =
+ new RocksDBStateBackend(rocksDbBackups, new MemoryStateBackend());
+ rocksDBStateBackend.setDbStoragePath(rocksDbPath);
+
KeySelector<Event, Integer> keySelector = new KeySelector<Event, Integer>() {
private static final long serialVersionUID = -4873366487571254798L;
@@ -207,6 +317,7 @@ public class CEPOperatorTest extends TestLogger {
IntSerializer.INSTANCE,
new NFAFactory()));
+ harness.setStateBackend(rocksDBStateBackend);
harness.configureForKeyedStream(keySelector, BasicTypeInfo.INT_TYPE_INFO);
harness.open();
@@ -229,6 +340,10 @@ public class CEPOperatorTest extends TestLogger {
IntSerializer.INSTANCE,
new NFAFactory()));
+ rocksDBStateBackend =
+ new RocksDBStateBackend(rocksDbBackups, new MemoryStateBackend());
+ rocksDBStateBackend.setDbStoragePath(rocksDbPath);
+ harness.setStateBackend(rocksDBStateBackend);
harness.configureForKeyedStream(keySelector, BasicTypeInfo.INT_TYPE_INFO);
harness.setup();
harness.restore(snapshot, 1);
@@ -253,6 +368,10 @@ public class CEPOperatorTest extends TestLogger {
IntSerializer.INSTANCE,
new NFAFactory()));
+ rocksDBStateBackend =
+ new RocksDBStateBackend(rocksDbBackups, new MemoryStateBackend());
+ rocksDBStateBackend.setDbStoragePath(rocksDbPath);
+ harness.setStateBackend(rocksDBStateBackend);
harness.configureForKeyedStream(keySelector, BasicTypeInfo.INT_TYPE_INFO);
harness.setup();
harness.restore(snapshot2, 2);