From ef827b50a5fa9a928001cb6f6bce08950b191330 Mon Sep 17 00:00:00 2001
From: phil3k <philip.limbeck@gmail.com>
Date: Wed, 25 Oct 2017 20:44:22 +0200
Subject: [PATCH 1/3] moved Esper Engine to value state, started with type
 serialization

---
 .../EsperEngineSerializer.java                | 88 +++++++++++++++++
 .../java/at/datasciencelabs/EsperStream.java  | 32 +++++-
 .../SelectEsperStreamOperator.java            | 71 ++++++++++----
 .../at/datasciencelabs/EsperStreamTest.java   | 97 ++++++++++++-------
 .../java/at/datasciencelabs/TestEvent.java    |  8 ++
 5 files changed, 237 insertions(+), 59 deletions(-)
 create mode 100644 src/main/java/at/datasciencelabs/EsperEngineSerializer.java

diff --git a/src/main/java/at/datasciencelabs/EsperEngineSerializer.java b/src/main/java/at/datasciencelabs/EsperEngineSerializer.java
new file mode 100644
index 0000000..56bfdc9
--- /dev/null
+++ b/src/main/java/at/datasciencelabs/EsperEngineSerializer.java
@@ -0,0 +1,88 @@
+package at.datasciencelabs;
+
+import com.espertech.esper.client.EPServiceProvider;
+import org.apache.flink.api.common.typeutils.CompatibilityResult;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerConfigSnapshot;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+
+import java.io.IOException;
+
+public class EsperEngineSerializer extends TypeSerializer<EPServiceProvider> {
+
+    @Override
+    public boolean isImmutableType() {
+        return false;
+    }
+
+    @Override
+    public TypeSerializer<EPServiceProvider> duplicate() {
+        return this;
+    }
+
+    @Override
+    public EPServiceProvider createInstance() {
+        return null;
+    }
+
+    @Override
+    public EPServiceProvider copy(EPServiceProvider from) {
+        return null;
+    }
+
+    @Override
+    public EPServiceProvider copy(EPServiceProvider from, EPServiceProvider reuse) {
+        return null;
+    }
+
+    @Override
+    public int getLength() {
+        return -1;
+    }
+
+    @Override
+    public void serialize(EPServiceProvider record, DataOutputView target) throws IOException {
+
+    }
+
+    @Override
+    public EPServiceProvider deserialize(DataInputView source) throws IOException {
+        return null;
+    }
+
+    @Override
+    public EPServiceProvider deserialize(EPServiceProvider reuse, DataInputView source) throws IOException {
+        return null;
+    }
+
+    @Override
+    public void copy(DataInputView source, DataOutputView target) throws IOException {
+
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        return false;
+    }
+
+    @Override
+    public boolean canEqual(Object obj) {
+        return false;
+    }
+
+    @Override
+    public int hashCode() {
+        return 0;
+    }
+
+    @Override
+    public TypeSerializerConfigSnapshot snapshotConfiguration() {
+        return null;
+    }
+
+    @Override
+    public CompatibilityResult<EPServiceProvider> ensureCompatibility(TypeSerializerConfigSnapshot configSnapshot) {
+        return null;
+    }
+}
diff --git a/src/main/java/at/datasciencelabs/EsperStream.java b/src/main/java/at/datasciencelabs/EsperStream.java
index dccdefa..140f23c 100644
--- a/src/main/java/at/datasciencelabs/EsperStream.java
+++ b/src/main/java/at/datasciencelabs/EsperStream.java
@@ -1,12 +1,19 @@
 package at.datasciencelabs;
 
+import org.apache.flink.api.common.functions.InvalidTypesException;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.api.java.functions.NullByteKeySelector;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.api.java.typeutils.TypeExtractionException;
+import org.apache.flink.api.java.typeutils.TypeExtractionUtils;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
 import org.apache.flink.streaming.api.TimeCharacteristic;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
 
+import java.lang.reflect.Type;
+
 
 /**
  * A DataStream which is able to detect event sequences and patterns using Esper
@@ -32,17 +39,36 @@ public EsperStream(DataStream<IN> inputStream, String esperQuery) {
      * Select from the EsperStream, must provide the return type of the output DataStream since no type information is
      * currently extracted from the @see {@link EsperSelectFunction}.
      */
-    public <R> SingleOutputStreamOperator<R> select(EsperSelectFunction<R> esperSelectFunction, TypeInformation<R> dataStreamReturnType) {
+    public <R> SingleOutputStreamOperator<R> select(EsperSelectFunction<R> esperSelectFunction) {
         KeySelector<IN, Byte> keySelector = new NullByteKeySelector<>();
 
         SingleOutputStreamOperator<R> patternStream;
 
-        // TODO until the type extractor is capable of extracting non-generic parameters, the return type has to be passed in manually
+        TypeInformation<R> typeInformation = getTypeInformation(esperSelectFunction);
 
         final boolean isProcessingTime = inputStream.getExecutionEnvironment().getStreamTimeCharacteristic() == TimeCharacteristic.ProcessingTime;
-        patternStream = inputStream.keyBy(keySelector).transform("SelectEsperOperator", dataStreamReturnType, new SelectEsperStreamOperator<Byte, IN, R>(inputStream.getType(), esperSelectFunction, isProcessingTime, esperQuery));
+        patternStream = inputStream.keyBy(keySelector).transform("SelectEsperOperator", typeInformation, new SelectEsperStreamOperator<Byte, IN, R>(inputStream.getType(), esperSelectFunction, isProcessingTime, esperQuery));
 
         return patternStream;
     }
 
+    @SuppressWarnings("unchecked")
+    private <OUT> TypeInformation<OUT> getTypeInformation(EsperSelectFunction<OUT> esperSelectFunction) {
+        try {
+            TypeExtractionUtils.LambdaExecutable lambdaExecutable = TypeExtractionUtils.checkAndExtractLambda(esperSelectFunction);
+            if (esperSelectFunction instanceof ResultTypeQueryable) {
+                return ((ResultTypeQueryable<OUT>) esperSelectFunction).getProducedType();
+            }
+            if (lambdaExecutable != null) {
+                Type type = lambdaExecutable.getReturnType();
+                return (TypeInformation<OUT>) TypeExtractor.createTypeInfo(type);
+            }
+            else {
+                return TypeExtractor.createTypeInfo(esperSelectFunction, EsperSelectFunction.class, esperSelectFunction.getClass(), 0);
+            }
+        } catch (TypeExtractionException e) {
+            throw new InvalidTypesException("Could not extract types.", e);
+        }
+    }
+
 }
