public static IntrinsicMapTaskExecutor withSharedCounterSet(
      List<Operation> operations,
      CounterSet counters,
      ExecutionStateTracker executionStateTracker) {
    return new IntrinsicMapTaskExecutor(operations, counters, executionStateTracker);
  }
@Test
  public void testExceptionInFinishAbortsAllOperations() throws Exception {
    Operation o1 = Mockito.mock(Operation.class);
    Operation o2 = Mockito.mock(Operation.class);
    Operation o3 = Mockito.mock(Operation.class);
    Mockito.doThrow(new Exception("in finish")).when(o2).finish();

    ExecutionStateTracker stateTracker = ExecutionStateTracker.newForTest();
    try (IntrinsicMapTaskExecutor executor =
        IntrinsicMapTaskExecutor.withSharedCounterSet(
            Arrays.<Operation>asList(o1, o2, o3), counterSet, stateTracker)) {
      executor.execute();
      fail("Should have thrown");
    } catch (Exception e) {
      InOrder inOrder = Mockito.inOrder(o1, o2, o3);
      inOrder.verify(o3).start();
      inOrder.verify(o2).start();
      inOrder.verify(o1).start();
      inOrder.verify(o1).finish();
      inOrder.verify(o2).finish();

      // Order of abort doesn't matter
      Mockito.verify(o1).abort();
      Mockito.verify(o2).abort();
      Mockito.verify(o3).abort();
      Mockito.verifyNoMoreInteractions(o1, o2, o3);
    }
  }