@SuppressWarnings({"unchecked", "rawtypes"})
  @Override
  public @Nullable <InputT> TransformEvaluator<InputT> forApplication(
      AppliedPTransform<?, ?, ?> application, CommittedBundle<?> inputBundle) {
    return createEvaluator((AppliedPTransform) application);
  }
@Test
  public void unboundedSourceInMemoryTransformEvaluatorProducesElements() throws Exception {
    when(context.createRootBundle()).thenReturn(bundleFactory.createRootBundle());

    Collection<CommittedBundle<?>> initialInputs =
        new UnboundedReadEvaluatorFactory.InputProvider(context, p.getOptions())
            .getInitialInputs(graph.getProducer(longs), 1);

    CommittedBundle<?> inputShards = Iterables.getOnlyElement(initialInputs);
    UnboundedSourceShard<Long, ?> inputShard =
        (UnboundedSourceShard<Long, ?>)
            Iterables.getOnlyElement(inputShards.getElements()).getValue();
    TransformEvaluator<? super UnboundedSourceShard<Long, ?>> evaluator =
        factory.forApplication(graph.getProducer(longs), inputShards);

    evaluator.processElement((WindowedValue) Iterables.getOnlyElement(inputShards.getElements()));
    TransformResult<? super UnboundedSourceShard<Long, ?>> result = evaluator.finishBundle();

    WindowedValue<? super UnboundedSourceShard<Long, ?>> residual =
        Iterables.getOnlyElement(result.getUnprocessedElements());
    assertThat(residual.getTimestamp(), Matchers.lessThan(DateTime.now().toInstant()));
    UnboundedSourceShard<Long, ?> residualShard =
        (UnboundedSourceShard<Long, ?>) residual.getValue();
    assertThat(residualShard.getSource(), equalTo(inputShard.getSource()));
    assertThat(residualShard.getCheckpoint(), not(nullValue()));
    assertThat(
        output.commit(Instant.now()).getElements(),
        containsInAnyOrder(
            tgw(1L), tgw(2L), tgw(4L), tgw(8L), tgw(9L), tgw(7L), tgw(6L), tgw(5L), tgw(3L),
            tgw(0L)));
  }