@Test
public void resultHandlerSendsBeforeFilter() {

  // 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));
  // Handler
  final TestRecordHandler handler = new TestRecordHandler();
  // Create Pipeline
  EvaluationPipeline pipeline = createPipelineWithRecordHandler(handler);
  // Create the Executor
  CrunchExecutor crunchExecutor = new CrunchExecutor(source, pipeline, NoOpSink.INSTANCE);
  // Run the executor
  crunchExecutor.run();
  // Assertions
  assertEquals(3, TestRecordHandler.records.size());
}