@SuppressWarnings({"unchecked", "rawtypes"})
  @Override
  public @Nullable <InputT> TransformEvaluator<InputT> forApplication(
      AppliedPTransform<?, ?, ?> application, CommittedBundle<?> inputBundle) {
    return createEvaluator((AppliedPTransform) application);
  }
@Test
  public void evaluatorReusesReaderAndClosesAtTheEnd() throws Exception {
    int numElements = 1000;
    ContiguousSet<Long> elems =
        ContiguousSet.create(Range.openClosed(0L, (long) numElements), DiscreteDomain.longs());
    TestUnboundedSource<Long> source =
        new TestUnboundedSource<>(BigEndianLongCoder.of(), elems.toArray(new Long[0]));
    source.advanceWatermarkToInfinity = true;

    PCollection<Long> pcollection = p.apply(Read.from(source));
    SplittableParDo.convertReadBasedSplittableDoFnsToPrimitiveReads(p);
    DirectGraph graph = DirectGraphs.getGraph(p);
    AppliedPTransform<?, ?, ?> sourceTransform = graph.getProducer(pcollection);

    when(context.createRootBundle()).thenReturn(bundleFactory.createRootBundle());
    UncommittedBundle<Long> output = mock(UncommittedBundle.class);
    when(context.createBundle(pcollection)).thenReturn(output);

    WindowedValue<UnboundedSourceShard<Long, TestCheckpointMark>> shard =
        WindowedValue.valueInGlobalWindow(
            UnboundedSourceShard.unstarted(source, NeverDeduplicator.create()));
    CommittedBundle<UnboundedSourceShard<Long, TestCheckpointMark>> inputBundle =
        bundleFactory
            .<UnboundedSourceShard<Long, TestCheckpointMark>>createRootBundle()
            .add(shard)
            .commit(Instant.now());
    UnboundedReadEvaluatorFactory factory =
        new UnboundedReadEvaluatorFactory(context, p.getOptions(), 1.0 /* Always reuse */);
    new UnboundedReadEvaluatorFactory.InputProvider(context, p.getOptions())
        .getInitialInputs(sourceTransform, 1);

    CommittedBundle<UnboundedSourceShard<Long, TestCheckpointMark>> residual = inputBundle;

    do {
      TransformEvaluator<UnboundedSourceShard<Long, TestCheckpointMark>> evaluator =
          factory.forApplication(sourceTransform, residual);
      evaluator.processElement(Iterables.getOnlyElement(residual.getElements()));
      TransformResult<UnboundedSourceShard<Long, TestCheckpointMark>> result =
          evaluator.finishBundle();
      residual =
          inputBundle.withElements(
              (Iterable<WindowedValue<UnboundedSourceShard<Long, TestCheckpointMark>>>)
                  result.getUnprocessedElements());
    } while (!Iterables.isEmpty(residual.getElements()));

    verify(output, times(numElements)).add(any());
    assertThat(TestUnboundedSource.readerCreatedCount, equalTo(1));
    assertThat(TestUnboundedSource.readerClosedCount, equalTo(1));
  }