@Override
  public GroupingShuffleReaderIterator<K, V> iterator() throws IOException {
    ApplianceShuffleEntryReader entryReader =
        new ApplianceShuffleEntryReader(
            shuffleReaderConfig, executionContext, operationContext, true);

    initCounter(entryReader.getDatasetId());

    return iterator(entryReader);
  }
@Test
  public void testConsumedParallelism() throws Exception {
    PipelineOptions options = PipelineOptionsFactory.create();
    BatchModeExecutionContext context = BatchModeExecutionContext.forTesting(options, "testStage");
    final int kFirstShard = 0;

    TestShuffleReader shuffleReader = new TestShuffleReader();
    final int kNumRecords = 5;
    for (int i = 0; i < kNumRecords; ++i) {
      byte[] key = CoderUtils.encodeToByteArray(BigEndianIntegerCoder.of(), i);
      ShuffleEntry entry =
          newShuffleEntry(fabricatePosition(kFirstShard, i), key, EMPTY_BYTE_ARRAY, key);
      shuffleReader.addEntry(entry);
    }

    TestOperationContext operationContext = TestOperationContext.create();
    GroupingShuffleReader<Integer, Integer> groupingShuffleReader =
        new GroupingShuffleReader<>(
            options,
            null,
            null,
            null,
            WindowedValue.getFullCoder(
                KvCoder.of(
                    BigEndianIntegerCoder.of(), IterableCoder.of(BigEndianIntegerCoder.of())),
                IntervalWindow.getCoder()),
            context,
            operationContext,
            ShuffleReadCounterFactory.INSTANCE,
            false /* do not sort values */);

    assertFalse(shuffleReader.isClosed());
    try (GroupingShuffleReaderIterator<Integer, Integer> iter =
        groupingShuffleReader.iterator(shuffleReader)) {

      // Iterator hasn't started; consumed parallelism is 0.
      assertEquals(0.0, consumedParallelismFromProgress(iter.getProgress()), 0);

      // The only way to set a stop *position* in tests is via a split. To do that,
      // we must call hasNext() first.

      // Should return entry at key 0.
      assertTrue(iter.start());

      // Iterator just started; consumed parallelism is 0.
      assertEquals(
          0.0,
          readerProgressToCloudProgress(iter.getProgress()).getConsumedParallelism().getValue(),
          0);
      assertNotNull(
          iter.requestDynamicSplit(
              splitRequestAtPosition(
                  makeShufflePosition(
                      fabricatePosition(kFirstShard, 2).immediateSuccessor().getPosition()))));
      // Split does not affect consumed parallelism; consumed parallelism is still 0.
      assertEquals(0.0, consumedParallelismFromProgress(iter.getProgress()), 0);

      // Should return entry at key 1.
      assertTrue(iter.advance());
      assertEquals(1.0, consumedParallelismFromProgress(iter.getProgress()), 0);

      // Should return entry at key 2 (last key, because the stop position
      // is its immediate successor.) Consumed parallelism increments by one to 2.
      assertTrue(iter.advance());
      assertEquals(2.0, consumedParallelismFromProgress(iter.getProgress()), 0);

      // Iterator advanced by one and consumes one more split point (total consumed: 3).
      assertFalse(iter.advance());
      assertEquals(3.0, consumedParallelismFromProgress(iter.getProgress()), 0);
    }
    assertTrue(shuffleReader.isClosed());
  }