@Test
public void testEsperPattern() throws Exception {

    StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
    executionEnvironment.setParallelism(1);
    List<ComplexEvent> expectedValues = Lists.newArrayList();
    ComplexEvent complexEvent = new ComplexEvent(Event.start(), Event.end());
    expectedValues.add(complexEvent);
    List<Event> events = Arrays.asList(complexEvent.getStartEvent(), complexEvent.getEndEvent());
    DataStream<Event> dataStream = executionEnvironment.fromCollection(events);
    EsperStream<Event> esperStream = Esper.pattern(dataStream, "every (A=Event(type='start') -> B=Event(type='end'))");
    DataStream<ComplexEvent> complexEventDataStream = esperStream.select(new EsperSelectFunction<ComplexEvent>() {
        @Override
        public ComplexEvent select(EventBean eventBean) throws Exception {
            return new ComplexEvent((Event) eventBean.get("A"), (Event) eventBean.get("B"));
        }
    });
    complexEventDataStream.addSink(new SinkFunction<ComplexEvent>() {
        @Override
        public void invoke(ComplexEvent value) throws Exception {
            System.err.println(value);
            resultingEvents.add(value);
        }
    });
    executionEnvironment.execute("test-2");
    assertThat(resultingEvents, is(expectedValues));
}