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

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