ReadOperation extends Operation { public NativeReader.Progress getProgress() { return progress.get(); } protected  ReadOperation(
      NativeReader<?> reader,
      OutputReceiver[] receivers,
      OperationContext context,
      CounterName bytesCounterName); protected  ReadOperation(
      NativeReader<?> reader,
      OutputReceiver[] receivers,
      ScheduledExecutorService scheduler,
      OperationContext context,
      CounterName bytesCounterName); static ReadOperation create(
      NativeReader<?> reader, OutputReceiver[] receivers, OperationContext context); @VisibleForTesting static ReadOperation forTest(
      NativeReader<?> reader, OutputReceiver outputReceiver, OperationContext context); void setProgressUpdatePeriodMs(long millis); NativeReader<?> getReader(); @Override void start(); @Override boolean supportsRestart(); @Override void finish(); @Override void abort(); void abortReadLoop(); NativeReader.Progress getProgress(); NativeReader.@Nullable DynamicSplitResult requestCheckpoint(); NativeReader.@Nullable DynamicSplitResult requestDynamicSplit(
      NativeReader.DynamicSplitRequest splitRequest); @Override String toString(); static final long DEFAULT_PROGRESS_UPDATE_PERIOD_MS; final NativeReader<?> reader; static final long DONT_UPDATE_PERIODICALLY; static final long UPDATE_ON_EACH_ITERATION; }
@Test public void testGetProgress() throws Exception { MockReaderIterator iterator = new MockReaderIterator(0, 5); MockOutputReceiver receiver = new MockOutputReceiver(); ManualScheduler scheduler = new ManualScheduler(); final ReadOperation readOperation = ReadOperation.forTest(new MockReader(iterator), receiver, scheduler.getExecutor(), context); Thread thread = runReadLoopInThread(readOperation); for (int i = 0; i < 5; ++i) { ApproximateReportedProgress progress = readerProgressToCloudProgress(readOperation.getProgress()); Long observedIndex = (progress == null) ? null : progress.getPosition().getRecordIndex().longValue(); assertTrue( "Actual: " + observedIndex + " instead of " + i, (i == 0 && progress == null) || i == observedIndex || i == observedIndex + 1); iterator.offerNext(i); scheduler.runOnce(); receiver.unblockProcess(); } thread.join(); }