@SuppressWarnings({"unchecked", "rawtypes"})
  @Override
  public @Nullable <InputT> TransformEvaluator<InputT> forApplication(
      AppliedPTransform<?, ?, ?> application, CommittedBundle<?> inputBundle) {
    return createEvaluator((AppliedPTransform) application);
  }
@Test
  public void noElementsAvailableReaderIncludedInResidual() throws Exception {
    // Read with a very slow rate so by the second read there are no more elements
    PCollection<Long> pcollection =
        p.apply(Read.from(new TestUnboundedSource<>(VarLongCoder.of(), 1L)));
    SplittableParDo.convertReadBasedSplittableDoFnsToPrimitiveReads(p);
    AppliedPTransform<?, ?, ?> sourceTransform = getProducer(pcollection);

    when(context.createRootBundle()).thenReturn(bundleFactory.createRootBundle());
    Collection<CommittedBundle<?>> initialInputs =
        new UnboundedReadEvaluatorFactory.InputProvider(context, p.getOptions())
            .getInitialInputs(sourceTransform, 1);

    // Process the initial shard. This might produce some output, and will produce a residual shard
    // which should produce no output when read from within the following day.
    when(context.createBundle(pcollection)).thenReturn(bundleFactory.createBundle(pcollection));
    CommittedBundle<?> inputBundle = Iterables.getOnlyElement(initialInputs);
    TransformEvaluator<UnboundedSourceShard<Long, TestCheckpointMark>> evaluator =
        factory.forApplication(sourceTransform, inputBundle);
    for (WindowedValue<?> value : inputBundle.getElements()) {
      evaluator.processElement(
          (WindowedValue<UnboundedSourceShard<Long, TestCheckpointMark>>) value);
    }
    TransformResult<UnboundedSourceShard<Long, TestCheckpointMark>> result =
        evaluator.finishBundle();

    // Read from the residual of the first read. This should not produce any output, but should
    // include a residual shard in the result.
    UncommittedBundle<Long> secondOutput = bundleFactory.createBundle(longs);
    when(context.createBundle(longs)).thenReturn(secondOutput);
    TransformEvaluator<UnboundedSourceShard<Long, TestCheckpointMark>> secondEvaluator =
        factory.forApplication(sourceTransform, inputBundle);
    WindowedValue<UnboundedSourceShard<Long, TestCheckpointMark>> residual =
        (WindowedValue<UnboundedSourceShard<Long, TestCheckpointMark>>)
            Iterables.getOnlyElement(result.getUnprocessedElements());
    secondEvaluator.processElement(residual);

    TransformResult<UnboundedSourceShard<Long, TestCheckpointMark>> secondResult =
        secondEvaluator.finishBundle();

    // Sanity check that nothing was output (The test would have to run for more than a day to do
    // so correctly.)
    assertThat(secondOutput.commit(Instant.now()).getElements(), Matchers.emptyIterable());

    // Test that even though the reader produced no outputs, there is still a residual shard with
    // the updated watermark.
    WindowedValue<UnboundedSourceShard<Long, TestCheckpointMark>> unprocessed =
        (WindowedValue<UnboundedSourceShard<Long, TestCheckpointMark>>)
            Iterables.getOnlyElement(secondResult.getUnprocessedElements());
    assertThat(unprocessed.getTimestamp(), Matchers.greaterThan(residual.getTimestamp()));
    assertThat(unprocessed.getValue().getExistingReader(), not(nullValue()));
  }