diff --git a/src/main/java/at/datasciencelabs/SelectEsperStreamOperator.java b/src/main/java/at/datasciencelabs/SelectEsperStreamOperator.java
index 92121dd..6f2d54e 100644
--- a/src/main/java/at/datasciencelabs/SelectEsperStreamOperator.java
+++ b/src/main/java/at/datasciencelabs/SelectEsperStreamOperator.java
@@ -4,6 +4,8 @@
 import com.espertech.esper.client.EPServiceProviderManager;
 import com.espertech.esper.client.EPStatement;
 import com.espertech.esper.client.EventBean;
+import org.apache.flink.api.common.state.ValueState;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.runtime.state.StateInitializationContext;
 import org.apache.flink.runtime.state.VoidNamespace;
@@ -13,27 +15,32 @@
 import org.apache.flink.streaming.api.operators.Triggerable;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 
+import java.io.IOException;
 import java.io.Serializable;
 
 /**
  * An operator which supports detecting event sequences and patterns using Esper.
+ *
  * @param <KEY> Type of the key
- * @param <IN> Type of the input stream
+ * @param <IN>  Type of the input stream
  * @param <OUT> Type of the output stream
  */
 public class SelectEsperStreamOperator<KEY, IN, OUT> extends AbstractUdfStreamOperator<OUT, EsperSelectFunction<OUT>> implements OneInputStreamOperator<IN, OUT>, Triggerable<KEY, VoidNamespace>, Serializable {
 
     private final String query;
     private final TypeInformation<IN> inputType;
-    private EPServiceProvider engine;
+    private ValueState<EPServiceProvider> engineState;
+    private static final String ESPER_SERVICE_PROVIDER_STATE = "esperServiceProviderState";
+    private final Object lock = new Object[0];
 
     /**
      * Constructs a new operator. Requires the type of the input DataStream to register its Event Type at Esper.
      * Currently only processing time evaluation is supported.
-     * @param inputStreamType type of the input DataStream
+     *
+     * @param inputStreamType     type of the input DataStream
      * @param esperSelectFunction function to select from Esper's output
-     * @param isProcessingTime Flag indicating how time is interpreted (processing time vs event time)
-     * @param esperQuery The esper query
+     * @param isProcessingTime    Flag indicating how time is interpreted (processing time vs event time)
+     * @param esperQuery          The esper query
      */
     public SelectEsperStreamOperator(TypeInformation<IN> inputStreamType, EsperSelectFunction<OUT> esperSelectFunction, boolean isProcessingTime, String esperQuery) {
         super(esperSelectFunction);
@@ -48,34 +55,58 @@ public SelectEsperStreamOperator(TypeInformation<IN> inputStreamType, EsperSelec
     @Override
     public void initializeState(StateInitializationContext context) throws Exception {
         super.initializeState(context);
-        engine = EPServiceProviderManager.getDefaultProvider();
-        engine.getEPAdministrator().getConfiguration().addEventType(inputType.getTypeClass());
-        EPStatement statement = engine.getEPAdministrator().createEPL(query);
-        statement.addListener((newData, oldData) -> {
-            for (EventBean event : newData) {
-                EsperSelectFunction<OUT> userFunction = getUserFunction();
-                try {
-                    output.collect(new StreamRecord<>((userFunction.select(event))));
-                } catch (Exception e) {
-                    throw new RuntimeException(e);
-                }
-            }
-        });
+
+        if (this.engineState == null) {
+            this.engineState = getRuntimeContext().getState(new ValueStateDescriptor<>(ESPER_SERVICE_PROVIDER_STATE, new EsperEngineSerializer()));
+        }
     }
 
     @Override
     public void processElement(StreamRecord<IN> streamRecord) throws Exception {
-        engine.getEPRuntime().sendEvent(streamRecord.getValue());
+        EPServiceProvider esperServiceProvider = getServiceProvider(this.hashCode() + "");
+        esperServiceProvider.getEPRuntime().sendEvent(streamRecord.getValue());
+        this.engineState.update(esperServiceProvider);
     }
 
     @Override
     public void onEventTime(InternalTimer<KEY, VoidNamespace> internalTimer) throws Exception {
         internalTimer.getTimestamp();
-
     }
 
     @Override
     public void onProcessingTime(InternalTimer<KEY, VoidNamespace> internalTimer) throws Exception {
 
     }
+
+    private EPServiceProvider getServiceProvider(String context) throws IOException {
+        EPServiceProvider serviceProvider = engineState.value();
+        if (serviceProvider != null) {
+            return serviceProvider;
+        }
+        synchronized (lock) {
+            serviceProvider = engineState.value();
+            if (serviceProvider == null) {
+                serviceProvider = EPServiceProviderManager.getProvider(context);
+                serviceProvider.getEPAdministrator().getConfiguration().addEventType(inputType.getTypeClass());
+                EPStatement statement = serviceProvider.getEPAdministrator().createEPL(query);
+
+                statement.addListener((newData, oldData) -> {
+                    for (EventBean event : newData) {
+                        System.out.println(event.get("name"));
+                        EsperSelectFunction<OUT> userFunction = getUserFunction();
+                        try {
+                            output.collect(new StreamRecord<>((userFunction.select(event))));
+                        } catch (Exception e) {
+                            throw new RuntimeException(e);
+                        }
+                    }
+                });
+                this.engineState.update(serviceProvider);
+                return serviceProvider;
+
+            } else {
+                return engineState.value();
+            }
+        }
+    }
 }
diff --git a/src/test/java/at/datasciencelabs/EsperStreamTest.java b/src/test/java/at/datasciencelabs/EsperStreamTest.java
index 9e88f30..475b3ae 100644
--- a/src/test/java/at/datasciencelabs/EsperStreamTest.java
+++ b/src/test/java/at/datasciencelabs/EsperStreamTest.java
@@ -1,67 +1,92 @@
 package at.datasciencelabs;
 
-import org.apache.flink.api.common.typeinfo.TypeInformation;
+import com.espertech.esper.client.EventBean;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.util.StreamingMultipleProgramsTestBase;
+import org.junit.Before;
 import org.junit.Test;
 
-public class EsperStreamTest {
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
 
-    @Test
-    public void testEsperStream() throws Exception {
-
-        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.IsNull.notNullValue;
 
-        DataStream<TestEvent> dataStream = executionEnvironment.fromElements(new TestEvent("peter", 10), new TestEvent("alex", 25), new TestEvent("maria", 30));
-
-        EsperStream<TestEvent> esperStream = new EsperStream<>(dataStream, "select name, age from TestEvent");
-
-        DataStream<TestEvent> resultStream = esperStream.select((EsperSelectFunction<TestEvent>) collector -> {
-            String name = (String) collector.get("name");
-            int age = (int) collector.get("age");
-            return new TestEvent(name, age);
-        }, TypeInformation.of(TestEvent.class));
+public class EsperStreamTest extends StreamingMultipleProgramsTestBase implements Serializable {
 
-        resultStream.printToErr();
+    private static List<TestEvent> result;
 
-        executionEnvironment.execute("test");
 
+    @Before
+    public void before() {
+        result = new ArrayList<>();
     }
 
     @Test
-    public void testMoreComplexEsperStream() throws Exception {
+    @SuppressWarnings("Convert2Lambda")
+    public void shouldSelectFromStreamUsingAnonymousClassSelect() throws Exception {
         StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
+        executionEnvironment.setParallelism(1);
 
         DataStream<TestEvent> dataStream = executionEnvironment.fromElements(new TestEvent("peter", 10), new TestEvent("alex", 25), new TestEvent("maria", 30));
 
-        EsperStream<TestEvent> eventEsperStream = new EsperStream<>(dataStream, "select avg(age) as average_age from TestEvent");
-
-        DataStream<Double> resultStream = eventEsperStream.select((EsperSelectFunction<Double>) collector -> {
-            Double age = (Double) collector.get("average_age");
-            return age;
-        }, TypeInformation.of(Double.class));
-
-        resultStream.printToErr();
+        EsperStream<TestEvent> esperStream = new EsperStream<>(dataStream, "select name, age from TestEvent");
 
-        executionEnvironment.execute("test");
+        DataStream<TestEvent> resultStream = esperStream.select(new EsperSelectFunction<TestEvent>() {
+            @Override
+            public TestEvent select(EventBean eventBean) throws Exception {
+                String name = (String) eventBean.get("name");
+                int age = (int) eventBean.get("age");
+                return new TestEvent(name, age);
+            }
+        });
+
+        resultStream.addSink(new SinkFunction<TestEvent>() {
+            @Override
+            public void invoke(TestEvent testEvent) throws Exception {
+                System.err.println(testEvent);
+                result.add(testEvent);
+            }
+        });
+
+        executionEnvironment.execute("test-2");
+
+        assertThat(result, is(notNullValue()));
+        assertThat(result.size(), is(3));
     }
 
     @Test
-    public void testStateChangeRule() throws Exception {
+    @SuppressWarnings("Convert2Lambda")
+    public void shouldSelectFromStreamUsingLambdaSelect() throws Exception {
+
         StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
+        executionEnvironment.setParallelism(1);
+
+        DataStream<TestEvent> dataStream = executionEnvironment.fromElements(new TestEvent("peter1", 10), new TestEvent("alex1", 25), new TestEvent("maria1", 30));
 
-        DataStream<TestEvent> dataStream = executionEnvironment.fromElements(new TestEvent("peter", 10), new TestEvent("alex", 25));
+        EsperStream<TestEvent> esperStream = new EsperStream<>(dataStream, "select name, age from TestEvent");
 
-        EsperStream<TestEvent> eventEsperStream = new EsperStream<>(dataStream, "select avg(age) as average_age from TestEvent");
+        DataStream<TestEvent> resultStream = esperStream.select((EsperSelectFunction<TestEvent>) collector -> {
+            String name = (String) collector.get("name");
+            int age = (int) collector.get("age");
+            return new TestEvent(name, age);
+        });
 
-        DataStream<Double> resultStream = eventEsperStream.select((EsperSelectFunction<Double>) collector -> {
-            Double age = (Double) collector.get("average_age");
-            return age;
-        }, TypeInformation.of(Double.class));
+        resultStream.addSink(new SinkFunction<TestEvent>() {
+            @Override
+            public void invoke(TestEvent testEvent) throws Exception {
+                result.add(testEvent);
+            }
+        });
 
-        resultStream.printToErr();
+        executionEnvironment.execute("test-1");
 
-        executionEnvironment.execute("stateChange1");
+        assertThat(result, is(notNullValue()));
+        assertThat(result.size(), is(3));
     }
 
 }
\ No newline at end of file
diff --git a/src/test/java/at/datasciencelabs/TestEvent.java b/src/test/java/at/datasciencelabs/TestEvent.java
index 9838ed8..6bf616a 100644
--- a/src/test/java/at/datasciencelabs/TestEvent.java
+++ b/src/test/java/at/datasciencelabs/TestEvent.java
@@ -37,4 +37,12 @@ public String getName() {
     public int getAge() {
         return age;
     }
+
+    @Override
+    public String toString() {
+        return "TestEvent{" +
+                "name='" + name + '\'' +
+                ", age=" + age +
+                '}';
+    }
 }

From c9a45076797661e8ee55d12d18a9cb039aedd804 Mon Sep 17 00:00:00 2001
From: phil3k <philip.limbeck@gmail.com>
Date: Mon, 13 Nov 2017 19:21:14 +0100
Subject: [PATCH 2/3] restructured maven project, added test for simple String
 datastream, first version of deployable test program

---
 flink-esper-test/flink-esper-test.iml         |  88 ++++++++++++++
 flink-esper-test/pom.xml                      |  61 ++++++++++
 .../datasciencelabs/test/FlinkTestClass.java  |  31 +++++
 flink-esper/pom.xml                           |  63 ++++++++++
 .../main/java/at/datasciencelabs/Esper.java   |  10 ++
 .../EsperEngineSerializer.java                |   0
 .../datasciencelabs/EsperSelectFunction.java  |   0
 .../java/at/datasciencelabs/EsperStream.java  |   0
 .../SelectEsperStreamOperator.java            |   1 -
 .../at/datasciencelabs/EsperStreamTest.java   |  34 ++++++
 .../java/at/datasciencelabs/EsperTest.java    |   0
 .../java/at/datasciencelabs/TestEvent.java    |   0
 pom.xml                                       | 111 ++++++++++--------
 13 files changed, 346 insertions(+), 53 deletions(-)
 create mode 100644 flink-esper-test/flink-esper-test.iml
 create mode 100644 flink-esper-test/pom.xml
 create mode 100644 flink-esper-test/src/main/java/at/datasciencelabs/test/FlinkTestClass.java
 create mode 100644 flink-esper/pom.xml
 create mode 100644 flink-esper/src/main/java/at/datasciencelabs/Esper.java
 rename {src => flink-esper/src}/main/java/at/datasciencelabs/EsperEngineSerializer.java (100%)
 rename {src => flink-esper/src}/main/java/at/datasciencelabs/EsperSelectFunction.java (100%)
 rename {src => flink-esper/src}/main/java/at/datasciencelabs/EsperStream.java (100%)
 rename {src => flink-esper/src}/main/java/at/datasciencelabs/SelectEsperStreamOperator.java (98%)
 rename {src => flink-esper/src}/test/java/at/datasciencelabs/EsperStreamTest.java (71%)
 rename {src => flink-esper/src}/test/java/at/datasciencelabs/EsperTest.java (100%)
 rename {src => flink-esper/src}/test/java/at/datasciencelabs/TestEvent.java (100%)

diff --git a/flink-esper-test/flink-esper-test.iml b/flink-esper-test/flink-esper-test.iml
new file mode 100644
index 0000000..12ebfeb
--- /dev/null
+++ b/flink-esper-test/flink-esper-test.iml
@@ -0,0 +1,88 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module org.jetbrains.idea.maven.project.MavenProjectsManager.isMavenModule="true" type="JAVA_MODULE" version="4">
+  <component name="NewModuleRootManager" LANGUAGE_LEVEL="JDK_1_8" inherit-compiler-output="false">
+    <output url="file://$MODULE_DIR$/target/classes" />
+    <output-test url="file://$MODULE_DIR$/target/test-classes" />
+    <content url="file://$MODULE_DIR$">
+      <sourceFolder url="file://$MODULE_DIR$/src/main/java" isTestSource="false" />
+      <sourceFolder url="file://$MODULE_DIR$/src/main/resources" type="java-resource" />
+      <sourceFolder url="file://$MODULE_DIR$/src/test/java" isTestSource="true" />
+      <excludeFolder url="file://$MODULE_DIR$/target" />
+    </content>
+    <orderEntry type="inheritedJdk" />
+    <orderEntry type="sourceFolder" forTests="false" />
+    <orderEntry type="module" module-name="flink-esper" />
+    <orderEntry type="library" name="Maven: com.espertech:esper:5.3.0" level="project" />
+    <orderEntry type="library" name="Maven: log4j:log4j:1.2.17" level="project" />
+    <orderEntry type="library" name="Maven: commons-logging:commons-logging:1.1.3" level="project" />
+    <orderEntry type="library" name="Maven: org.antlr:antlr4-runtime:4.1" level="project" />
+    <orderEntry type="library" name="Maven: org.abego.treelayout:org.abego.treelayout.core:1.0.1" level="project" />
+    <orderEntry type="library" name="Maven: cglib:cglib-nodep:3.1" level="project" />
+    <orderEntry type="library" name="Maven: org.apache.flink:flink-shaded-guava:18.0-1.0" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-core:1.3.2" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-annotations:1.3.2" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-metrics-core:1.3.2" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.commons:commons-lang3:3.3.2" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: com.esotericsoftware.kryo:kryo:2.24.0" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: com.esotericsoftware.minlog:minlog:1.2" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.objenesis:objenesis:2.1" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: commons-collections:commons-collections:3.2.2" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.commons:commons-compress:1.4.1" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.tukaani:xz:1.0" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.avro:avro:1.7.7" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.codehaus.jackson:jackson-core-asl:1.9.13" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.codehaus.jackson:jackson-mapper-asl:1.9.13" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: com.thoughtworks.paranamer:paranamer:2.3" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.xerial.snappy:snappy-java:1.0.5" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.slf4j:slf4j-api:1.7.7" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: com.google.code.findbugs:jsr305:1.3.9" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:force-shading:1.3.2" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-streaming-java_2.11:1.3.2" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-runtime_2.11:1.3.2" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-java:1.3.2" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-shaded-hadoop2:1.3.2" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: xmlenc:xmlenc:0.52" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: commons-codec:commons-codec:1.4" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: commons-io:commons-io:2.4" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: commons-net:commons-net:3.1" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: javax.servlet:servlet-api:2.5" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.mortbay.jetty:jetty-util:6.1.26" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: com.sun.jersey:jersey-core:1.9" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: commons-el:commons-el:1.0" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: com.jamesmurty.utils:java-xmlbuilder:0.4" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: commons-lang:commons-lang:2.6" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: commons-configuration:commons-configuration:1.7" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: commons-digester:commons-digester:1.8.1" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: com.jcraft:jsch:0.1.42" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: commons-beanutils:commons-beanutils-bean-collections:1.8.3" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: commons-daemon:commons-daemon:1.0.13" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: javax.xml.bind:jaxb-api:2.2.2" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: javax.xml.stream:stax-api:1.0-2" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: javax.activation:activation:1.1" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: commons-cli:commons-cli:1.3.1" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: io.netty:netty-all:4.0.27.Final" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.javassist:javassist:3.18.2-GA" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.scala-lang:scala-library:2.11.7" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: com.data-artisans:flakka-actor_2.11:2.3-custom" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: com.typesafe:config:1.2.1" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: com.data-artisans:flakka-remote_2.11:2.3-custom" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: io.netty:netty:3.8.0.Final" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: com.google.protobuf:protobuf-java:2.5.0" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.uncommons.maths:uncommons-maths:1.2.2a" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: com.data-artisans:flakka-slf4j_2.11:2.3-custom" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.clapper:grizzled-slf4j_2.11:1.0.2" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: com.github.scopt:scopt_2.11:3.5.0" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: com.fasterxml.jackson.core:jackson-core:2.7.4" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: com.fasterxml.jackson.core:jackson-databind:2.7.4" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: com.fasterxml.jackson.core:jackson-annotations:2.7.0" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.zookeeper:zookeeper:3.4.6" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: jline:jline:0.9.94" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: junit:junit:3.8.1" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: com.twitter:chill_2.11:0.7.4" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: com.twitter:chill-java:0.7.4" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-clients_2.11:1.3.2" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-optimizer_2.11:1.3.2" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.commons:commons-math3:3.5" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.sling:org.apache.sling.commons.json:2.0.6" level="project" />
+  </component>
+</module>
\ No newline at end of file
diff --git a/flink-esper-test/pom.xml b/flink-esper-test/pom.xml
new file mode 100644
index 0000000..7e29f91
--- /dev/null
+++ b/flink-esper-test/pom.xml
@@ -0,0 +1,61 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>flink-esper-parent</artifactId>
+        <groupId>at.datasciencelabs.research</groupId>
+        <version>0.0.1-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>flink-esper-test</artifactId>
+
+    <build>
+        <plugins>
+            <plugin>
+                <artifactId>maven-assembly-plugin</artifactId>
+                <configuration>
+                    <archive>
+                        <manifest>
+                            <mainClass>at.datasciencelabs.test.FlinkTestClass</mainClass>
+                        </manifest>
+                    </archive>
+                    <descriptorRefs>
+                        <descriptorRef>jar-with-dependencies</descriptorRef>
+                    </descriptorRefs>
+                </configuration>
+                <executions>
+                    <execution>
+                        <id>make-assembly</id> <!-- this is used for inheritance merges -->
+                        <phase>package</phase> <!-- bind to the packaging phase -->
+                        <goals>
+                            <goal>single</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+
+
+    <dependencies>
+        <dependency>
+            <groupId>at.datasciencelabs.research</groupId>
+            <artifactId>flink-esper</artifactId>
+            <version>${parent.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-core</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
+</project>
\ No newline at end of file
diff --git a/flink-esper-test/src/main/java/at/datasciencelabs/test/FlinkTestClass.java b/flink-esper-test/src/main/java/at/datasciencelabs/test/FlinkTestClass.java
new file mode 100644
index 0000000..c37f240
--- /dev/null
+++ b/flink-esper-test/src/main/java/at/datasciencelabs/test/FlinkTestClass.java
@@ -0,0 +1,31 @@
+package at.datasciencelabs.test;
+
+import at.datasciencelabs.Esper;
+import at.datasciencelabs.EsperSelectFunction;
+import at.datasciencelabs.EsperStream;
+import com.espertech.esper.client.EventBean;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;
+
+public class FlinkTestClass {
+
+    public static void main(String[] args) throws Exception {
+        StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
+        DataStream<String> dataStream = streamExecutionEnvironment.readTextFile("file:///tmp/flink-test");
+
+        EsperStream<String> esperStream = Esper.pattern(dataStream, "select bytes from String");
+
+        DataStream<String> result = esperStream.select(new EsperSelectFunction<String>() {
+            @Override
+            public String select(EventBean eventBean) throws Exception {
+                return new String((byte[]) eventBean.get("bytes"));
+            }
+        });
+
+        result.addSink(new PrintSinkFunction<>(true));
+
+        streamExecutionEnvironment.execute("Kafka 0.10 Example");
+    }
+
+}
diff --git a/flink-esper/pom.xml b/flink-esper/pom.xml
new file mode 100644
index 0000000..b22d626
--- /dev/null
+++ b/flink-esper/pom.xml
@@ -0,0 +1,63 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>flink-esper-parent</artifactId>
+        <groupId>at.datasciencelabs.research</groupId>
+        <version>0.0.1-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>flink-esper</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.espertech</groupId>
+            <artifactId>esper</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-core</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-shaded-guava</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-test-utils_${scala.binary.version}</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-runtime_${scala.binary.version}</artifactId>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-statebackend-rocksdb_${scala.binary.version}</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>
\ No newline at end of file
diff --git a/flink-esper/src/main/java/at/datasciencelabs/Esper.java b/flink-esper/src/main/java/at/datasciencelabs/Esper.java
new file mode 100644
index 0000000..ac4c8a1
--- /dev/null
+++ b/flink-esper/src/main/java/at/datasciencelabs/Esper.java
@@ -0,0 +1,10 @@
+package at.datasciencelabs;
+
+import org.apache.flink.streaming.api.datastream.DataStream;
+
+public class Esper {
+
+    public static <IN> EsperStream<IN> pattern(DataStream<IN> dataStream, String query) {
+        return new EsperStream<IN>(dataStream, query);
+    }
+}
diff --git a/src/main/java/at/datasciencelabs/EsperEngineSerializer.java b/flink-esper/src/main/java/at/datasciencelabs/EsperEngineSerializer.java
similarity index 100%
rename from src/main/java/at/datasciencelabs/EsperEngineSerializer.java
rename to flink-esper/src/main/java/at/datasciencelabs/EsperEngineSerializer.java
diff --git a/src/main/java/at/datasciencelabs/EsperSelectFunction.java b/flink-esper/src/main/java/at/datasciencelabs/EsperSelectFunction.java
similarity index 100%
rename from src/main/java/at/datasciencelabs/EsperSelectFunction.java
rename to flink-esper/src/main/java/at/datasciencelabs/EsperSelectFunction.java
diff --git a/src/main/java/at/datasciencelabs/EsperStream.java b/flink-esper/src/main/java/at/datasciencelabs/EsperStream.java
similarity index 100%
rename from src/main/java/at/datasciencelabs/EsperStream.java
rename to flink-esper/src/main/java/at/datasciencelabs/EsperStream.java
diff --git a/src/main/java/at/datasciencelabs/SelectEsperStreamOperator.java b/flink-esper/src/main/java/at/datasciencelabs/SelectEsperStreamOperator.java
similarity index 98%
rename from src/main/java/at/datasciencelabs/SelectEsperStreamOperator.java
rename to flink-esper/src/main/java/at/datasciencelabs/SelectEsperStreamOperator.java
index 6f2d54e..bf6a54e 100644
--- a/src/main/java/at/datasciencelabs/SelectEsperStreamOperator.java
+++ b/flink-esper/src/main/java/at/datasciencelabs/SelectEsperStreamOperator.java
@@ -92,7 +92,6 @@ private EPServiceProvider getServiceProvider(String context) throws IOException
 
                 statement.addListener((newData, oldData) -> {
                     for (EventBean event : newData) {
-                        System.out.println(event.get("name"));
                         EsperSelectFunction<OUT> userFunction = getUserFunction();
                         try {
                             output.collect(new StreamRecord<>((userFunction.select(event))));
diff --git a/src/test/java/at/datasciencelabs/EsperStreamTest.java b/flink-esper/src/test/java/at/datasciencelabs/EsperStreamTest.java
similarity index 71%
rename from src/test/java/at/datasciencelabs/EsperStreamTest.java
rename to flink-esper/src/test/java/at/datasciencelabs/EsperStreamTest.java
index 475b3ae..2f5c5c4 100644
--- a/src/test/java/at/datasciencelabs/EsperStreamTest.java
+++ b/flink-esper/src/test/java/at/datasciencelabs/EsperStreamTest.java
@@ -10,6 +10,7 @@
 
 import java.io.Serializable;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 
 import static org.hamcrest.CoreMatchers.is;
@@ -19,11 +20,13 @@
 public class EsperStreamTest extends StreamingMultipleProgramsTestBase implements Serializable {
 
     private static List<TestEvent> result;
+    private static List<String> stringResult;
 
 
     @Before
     public void before() {
         result = new ArrayList<>();
+        stringResult = new ArrayList<>();
     }
 
     @Test
@@ -89,4 +92,35 @@ public void invoke(TestEvent testEvent) throws Exception {
         assertThat(result.size(), is(3));
     }
 
+    @Test
+    @SuppressWarnings("Convert2Lambda")
+    public void shouldSelectFromStringDataStream() throws Exception {
+        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
+        executionEnvironment.setParallelism(1);
+
+        List<String> expectedValues = Arrays.asList("first", "second");
+        DataStream<String> dataStream = executionEnvironment.fromCollection(expectedValues);
+
+        EsperStream<String> esperStream = Esper.pattern(dataStream, "select bytes from String");
+
+        DataStream<String> resultStream = esperStream.select((EsperSelectFunction<String>) collector -> {
+            byte[] bytes = (byte[]) collector.get("bytes");
+            return new String(bytes);
+        });
+
+        resultStream.addSink(new SinkFunction<String>() {
+            @Override
+            public void invoke(String testEvent) throws Exception {
+                System.err.println(testEvent);
+                stringResult.add(testEvent);
+            }
+        });
+
+        executionEnvironment.execute("test-2");
+
+        assertThat(stringResult, is(notNullValue()));
+        assertThat(stringResult.size(), is(2));
+        assertThat(stringResult, is(expectedValues));
+    }
+
 }
\ No newline at end of file
diff --git a/src/test/java/at/datasciencelabs/EsperTest.java b/flink-esper/src/test/java/at/datasciencelabs/EsperTest.java
similarity index 100%
rename from src/test/java/at/datasciencelabs/EsperTest.java
rename to flink-esper/src/test/java/at/datasciencelabs/EsperTest.java
diff --git a/src/test/java/at/datasciencelabs/TestEvent.java b/flink-esper/src/test/java/at/datasciencelabs/TestEvent.java
similarity index 100%
rename from src/test/java/at/datasciencelabs/TestEvent.java
rename to flink-esper/src/test/java/at/datasciencelabs/TestEvent.java
diff --git a/pom.xml b/pom.xml
index 1c6c2ec..42ae760 100644
--- a/pom.xml
+++ b/pom.xml
@@ -5,8 +5,13 @@
     <modelVersion>4.0.0</modelVersion>
 
     <groupId>at.datasciencelabs.research</groupId>
-    <artifactId>flink-esper</artifactId>
+    <artifactId>flink-esper-parent</artifactId>
+    <packaging>pom</packaging>
     <version>0.0.1-SNAPSHOT</version>
+    <modules>
+        <module>flink-esper</module>
+        <module>flink-esper-test</module>
+    </modules>
 
     <properties>
         <scala.binary.version>2.11</scala.binary.version>
@@ -15,65 +20,67 @@
         <maven.compiler.target>1.8</maven.compiler.target>
     </properties>
 
-    <dependencies>
-        <dependency>
-            <groupId>com.espertech</groupId>
-            <artifactId>esper</artifactId>
-            <version>5.3.0</version>
-        </dependency>
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>com.espertech</groupId>
+                <artifactId>esper</artifactId>
+                <version>5.3.0</version>
+            </dependency>
 
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-core</artifactId>
-            <version>${flink.version}</version>
-            <scope>provided</scope>
-        </dependency>
+            <dependency>
+                <groupId>org.apache.flink</groupId>
+                <artifactId>flink-core</artifactId>
+                <version>${flink.version}</version>
+                <scope>provided</scope>
+            </dependency>
 
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
-            <version>${flink.version}</version>
-            <scope>provided</scope>
-        </dependency>
+            <dependency>
+                <groupId>org.apache.flink</groupId>
+                <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+                <version>${flink.version}</version>
+                <scope>provided</scope>
+            </dependency>
 
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-shaded-guava</artifactId>
-            <version>18.0-1.0</version>
-        </dependency>
+            <dependency>
+                <groupId>org.apache.flink</groupId>
+                <artifactId>flink-shaded-guava</artifactId>
+                <version>18.0-1.0</version>
+            </dependency>
 
-        <!-- test dependencies -->
+            <!-- test dependencies -->
 
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-test-utils_${scala.binary.version}</artifactId>
-            <version>${flink.version}</version>
-            <scope>test</scope>
-        </dependency>
+            <dependency>
+                <groupId>org.apache.flink</groupId>
+                <artifactId>flink-test-utils_${scala.binary.version}</artifactId>
+                <version>${flink.version}</version>
+                <scope>test</scope>
+            </dependency>
 
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
-            <version>${flink.version}</version>
-            <type>test-jar</type>
-            <scope>test</scope>
-        </dependency>
+            <dependency>
+                <groupId>org.apache.flink</groupId>
+                <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+                <version>${flink.version}</version>
+                <type>test-jar</type>
+                <scope>test</scope>
+            </dependency>
 
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-runtime_${scala.binary.version}</artifactId>
-            <version>${flink.version}</version>
-            <type>test-jar</type>
-            <scope>test</scope>
-        </dependency>
+            <dependency>
+                <groupId>org.apache.flink</groupId>
+                <artifactId>flink-runtime_${scala.binary.version}</artifactId>
+                <version>${flink.version}</version>
+                <type>test-jar</type>
+                <scope>test</scope>
+            </dependency>
 
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-statebackend-rocksdb_${scala.binary.version}</artifactId>
-            <version>${flink.version}</version>
-            <scope>test</scope>
-        </dependency>
+            <dependency>
+                <groupId>org.apache.flink</groupId>
+                <artifactId>flink-statebackend-rocksdb_${scala.binary.version}</artifactId>
+                <version>${flink.version}</version>
+                <scope>test</scope>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
 
-    </dependencies>
 
 </project>
\ No newline at end of file

From 2d6214b194b5a5ab1d974f181a891090b1530d26 Mon Sep 17 00:00:00 2001
From: phil3k <philip.limbeck@gmail.com>
Date: Mon, 13 Nov 2017 19:23:31 +0100
Subject: [PATCH 3/3] removed project file

---
 flink-esper.iml | 96 -------------------------------------------------
 1 file changed, 96 deletions(-)
 delete mode 100644 flink-esper.iml

diff --git a/flink-esper.iml b/flink-esper.iml
deleted file mode 100644
index 1d93ba2..0000000
--- a/flink-esper.iml
+++ /dev/null
@@ -1,96 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<module org.jetbrains.idea.maven.project.MavenProjectsManager.isMavenModule="true" type="JAVA_MODULE" version="4">
-  <component name="NewModuleRootManager" LANGUAGE_LEVEL="JDK_1_8" inherit-compiler-output="false">
-    <output url="file://$MODULE_DIR$/target/classes" />
-    <output-test url="file://$MODULE_DIR$/target/test-classes" />
-    <content url="file://$MODULE_DIR$">
-      <sourceFolder url="file://$MODULE_DIR$/src/main/java" isTestSource="false" />
-      <sourceFolder url="file://$MODULE_DIR$/src/main/resources" type="java-resource" />
-      <sourceFolder url="file://$MODULE_DIR$/src/test/java" isTestSource="true" />
-      <excludeFolder url="file://$MODULE_DIR$/target" />
-    </content>
-    <orderEntry type="inheritedJdk" />
-    <orderEntry type="sourceFolder" forTests="false" />
-    <orderEntry type="library" name="Maven: com.espertech:esper:5.3.0" level="project" />
-    <orderEntry type="library" name="Maven: log4j:log4j:1.2.17" level="project" />
-    <orderEntry type="library" name="Maven: commons-logging:commons-logging:1.1.3" level="project" />
-    <orderEntry type="library" name="Maven: org.antlr:antlr4-runtime:4.1" level="project" />
-    <orderEntry type="library" name="Maven: org.abego.treelayout:org.abego.treelayout.core:1.0.1" level="project" />
-    <orderEntry type="library" name="Maven: cglib:cglib-nodep:3.1" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-core:1.3.2" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-annotations:1.3.2" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-metrics-core:1.3.2" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.commons:commons-lang3:3.3.2" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: com.esotericsoftware.kryo:kryo:2.24.0" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: com.esotericsoftware.minlog:minlog:1.2" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.objenesis:objenesis:2.1" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: commons-collections:commons-collections:3.2.2" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.commons:commons-compress:1.4.1" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.tukaani:xz:1.0" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.avro:avro:1.7.7" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.codehaus.jackson:jackson-core-asl:1.9.13" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.codehaus.jackson:jackson-mapper-asl:1.9.13" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: com.thoughtworks.paranamer:paranamer:2.3" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.xerial.snappy:snappy-java:1.0.5" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.slf4j:slf4j-api:1.7.7" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: com.google.code.findbugs:jsr305:1.3.9" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:force-shading:1.3.2" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-streaming-java_2.11:1.3.2" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-runtime_2.11:1.3.2" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-clients_2.11:1.3.2" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-optimizer_2.11:1.3.2" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.commons:commons-math3:3.5" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.sling:org.apache.sling.commons.json:2.0.6" level="project" />
-    <orderEntry type="library" name="Maven: org.apache.flink:flink-shaded-guava:18.0-1.0" level="project" />
-    <orderEntry type="library" scope="TEST" name="Maven: org.apache.flink:flink-test-utils_2.11:1.3.2" level="project" />
-    <orderEntry type="library" scope="TEST" name="Maven: org.apache.flink:flink-test-utils-junit:1.3.2" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: junit:junit:4.12" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.hamcrest:hamcrest-core:1.3" level="project" />
-    <orderEntry type="library" scope="TEST" name="Maven: org.apache.curator:curator-test:2.12.0" level="project" />
-    <orderEntry type="library" scope="TEST" name="Maven: com.google.guava:guava:16.0.1" level="project" />
-    <orderEntry type="library" scope="TEST" name="Maven: org.apache.flink:flink-streaming-java_2.11:test-jar:tests:1.3.2" level="project" />
-    <orderEntry type="library" scope="TEST" name="Maven: org.apache.flink:flink-runtime_2.11:test-jar:tests:1.3.2" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-java:1.3.2" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-shaded-hadoop2:1.3.2" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: xmlenc:xmlenc:0.52" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: commons-codec:commons-codec:1.4" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: commons-io:commons-io:2.4" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: commons-net:commons-net:3.1" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: javax.servlet:servlet-api:2.5" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.mortbay.jetty:jetty-util:6.1.26" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: com.sun.jersey:jersey-core:1.9" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: commons-el:commons-el:1.0" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: com.jamesmurty.utils:java-xmlbuilder:0.4" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: commons-lang:commons-lang:2.6" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: commons-configuration:commons-configuration:1.7" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: commons-digester:commons-digester:1.8.1" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: com.jcraft:jsch:0.1.42" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: commons-beanutils:commons-beanutils-bean-collections:1.8.3" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: commons-daemon:commons-daemon:1.0.13" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: javax.xml.bind:jaxb-api:2.2.2" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: javax.xml.stream:stax-api:1.0-2" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: javax.activation:activation:1.1" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: commons-cli:commons-cli:1.3.1" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: io.netty:netty-all:4.0.27.Final" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.javassist:javassist:3.18.2-GA" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.scala-lang:scala-library:2.11.7" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: com.data-artisans:flakka-actor_2.11:2.3-custom" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: com.typesafe:config:1.2.1" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: com.data-artisans:flakka-remote_2.11:2.3-custom" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: io.netty:netty:3.8.0.Final" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: com.google.protobuf:protobuf-java:2.5.0" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.uncommons.maths:uncommons-maths:1.2.2a" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: com.data-artisans:flakka-slf4j_2.11:2.3-custom" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.clapper:grizzled-slf4j_2.11:1.0.2" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: com.github.scopt:scopt_2.11:3.5.0" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: com.fasterxml.jackson.core:jackson-core:2.7.4" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: com.fasterxml.jackson.core:jackson-databind:2.7.4" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: com.fasterxml.jackson.core:jackson-annotations:2.7.0" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.zookeeper:zookeeper:3.4.6" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: jline:jline:0.9.94" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: com.twitter:chill_2.11:0.7.4" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: com.twitter:chill-java:0.7.4" level="project" />
-    <orderEntry type="library" scope="TEST" name="Maven: org.apache.flink:flink-statebackend-rocksdb_2.11:1.3.2" level="project" />
-    <orderEntry type="library" scope="TEST" name="Maven: com.data-artisans:frocksdbjni:4.11.2-artisans" level="project" />
-  </component>
-</module>
\ No newline at end of file