@Test
@SuppressWarnings("unchecked") // is manually checked
public void create() throws Exception {

    // implement this
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    List<SubStream<GenericEvent>> subStreams = new ArrayList<>();
    // create sub streams
    List<EvaluationFunction<GenericEvent>> evaluationFunctions = new ArrayList<>();
    List<RecordHandler> recordHandlers = new ArrayList<>();
    final AtomicInteger i = new AtomicInteger(0);
    // create evaluation functions
    evaluationFunctions.add(
            new LambdaEvaluationFunction(
                    ctx -> ((EvaluationContext<GenericEvent>)ctx).collect(
                            new GenericEvent(
                                    System.currentTimeMillis(),
                                    "test" + i.addAndGet(1),
                                    "testSource"
                            )
                    ),
                    () -> new HashSet<>(Collections.singletonList("test"))
            )
    );
    // create 3 no op record handlers
    recordHandlers.add(new NoopRecordHandler());
    recordHandlers.add(new NoopRecordHandler());
    recordHandlers.add(new NoopRecordHandler());
    subStreams.add(
            SubStream.<GenericEvent>builder()
                    .withIdentifier("testSubStream1")
                    .withSortWindow(100L)
                    .withPredicate(values -> true)
                    .withRecordHandlers(recordHandlers)
                    .withEvaluationFunctions(evaluationFunctions)
                    .build()
    );
    subStreams.add(
            SubStream.<GenericEvent>builder()
                    .withIdentifier("testSubStream2")
                    .withSortWindow(100L)
                    .withPredicate(values -> true)
                    .withRecordHandlers(recordHandlers)
                    .withEvaluationFunctions(evaluationFunctions)
                    .build()
    );
    // Create a Pipeline
    EvaluationPipeline<GenericEvent> pipeline = EvaluationPipeline.<GenericEvent>builder()
            .withIdentifier("testPipe")
            .withSubStreams(subStreams)
            .build();
    // Convert to Flink Pipeline
    DataStream<MRecord> in = createDataStream(env);
    in.print();
    DataStream<GenericEvent> outEventDataStream = new CrunchFlinkPipelineFactory<GenericEvent>()
            .create(in, pipeline, GenericEvent.class);
    GenericEventCollectSink collectSink = new GenericEventCollectSink();
    outEventDataStream.addSink(collectSink);
    // Has a side effect (CollectSink)
    env.execute("testJob");
    Assert.assertEquals(200, GenericEventCollectSink.values.size());
}