public static IntrinsicMapTaskExecutor withSharedCounterSet(
      List<Operation> operations,
      CounterSet counters,
      ExecutionStateTracker executionStateTracker) {
    return new IntrinsicMapTaskExecutor(operations, counters, executionStateTracker);
  }
@Test
  public void testGetProgressAndRequestSplit() throws Exception {
    TestOutputReceiver receiver =
        new TestOutputReceiver(counterSet, NameContextsForTests.nameContextForTest());
    TestReadOperation operation = new TestReadOperation(receiver, createContext("ReadOperation"));
    ExecutionStateTracker stateTracker = ExecutionStateTracker.newForTest();
    try (IntrinsicMapTaskExecutor executor =
        IntrinsicMapTaskExecutor.withSharedCounterSet(
            Arrays.asList(new Operation[] {operation}), counterSet, stateTracker)) {
      operation.setProgress(approximateProgressAtIndex(1L));
      Assert.assertEquals(positionAtIndex(1L), positionFromProgress(executor.getWorkerProgress()));
      Assert.assertEquals(
          positionAtIndex(1L),
          positionFromSplitResult(executor.requestDynamicSplit(splitRequestAtIndex(1L))));
    }
  }