@Test
public void run() {

    // Create source
  Function<Long, UntypedValues> values = VALUES_FACTORY;
  MRecordSource source = MRecordSources.of(values.apply(123L), values.apply(124L));
    // Create Pipeline
    EvaluationPipeline pipeline = createPipeline();
    // 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
  // It shoudl only fire 1 event because timestamp increases 1 time
  Mockito.verify(sink, times(1)).handle(any());
}