public static IntrinsicMapTaskExecutor withSharedCounterSet(
      List<Operation> operations,
      CounterSet counters,
      ExecutionStateTracker executionStateTracker) {
    return new IntrinsicMapTaskExecutor(operations, counters, executionStateTracker);
  }
@Test
  public void testNoReadOperation() throws Exception {
    // Test MapTaskExecutor without ReadOperation.
    List<Operation> operations =
        Arrays.<Operation>asList(createOperation("o1", 1), createOperation("o2", 2));
    ExecutionStateTracker stateTracker = ExecutionStateTracker.newForTest();
    try (IntrinsicMapTaskExecutor executor =
        IntrinsicMapTaskExecutor.withSharedCounterSet(operations, counterSet, stateTracker)) {
      thrown.expect(IllegalStateException.class);
      thrown.expectMessage("is not a ReadOperation");
      executor.getReadOperation();
    }
  }