@Test
public void runWithTwoSubstreams() {

    // Create source
    UntypedValues values = UntypedValues.builder()
            .source("test")
            .prefix("")
            .timestamp(123L)
            .values(Collections.singletonMap("test", "test"))
            .build();
  MRecordSource source = MRecordSources.of(VALUES_FACTORY.apply(123L),
      VALUES_FACTORY.apply(124L), VALUES_FACTORY.apply(125L));
    // Create Pipeline
    EvaluationPipeline pipeline = createPipelineWithTwoSubstreams();
    // Create Sink
    EventSink sink = Mockito.mock(EventSink.class);
    // Create the Executor
    CrunchExecutor crunchExecutor = new CrunchExecutor(source, pipeline, sink);
    // Run the executor
    crunchExecutor.run();
    // Ensure that two events have been reported
    Mockito.verify(sink, times(4)).handle(any());
}