@Override
  public <InputT> TransformEvaluator<InputT> forApplication(
      AppliedPTransform<?, ?, ?> application, CommittedBundle<?> inputBundle) {
    @SuppressWarnings({"cast", "unchecked", "rawtypes"})
    TransformEvaluator<InputT> evaluator =
        (TransformEvaluator<InputT>) createInMemoryEvaluator((AppliedPTransform) application);
    return evaluator;
  }
@Test
  public void testFlattenInMemoryEvaluatorWithEmptyPCollectionList() throws Exception {
    PCollectionList<Integer> list = PCollectionList.empty(p);

    PCollection<Integer> flattened = list.apply(Flatten.pCollections());
    flattened.setCoder(VarIntCoder.of());

    EvaluationContext evaluationContext = mock(EvaluationContext.class);
    when(evaluationContext.createBundle(flattened))
        .thenReturn(bundleFactory.createBundle(flattened));

    FlattenEvaluatorFactory factory = new FlattenEvaluatorFactory(evaluationContext);
    AppliedPTransform<?, ?, ?> flattendProducer = DirectGraphs.getProducer(flattened);
    TransformEvaluator<Integer> emptyEvaluator =
        factory.forApplication(
            flattendProducer,
            bundleFactory.createRootBundle().commit(BoundedWindow.TIMESTAMP_MAX_VALUE));

    TransformResult<Integer> leftSideResult = emptyEvaluator.finishBundle();

    CommittedBundle<?> outputBundle =
        Iterables.getOnlyElement(leftSideResult.getOutputBundles()).commit(Instant.now());
    assertThat(outputBundle.getElements(), emptyIterable());
    assertThat(
        leftSideResult.getTransform(),
        Matchers.<AppliedPTransform<?, ?, ?>>equalTo(flattendProducer));
  }